Merge pull request #6 from iluhinsky/feature/es

Added ElasticSearch support

Author: viatoriche

acronis-db-bench/events.go

@@ -327,13 +327,24 @@ func (e *EventBus) DoAlign() (bool, error) {
 		 * step #1 - get fresh events
 		 */
 
-		var rows, err = tx.Search("acronis_db_bench_eventbus_events",
-			"internal_id, topic_internal_id, event_type_internal_id, event_id, source, sequence, tenant_id, client_id, trace_parent, "+
-				"subject_id, data_ref, data, data_base64, created_at",
-			"",
-			"internal_id",
-			e.batchSize,
-			false)
+		var rows, err = tx.Query(fmt.Sprintf(`
+			SELECT internal_id,
+				   topic_internal_id,
+				   event_type_internal_id,
+				   event_id,
+				   source,
+				   sequence,
+				   tenant_id,
+				   client_id,
+				   trace_parent,
+				   subject_id,
+				   data_ref,
+				   data,
+				   data_base64,
+				   created_at
+			FROM acronis_db_bench_eventbus_events
+			ORDER BY internal_id
+			LIMIT %d;`, e.batchSize))
 		if err != nil {
 			return err
 		}
@@ -501,13 +512,16 @@ func (e *EventBus) DoFetch() (bool, error) {
 			return false, err
 		}
 
-		var rows, err = sess.Search("acronis_db_bench_eventbus_stream s INNER JOIN acronis_db_bench_eventbus_data d ON s.int_id = d.int_id",
-			"s.int_id, s.topic_id, d.type_id, s.seq, s.seq_time, d.data",
-			"s.topic_id = $1 AND s.seq IS NOT NULL AND s.seq > $2",
-			"s.seq",
-			e.batchSize,
-			false,
-			t, cur64)
+		var rows, err = sess.Query(fmt.Sprintf(`
+			SELECT s.int_id, s.topic_id, d.type_id, s.seq, s.seq_time, d.data
+			FROM acronis_db_bench_eventbus_stream s
+					 INNER JOIN acronis_db_bench_eventbus_data d ON s.int_id = d.int_id
+			WHERE s.topic_id = %s
+			  AND s.seq IS NOT NULL
+			  AND s.seq > %d
+			ORDER BY s.seq
+			LIMIT %d;`,
+			t, cur64, e.batchSize))
 		if err != nil {
 			return false, err
 		}

acronis-db-bench/tables.go

@@ -454,7 +454,7 @@ var TestTableTimeSeriesSQL = TestTable{
 	CreateQueryPatchFuncs: []CreateQueryPatchFunc{
 		func(table string, query string, dialect db.DialectName) (string, error) { //nolint:revive
 			if dialect == db.CASSANDRA {
-				query = strings.ReplaceAll(query, "{$bigint_autoinc_pk}", "{$bigint_autoinc}")
+				query = strings.ReplaceAll(query, string(db.DataTypeBigIntAutoIncPK), string(db.DataTypeBigIntAutoInc))
 				query = strings.ReplaceAll(query, "value int {$notnull}", `value int,
 						PRIMARY KEY ((tenant_id, device_id, metric_id), id, ts)
 					`)

acronis-db-bench/tenant_generator.go

@@ -453,7 +453,7 @@ func (tc *TenantsCache) PopulateUuidsFromDB(database db.Database) {
 
 	var session = database.Session(database.Context(context.Background()))
 
-	var rows, err = session.Search(TableNameTenants, "uuid, id, kind, nesting_level", "", "", 0, false)
+	var rows, err = session.Query(fmt.Sprintf("SELECT uuid, id, kind, nesting_level FROM %s", TableNameTenants))
 	if err != nil {
 		return
 	}
@@ -477,7 +477,7 @@ func (tc *TenantsCache) PopulateUuidsFromDB(database db.Database) {
 	}
 
 	var ctiRows db.Rows
-	if ctiRows, err = session.Search(TableNameCtiEntities, "uuid", "", "", 0, false); err != nil {
+	if ctiRows, err = session.Query(fmt.Sprintf("SELECT uuid FROM %s;", TableNameCtiEntities)); err != nil {
 		return
 	}
 
@@ -520,7 +520,7 @@ func (tc *TenantsCache) CreateTenant(rw *benchmark.RandomizerWorker, tx db.Datab
 	newTenantClosure := TenantClosureObj{ParentID: t.ID, ChildID: t.ID, ParentKind: t.Kind, Barrier: 0}
 	tcToCreate = append(tcToCreate, newTenantClosure)
 
-	rows, err := tx.Search(TableNameTenantClosure, "parent_id, parent_kind, barrier", fmt.Sprintf("child_id = %d", t.ParentID), "", 0, false)
+	rows, err := tx.Query(fmt.Sprintf("SELECT parent_id, parent_kind, barrier FROM %s WHERE child_id = %d", TableNameTenantClosure, t.ParentID))
 	if err != nil {
 		tc.logger.Log(benchmark.LogTrace, 0, fmt.Sprintf("error selecting from table %s: %v", TableNameTenantClosure, err))
 		return "", err
@@ -599,7 +599,7 @@ func (tc *TenantsCache) createRandomCtiEntity(rw *benchmark.RandomizerWorker) (*
 func getMax(database db.Database, field string) int {
 	var session = database.Session(database.Context(context.Background()))
 
-	var maxRows, err = session.Search(TableNameTenants, fmt.Sprintf("COALESCE(MAX(%s),0)", field), "", "", 0, false)
+	var maxRows, err = session.Query(fmt.Sprintf("SELECT COALESCE(MAX(%s),0) FROM %s;", field, TableNameTenants))
 	if err != nil {
 		return 0
 	}

acronis-db-bench/workers.go

@@ -176,7 +176,7 @@ func testSelect(
 		}
 
 		var session = c.database.Session(c.database.Context(context.Background()))
-		var rows, err = session.Search(from, what, where, orderBy, batch, explain)
+		var rows, err = session.SearchRaw(from, what, where, orderBy, batch, explain)
 		if err != nil {
 			b.Exit(err)
 		}

db/consts.go

@@ -4,15 +4,7 @@ import "strings"
 
 type DialectName string
 
-const (
-	SQLITE     DialectName = "sqlite"     // SQLITE is the SQLite driver name
-	SQLITE3    DialectName = "sqlite3"    // SQLITE3 is the SQLite driver name
-	POSTGRES   DialectName = "postgres"   // POSTGRES is the PostgreSQL driver name
-	MYSQL      DialectName = "mysql"      // MYSQL is the MySQL driver name
-	MSSQL      DialectName = "mssql"      // MSSQL is the Microsoft SQL Server driver name
-	CLICKHOUSE DialectName = "clickhouse" // CLICKHOUSE is the ClickHouse driver name
-	CASSANDRA  DialectName = "cassandra"  // CASSANDRA is the Cassandra driver name
-)
+const ()
 
 var (
 	// SupportedDrivers is a string containing all supported drivers

db/db.go

@@ -7,6 +7,24 @@ import (
 	"time"
 )
 
+// Supported dialect
+const (
+	SQLITE        DialectName = "sqlite"        // SQLITE is the SQLite driver name
+	SQLITE3       DialectName = "sqlite3"       // SQLITE3 is the SQLite driver name
+	POSTGRES      DialectName = "postgres"      // POSTGRES is the PostgreSQL driver name
+	MYSQL         DialectName = "mysql"         // MYSQL is the MySQL driver name
+	MSSQL         DialectName = "mssql"         // MSSQL is the Microsoft SQL Server driver name
+	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
+)
+
+// Special conditions for searching
+const (
+	SpecialConditionIsNull    = "isnull()"
+	SpecialConditionIsNotNull = "notnull()"
+)
+
 // Connector is an interface for registering database connectors without knowing the specific connector implementations
 type Connector interface {
 	ConnectionPool(cfg Config) (Database, error)
@@ -41,6 +59,9 @@ type Config struct {
 	DryRun          bool
 	UseTruncate     bool
 
+	TLSEnabled bool
+	TLSCACert  []byte
+
 	QueryLogger      Logger
 	ReadedRowsLogger Logger
 	QueryTimeLogger  Logger
@@ -93,10 +114,38 @@ type databaseQueryRegistrator interface {
 	StatementExit(statement string, startTime time.Time, err error, showRowsAffected bool, result Result, format string, args []interface{}, rows Rows, dest []interface{})
 }
 
+// Page is a struct for storing pagination information
+type Page struct {
+	Limit  int64
+	Offset int64
+}
+
+// SelectCtrl is a struct for storing select control information
+type SelectCtrl struct {
+	Fields []string // empty means select count
+	Where  map[string][]string
+	Order  []string
+	Page   Page
+
+	OptimizeConditions bool
+}
+
 // databaseSearcher is an interface for searching the database
 type databaseSearcher interface {
-	Search(from string, what string, where string, orderBy string, limit int, explain bool, args ...interface{}) (Rows, error)
-	Aggregate(from string, what string, where string, groupBy string, orderBy string, limit int, explain bool, args ...interface{}) (Rows, error)
+	SearchRaw(from string, what string, where string, orderBy string, limit int, explain bool, args ...interface{}) (Rows, error)
+	Search(tableName string, c *SelectCtrl) (Rows, error)
+}
+
+// InsertStats is a struct for storing insert statistics
+type InsertStats struct {
+	Successful        int64
+	Failed            int64
+	Total             int64
+	ExpectedSuccesses int64
+}
+
+func (s *InsertStats) String() string {
+	return fmt.Sprintf("successful: %d, failed: %d, total: %d", s.Successful, s.Failed, s.Total)
 }
 
 // databaseInserter is an interface for inserting data into the database
@@ -149,14 +198,22 @@ type Session interface {
 
 type TableRow struct {
 	Name    string
-	Type    string
+	Type    DataType
 	NotNull bool
+	Indexed bool // only for Elasticsearch
+}
+
+type ResilienceSettings struct {
+	NumberOfShards   int
+	NumberOfReplicas int
 }
 
 type TableDefinition struct {
 	TableRows  []TableRow
 	PrimaryKey []string
 	Engine     string
+	Resilience ResilienceSettings
+	LMPolicy   string // only for Elasticsearch
 }
 
 type IndexType string
@@ -235,9 +292,41 @@ type Database interface {
 	Close() error
 }
 
+type DataType string
+
+const (
+	DataTypeId                DataType = "{$id}"
+	DataTypeInt               DataType = "{$int}"
+	DataTypeString            DataType = "{$string}"
+	DataTypeString256         DataType = "{$string256}"
+	DataTypeBigIntAutoIncPK   DataType = "{$bigint_autoinc_pk}"
+	DataTypeBigIntAutoInc     DataType = "{$bigint_autoinc}"
+	DataTypeAscii             DataType = "{$ascii}"
+	DataTypeUUID              DataType = "{$uuid}"
+	DataTypeVarCharUUID       DataType = "{$varchar_uuid}"
+	DataTypeLongBlob          DataType = "{$longblob}"
+	DataTypeHugeBlob          DataType = "{$hugeblob}"
+	DataTypeDateTime          DataType = "{$datetime}"
+	DataTypeDateTime6         DataType = "{$datetime6}"
+	DataTypeTimestamp6        DataType = "{$timestamp6}"
+	DataTypeCurrentTimeStamp6 DataType = "{$current_timestamp6}"
+	DataTypeBinary20          DataType = "{$binary20}"
+	DataTypeBinaryBlobType    DataType = "{$binaryblobtype}"
+	DataTypeBoolean           DataType = "{$boolean}"
+	DataTypeBooleanFalse      DataType = "{$boolean_false}"
+	DataTypeBooleanTrue       DataType = "{$boolean_true}"
+	DataTypeTinyInt           DataType = "{$tinyint}"
+	DataTypeLongText          DataType = "{$longtext}"
+	DataTypeUnique            DataType = "{$unique}"
+	DataTypeEngine            DataType = "{$engine}"
+	DataTypeNotNull           DataType = "{$notnull}"
+	DataTypeNull              DataType = "{$null}"
+	DataTypeTenantUUIDBoundID DataType = "{$tenant_uuid_bound_id}"
+)
+
 // Dialect is an interface for database dialects
 type Dialect interface {
-	GetType(id string) string
+	GetType(id DataType) string
 }
 
 // Recommendation is a struct for storing DB recommendation
@@ -267,6 +356,7 @@ func GetDatabases() []DBType {
 	ret = append(ret, DBType{Driver: CLICKHOUSE, Symbol: "C", Name: "ClickHouse"})
 	// "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"})
 
 	return ret
 }