Added OpenSearch dialect support to acronis-db-bench

Author: iluhinsky

acronis-db-bench/tests.go

@@ -105,7 +105,7 @@ func (t *TestDesc) getDBs() string {
 
 var (
 	// ALL is a list of all supported databases
-	ALL = []db.DialectName{db.POSTGRES, db.MYSQL, db.MSSQL, db.SQLITE, db.CLICKHOUSE, db.CASSANDRA, db.ELASTICSEARCH}
+	ALL = []db.DialectName{db.POSTGRES, db.MYSQL, db.MSSQL, db.SQLITE, db.CLICKHOUSE, db.CASSANDRA, db.ELASTICSEARCH, db.OPENSEARCH}
 	// RELATIONAL is a list of all supported relational databases
 	RELATIONAL = []db.DialectName{db.POSTGRES, db.MYSQL, db.MSSQL, db.SQLITE}
 	// PMWSA is a list of all supported databases except ClickHouse

db/db.go

@@ -19,6 +19,7 @@ const (
 	CLICKHOUSE    DialectName = "clickhouse"    // CLICKHOUSE is the ClickHouse driver name
 	CASSANDRA     DialectName = "cassandra"     // CASSANDRA is the Cassandra driver name
 	ELASTICSEARCH DialectName = "elasticsearch" // ELASTICSEARCH is the Elasticsearch driver name
+	OPENSEARCH    DialectName = "opensearch"    // OPENSEARCH is the OpenSearch driver name
 )
 
 // Special conditions for searching
@@ -359,6 +360,7 @@ func GetDatabases() []DBType {
 	// "A" is used as the latest symbol of the "Cassandra" due to duplicate with ClickHouse "C"
 	ret = append(ret, DBType{Driver: CASSANDRA, Symbol: "A", Name: "Cassandra"})
 	ret = append(ret, DBType{Driver: ELASTICSEARCH, Symbol: "E", Name: "Elasticsearch"})
+	ret = append(ret, DBType{Driver: OPENSEARCH, Symbol: "O", Name: "OpenSearch"})
 
 	return ret
 }

db/es/elasticsearch.go

@@ -33,6 +33,12 @@ func init() {
 	}
 }
 
+type elasticSearchDialect struct{}
+
+func (d *elasticSearchDialect) name() db.DialectName {
+	return db.ELASTICSEARCH
+}
+
 // nolint:gocritic //TODO refactor unnamed returns
 func elasticCredentialsAndConnString(cs string, tlsEnabled bool) (string, string, string, error) {
 	var u, err = url.Parse(cs)
@@ -126,6 +132,7 @@ func (c *esConnector) ConnectionPool(cfg db.Config) (db.Database, error) {
 	return &esDatabase{
 		rw:          rw,
 		mig:         rw,
+		dialect:     &elasticSearchDialect{},
 		queryLogger: cfg.QueryLogger,
 	}, nil
 }

db/es/es.go

@@ -40,8 +40,9 @@ func (s *esSession) Transact(fn func(tx db.DatabaseAccessor) error) error {
 }
 
 type esDatabase struct {
-	rw  accessor
-	mig migrator
+	rw      accessor
+	mig     migrator
+	dialect dialect
 
 	queryLogger db.Logger
 }
@@ -60,7 +61,7 @@ func (d *esDatabase) UseTruncate() bool {
 }
 
 func (d *esDatabase) GetVersion() (db.DialectName, string, error) {
-	return getVersion(d.rw)
+	return getVersion(d.dialect)
 }
 
 func (d *esDatabase) GetInfo(version string) (ret []string, dbInfo *db.Info, err error) {
@@ -193,3 +194,7 @@ func (tq timedQuerier) count(ctx context.Context, idxName indexName, request *Co
 
 	return tq.q.count(ctx, idxName, request)
 }
+
+type dialect interface {
+	name() db.DialectName
+}

db/es/info.go

@@ -3,8 +3,8 @@ package es
 import "github.com/acronis/perfkit/db"
 
 // GetVersion returns DB version and driver name
-func getVersion(q querier) (db.DialectName, string, error) {
-	return db.ELASTICSEARCH, "", nil
+func getVersion(dia dialect) (db.DialectName, string, error) {
+	return dia.name(), "", nil
 }
 
 // GetInfo returns DB info

db/es/opensearch.go

@@ -27,6 +27,12 @@ func init() {
 	}
 }
 
+type openSearchDialect struct{}
+
+func (d *openSearchDialect) name() db.DialectName {
+	return db.OPENSEARCH
+}
+
 type openSearchConnector struct{}
 
 func (c *openSearchConnector) ConnectionPool(cfg db.Config) (db.Database, error) {
@@ -97,12 +103,13 @@ func (c *openSearchConnector) ConnectionPool(cfg db.Config) (db.Database, error)
 	return &esDatabase{
 		rw:          rw,
 		mig:         mig,
+		dialect:     &openSearchDialect{},
 		queryLogger: cfg.QueryLogger,
 	}, nil
 }
 
 func (c *openSearchConnector) DialectName(scheme string) (db.DialectName, error) {
-	return db.ELASTICSEARCH, nil
+	return db.OPENSEARCH, nil
 }
 
 type openSearchQuerier struct {