Refactor towards better table creation (#1439)

jakozaur · web-flow · commit 6803ecdd9f46 · 2025-06-04T14:26:48.000Z
Our logic of parsing `CREATE TABLE` needs to be replaced. This is code
duplication and a potential issue.

First steps:
1. Add logic to log proper bugs instead of eating them silently.
2. Add some tests that reproduce the client issue. Though we decided
against fixing our parser.
3. Start by removing the confusing field `Created`. It is not used now,
it is artifact from legacy hardcoding table.
diff --git a/platform/clickhouse/clickhouse.go b/platform/clickhouse/clickhouse.go
@@ -318,16 +318,12 @@ func (lm *LogManager) GetTableDefinitions() (TableMap, error) {
 func (lm *LogManager) AddTableIfDoesntExist(table *Table) bool {
 	t := lm.FindTable(table.Name)
 	if t == nil {
-		table.Created = true
-
 		table.ApplyIndexConfig(lm.cfg)
 
 		lm.tableDiscovery.TableDefinitions().Store(table.Name, table)
 		return true
 	}
-	wasntCreated := !t.Created
-	t.Created = true
-	return wasntCreated
+	return false
 }
 
 func (lm *LogManager) Ping() error {
diff --git a/platform/clickhouse/clickhouse_test.go b/platform/clickhouse/clickhouse_test.go
@@ -468,8 +468,7 @@ func TestJsonConvertingBoolToStringAttr(t *testing.T) {
 // Doesn't test for 100% equality, as map iteration order isn't deterministic, but should definitely be good enough.
 func TestCreateTableString_1(t *testing.T) {
 	table := Table{
-		Created: false,
-		Name:    "/_bulk?refresh=false&_source_includes=originId&require_alias=true_16",
+		Name: "/_bulk?refresh=false&_source_includes=originId&require_alias=true_16",
 		Cols: map[string]*Column{
 			"doc": {
 				Name: "doc",
@@ -570,8 +569,7 @@ func TestCreateTableString_1(t *testing.T) {
 // Doesn't test for 100% equality, as map iteration order isn't deterministic, but should definitely be good enough.
 func TestCreateTableString_NewDateTypes(t *testing.T) {
 	table := Table{
-		Created: false,
-		Name:    "abc",
+		Name: "abc",
 		Cols: map[string]*Column{
 			"low_card_string": {
 				Name:    "low_card_string",
diff --git a/platform/clickhouse/table.go b/platform/clickhouse/table.go
@@ -19,7 +19,6 @@ type Table struct {
 	ClusterName  string `default:""`
 	Cols         map[string]*Column
 	Config       *ChTableConfig
-	Created      bool // do we need to create it during first insert
 	Indexes      []IndexStatement
 	aliases      map[string]string //deprecated
 	// TODO: we should use aliases directly from configuration, not store them here
diff --git a/platform/clickhouse/table_discovery.go b/platform/clickhouse/table_discovery.go
@@ -412,7 +412,6 @@ func (td *tableDiscovery) populateTableDefinitions(configuredTables map[string]d
 
 		if !partiallyResolved {
 			table := Table{
-				Created:      true,
 				Name:         tableName,
 				Comment:      resTable.comment,
 				DatabaseName: databaseName,
diff --git a/platform/frontend_connectors/count_test.go b/platform/frontend_connectors/count_test.go
@@ -29,16 +29,16 @@ func TestCountEndpoint(t *testing.T) {
 
 	tables := clickhouse.NewTableMap()
 	tables.Store("no_db_name", &clickhouse.Table{
-		Name: "no_db_name", Config: clickhouse.NewChTableConfigTimestampStringAttr(), Created: true, Cols: map[string]*clickhouse.Column{},
+		Name: "no_db_name", Config: clickhouse.NewChTableConfigTimestampStringAttr(), Cols: map[string]*clickhouse.Column{},
 	})
 	tables.Store("with_db_name", &clickhouse.Table{
-		Name: "with_db_name", Config: clickhouse.NewChTableConfigTimestampStringAttr(), Created: true, Cols: map[string]*clickhouse.Column{}, DatabaseName: "db_name",
+		Name: "with_db_name", Config: clickhouse.NewChTableConfigTimestampStringAttr(), Cols: map[string]*clickhouse.Column{}, DatabaseName: "db_name",
 	})
 	tables.Store("common_prefix_1", &clickhouse.Table{
-		Name: "common_prefix_1", Config: clickhouse.NewChTableConfigTimestampStringAttr(), Created: true, Cols: map[string]*clickhouse.Column{}, DatabaseName: "db_name",
+		Name: "common_prefix_1", Config: clickhouse.NewChTableConfigTimestampStringAttr(), Cols: map[string]*clickhouse.Column{}, DatabaseName: "db_name",
 	})
 	tables.Store("common_prefix_2", &clickhouse.Table{
-		Name: "common_prefix_2", Config: clickhouse.NewChTableConfigTimestampStringAttr(), Created: true, Cols: map[string]*clickhouse.Column{},
+		Name: "common_prefix_2", Config: clickhouse.NewChTableConfigTimestampStringAttr(), Cols: map[string]*clickhouse.Column{},
 	})
 
 	conn, mock := util.InitSqlMockWithPrettySqlAndPrint(t, false)
diff --git a/platform/frontend_connectors/search_opensearch_test.go b/platform/frontend_connectors/search_opensearch_test.go
@@ -27,7 +27,6 @@ func TestSearchOpensearch(t *testing.T) {
 			"message_____": {Name: "message_____", Type: clickhouse.NewBaseType("String")},
 			"__bytes":      {Name: "__bytes", Type: clickhouse.NewBaseType("Int64")},
 		},
-		Created: true,
 	}
 
 	s := &schema.StaticRegistry{Tables: map[schema.IndexName]schema.Schema{}}
@@ -174,7 +173,6 @@ func TestHighlighter(t *testing.T) {
 			"host_name":    {Name: "host_name", Type: clickhouse.NewBaseType("String")},
 			"_timestamp":   {Name: "_timestamp", Type: clickhouse.NewBaseType("DateTime64")},
 		},
-		Created: true,
 	}
 	fields := map[schema.FieldName]schema.Field{
 		"messeage$*%:;": {PropertyName: "message$*%:;", InternalPropertyName: "message_____", Type: schema.QuesmaTypeText},
diff --git a/platform/frontend_connectors/search_test.go b/platform/frontend_connectors/search_test.go
@@ -69,7 +69,6 @@ func TestAsyncSearchHandler(t *testing.T) {
 				Type: clickhouse.NewBaseType("String"),
 			},
 		},
-		Created: true,
 	})
 	fields := map[schema.FieldName]schema.Field{
 		"@timestamp":        {PropertyName: "@timestamp", InternalPropertyName: "@timestamp", Type: schema.QuesmaTypeDate},
@@ -114,7 +113,6 @@ func TestAsyncSearchHandlerSpecialCharacters(t *testing.T) {
 			"message_____": {Name: "message_____", Type: clickhouse.NewBaseType("String")},
 			"__bytes":      {Name: "__bytes", Type: clickhouse.NewBaseType("Int64")},
 		},
-		Created: true,
 	}
 	fields := map[schema.FieldName]schema.Field{
 		"host.name":         {PropertyName: "host.name", InternalPropertyName: "host_name", Type: schema.QuesmaTypeObject},
@@ -168,7 +166,6 @@ var table = util.NewSyncMapWith(tableName, &clickhouse.Table{
 			Type: clickhouse.NewBaseType("String"),
 		},
 	},
-	Created: true,
 })
 
 func TestSearchHandler(t *testing.T) {
@@ -262,7 +259,6 @@ func TestSearchHandler(t *testing.T) {
 				Type: clickhouse.NewBaseType("String"),
 			},
 		},
-		Created: true,
 	}
 
 	tableMap := util.NewSyncMapWith(tableName, tab)
@@ -373,7 +369,6 @@ func TestSearchHandlerRuntimeMappings(t *testing.T) {
 				Type: clickhouse.NewBaseType("String"),
 			},
 		},
-		Created: true,
 	})
 
 	s := &schema.StaticRegistry{
@@ -425,7 +420,6 @@ func TestSearchHandlerNoAttrsConfig(t *testing.T) {
 			"message":    {Name: "message", Type: clickhouse.NewBaseType("String")},
 			"@timestamp": {Name: "@timestamp", Type: clickhouse.NewBaseType("DateTime64")},
 		},
-		Created: true,
 	})
 
 	s := &schema.StaticRegistry{
@@ -477,7 +471,6 @@ func TestAsyncSearchFilter(t *testing.T) {
 				Type: clickhouse.BaseType{Name: "Map(String, String)"},
 			},
 		},
-		Created: true,
 	}
 	s := &schema.StaticRegistry{
 		Tables: map[schema.IndexName]schema.Schema{
@@ -553,7 +546,7 @@ func TestHandlingDateTimeFields(t *testing.T) {
 		},
 	}
 
-	table := clickhouse.Table{Name: tableName, Config: clickhouse.NewChTableConfigTimestampStringAttr(), Created: true,
+	table := clickhouse.Table{Name: tableName, Config: clickhouse.NewChTableConfigTimestampStringAttr(),
 		Cols: map[string]*clickhouse.Column{
 			"timestamp":   {Name: "timestamp", Type: clickhouse.NewBaseType("DateTime")},
 			"timestamp64": {Name: "timestamp64", Type: clickhouse.NewBaseType("DateTime64")},
@@ -685,7 +678,6 @@ func TestNumericFacetsQueries(t *testing.T) {
 				Type: clickhouse.NewBaseType("Float64"),
 			},
 		},
-		Created: true,
 	})
 	handlers := []string{"handleSearch", "handleAsyncSearch"}
 	for i, tt := range testdata.TestsNumericFacets {
@@ -779,7 +771,6 @@ func TestSearchTrackTotalCount(t *testing.T) {
 			"message":    {Name: "message", Type: clickhouse.NewBaseType("String")},
 			"@timestamp": {Name: "@timestamp", Type: clickhouse.NewBaseType("DateTime64")},
 		},
-		Created: true,
 	})
 
 	test := func(handlerName string, testcase testdata.FullSearchTestCase) {
@@ -885,7 +876,6 @@ func TestFullQueryTestWIP(t *testing.T) {
 			"message":    {Name: "message", Type: clickhouse.NewBaseType("String")},
 			"@timestamp": {Name: "@timestamp", Type: clickhouse.NewBaseType("DateTime64")},
 		},
-		Created: true,
 	})
 
 	test := func(handlerName string, testcase testdata.FullSearchTestCase) {
@@ -996,7 +986,6 @@ func TestSearchAfterParameter_sortByJustTimestamp(t *testing.T) {
 			"message":    {Name: "message", Type: clickhouse.NewBaseType("String")},
 			"@timestamp": {Name: "@timestamp", Type: clickhouse.NewBaseType("DateTime64")},
 		},
-		Created: true,
 	})
 
 	someTime := time.Date(2024, 1, 29, 18, 11, 36, 491000000, time.UTC) // 1706551896491 in UnixMilli
@@ -1133,7 +1122,6 @@ func TestSearchAfterParameter_sortByJustOneStringField(t *testing.T) {
 		Cols: map[string]*clickhouse.Column{
 			"message": {Name: "message", Type: clickhouse.NewBaseType("String")},
 		},
-		Created: true,
 	})
 
 	iterations := []struct {
@@ -1252,7 +1240,6 @@ func TestSearchAfterParameter_sortByMultipleFields(t *testing.T) {
 			"bicep_size": {Name: "bicep_size", Type: clickhouse.NewBaseType("Int64")},
 			"@timestamp": {Name: "@timestamp", Type: clickhouse.NewBaseType("DateTime64")},
 		},
-		Created: true,
 	})
 
 	someTime := time.Date(2024, 1, 29, 18, 11, 36, 491000000, time.UTC) // 1706551896491 in UnixMilli
@@ -1410,7 +1397,6 @@ func TestSearchAfterParameter_sortByNoField(t *testing.T) {
 			"bicep_size": {Name: "bicep_size", Type: clickhouse.NewBaseType("Int64")},
 			"@timestamp": {Name: "@timestamp", Type: clickhouse.NewBaseType("DateTime64")},
 		},
-		Created: true,
 	})
 
 	someTime := time.Date(2024, 1, 29, 18, 11, 36, 491000000, time.UTC) // 1706551896491 in UnixMilli
diff --git a/platform/frontend_connectors/terms_enum_test.go b/platform/frontend_connectors/terms_enum_test.go
@@ -94,7 +94,6 @@ func testHandleTermsEnumRequest(t *testing.T, requestBody []byte, fieldName stri
 				Type: clickhouse.NewBaseType("LowCardinality(String)"),
 			},
 		},
-		Created: true,
 	}
 	tableResolver := table_resolver.NewEmptyTableResolver()
 	managementConsole := ui.NewQuesmaManagementConsole(&config.QuesmaConfiguration{}, nil, make(<-chan logger.LogWithLevel, 50000), diag.EmptyPhoneHomeRecentStatsProvider(), nil, tableResolver)
diff --git a/platform/ingest/common_table_test.go b/platform/ingest/common_table_test.go
@@ -171,8 +171,7 @@ func TestIngestToCommonTable(t *testing.T) {
 						Type: clickhouse.BaseType{Name: "Map(String, String)"},
 					},
 				},
-				Config:  NewDefaultCHConfig(),
-				Created: true,
+				Config: NewDefaultCHConfig(),
 			}
 
 			for _, col := range tt.alreadyExistingColumns {
@@ -220,7 +219,6 @@ func TestIngestToCommonTable(t *testing.T) {
 					Name:         indexName,
 					Cols:         map[string]*clickhouse.Column{},
 					Config:       NewDefaultCHConfig(),
-					Created:      true,
 					VirtualTable: true,
 				}
 
diff --git a/platform/ingest/ingest_validator_test.go b/platform/ingest/ingest_validator_test.go
@@ -234,7 +234,6 @@ func TestIngestValidation(t *testing.T) {
 				},
 			}},
 		},
-		Created: true,
 	})
 	for i := range inputJson {
 		conn, mock := util.InitSqlMockWithPrettyPrint(t, true)
diff --git a/platform/ingest/insert_test.go b/platform/ingest/insert_test.go
@@ -151,7 +151,6 @@ func ingestProcessorsNonEmpty(cfg *clickhouse.ChTableConfig) []ingestProcessorHe
 				"message":          lowCardinalityString("message"),
 				"non-insert-field": genericString("non-insert-field"),
 			},
-			Created: created,
 		})
 		lms = append(lms, ingestProcessorHelper{newIngestProcessorWithEmptyTableMap(full, &config.QuesmaConfiguration{}), created})
 	}
@@ -198,7 +197,7 @@ func TestAutomaticTableCreationAtInsert(t *testing.T) {
 					// check if we properly create table in our tables table :) (:) suggested by Copilot) if needed
 					tableInMemory := ip.ip.FindTable(tableName)
 					needCreate := true
-					if tableInMemory != nil && tableInMemory.Created {
+					if tableInMemory != nil {
 						needCreate = false
 					}
 					noSuchTable := ip.ip.AddTableIfDoesntExist(table)
@@ -207,7 +206,6 @@ func TestAutomaticTableCreationAtInsert(t *testing.T) {
 					// and Created is set to true
 					tableInMemory = ip.ip.FindTable(tableName)
 					assert.NotNil(t, tableInMemory)
-					assert.True(t, tableInMemory.Created)
 
 					// and we have a schema in memory in every case
 					assert.Equal(t, 1, ip.ip.tableDiscovery.TableDefinitions().Size())
@@ -253,9 +251,7 @@ func TestProcessInsertQuery(t *testing.T) {
 					// info: result values aren't important, this '.WillReturnResult[...]' just needs to be there
 					if !ip.tableAlreadyCreated {
 						// we check here if we try to create table from predefined schema, not from insert's JSON
-						if ip.ip.tableDiscovery.TableDefinitions().Size() > 0 {
-							mock.ExpectExec(`CREATE TABLE IF NOT EXISTS "` + tableName + `.*non-insert-field`).WillReturnResult(sqlmock.NewResult(0, 0))
-						} else {
+						if ip.ip.tableDiscovery.TableDefinitions().Size() == 0 {
 							mock.ExpectExec(`CREATE TABLE IF NOT EXISTS "` + tableName).WillReturnResult(sqlmock.NewResult(0, 0))
 						}
 					}
@@ -311,10 +307,9 @@ func TestInsertVeryBigIntegers(t *testing.T) {
 
 	// big integer as an attribute field
 	tableMapNoSchemaFields := util.NewSyncMapWith(tableName, &clickhouse.Table{
-		Name:    tableName,
-		Config:  NewChTableConfigFourAttrs(),
-		Cols:    map[string]*clickhouse.Column{},
-		Created: true,
+		Name:   tableName,
+		Config: NewChTableConfigFourAttrs(),
+		Cols:   map[string]*clickhouse.Column{},
 	})
 
 	for i, bigInt := range bigInts {
diff --git a/platform/ingest/parserCreateTable_test.go b/platform/ingest/parserCreateTable_test.go
@@ -364,3 +364,24 @@ func TestParseCreateTableWithNullable(t *testing.T) {
 	assert.True(t, table.Cols["array-tuple"].Type.(clickhouse.CompoundType).BaseType.(clickhouse.MultiValueType).Cols[0].Type.IsNullable())
 	assert.False(t, table.Cols["array-tuple"].Type.(clickhouse.CompoundType).BaseType.(clickhouse.MultiValueType).Cols[1].Type.IsNullable())
 }
+
+func IgnoredTestParseCreateTableWithDots(t *testing.T) {
+	const columnNr = 5
+	q := `CREATE TABLE IF NOT EXISTS "my-index-2.3" 
+			(
+				"attributes_values" Map(String,String),
+				"attributes_metadata" Map(String,String),
+			
+				"level" Nullable(String) COMMENT 'quesmaMetadataV1:fieldName=level',
+				"@timestamp" DateTime64 DEFAULT now64() COMMENT 'quesmaMetadataV1:fieldName=%40timestamp',
+				"message" Nullable(String) COMMENT 'quesmaMetadataV1:fieldName=message'
+			)
+			ENGINE = MergeTree
+			ORDER BY ("@timestamp")
+			
+			COMMENT 'created by Quesma'`
+	table, err := clickhouse.NewTable(q, nil)
+	assert.NoError(t, err)
+	assert.Equal(t, columnNr, len(table.Cols))
+	assert.Equal(t, "my-index-2.3", table.Name)
+}
diff --git a/platform/ingest/processor.go b/platform/ingest/processor.go
@@ -667,11 +667,11 @@ func (ip *IngestProcessor) processInsertQuery(ctx context.Context,
 		if err != nil {
 			logger.ErrorWithCtx(ctx).Msgf("error createTableObjectAndAttributes, can't create table: %v", err)
 			return nil, err
+		} else {
+			logger.InfoWithCtx(ctx).Msgf("created table '%s' with query: %s", tableName, createTableCmd)
 		}
 		// Set pointer to table after creating it
 		table = ip.FindTable(tableName)
-	} else if !table.Created {
-		createTableCmd = table.CreateTableString()
 	}
 	if table == nil {
 		return nil, fmt.Errorf("table %s not found", tableName)
@@ -824,15 +824,18 @@ func (ip *IngestProcessor) processInsertQueryInternal(ctx context.Context, table
 	tableFormatter TableColumNameFormatter, isVirtualTable bool) error {
 	statements, err := ip.processInsertQuery(ctx, tableName, jsonData, transformer, tableFormatter, isVirtualTable)
 	if err != nil {
+		logger.ErrorWithCtx(ctx).Msgf("error processing insert query: %v", err)
 		return err
 	}
 
 	var logVirtualTableDDL bool // maybe this should be a part of the config or sth
 
-	if isVirtualTable && logVirtualTableDDL {
-		for _, statement := range statements {
-			if strings.HasPrefix(statement, "ALTER") || strings.HasPrefix(statement, "CREATE") {
+	for _, statement := range statements {
+		if strings.HasPrefix(statement, "ALTER") || strings.HasPrefix(statement, "CREATE") {
+			if isVirtualTable && logVirtualTableDDL {
 				logger.InfoWithCtx(ctx).Msgf("VIRTUAL DDL EXECUTION: %s", statement)
+			} else {
+				logger.InfoWithCtx(ctx).Msgf("DDL EXECUTION: %s", statement)
 			}
 		}
 	}
@@ -975,8 +978,6 @@ func (ip *IngestProcessor) storeVirtualTable(table *chLib.Table) error {
 func (ip *IngestProcessor) AddTableIfDoesntExist(table *chLib.Table) bool {
 	t := ip.FindTable(table.Name)
 	if t == nil {
-		table.Created = true
-
 		table.ApplyIndexConfig(ip.cfg)
 
 		if table.VirtualTable {
@@ -988,9 +989,7 @@ func (ip *IngestProcessor) AddTableIfDoesntExist(table *chLib.Table) bool {
 		ip.tableDiscovery.AddTable(table.Name, table)
 		return true
 	}
-	wasntCreated := !t.Created
-	t.Created = true
-	return wasntCreated
+	return false
 }
 
 func (ip *IngestProcessor) GetSchemaRegistry() schema.Registry {
diff --git a/platform/ingest/processor_test.go b/platform/ingest/processor_test.go
@@ -604,8 +604,7 @@ func TestJsonConvertingBoolToStringAttr(t *testing.T) {
 // Doesn't test for 100% equality, as map iteration order isn't deterministic, but should definitely be good enough.
 func TestCreateTableString_1(t *testing.T) {
 	table := clickhouse.Table{
-		Created: false,
-		Name:    "/_bulk?refresh=false&_source_includes=originId&require_alias=true_16",
+		Name: "/_bulk?refresh=false&_source_includes=originId&require_alias=true_16",
 		Cols: map[string]*clickhouse.Column{
 			"doc": {
 				Name: "doc",
@@ -706,8 +705,7 @@ func TestCreateTableString_1(t *testing.T) {
 // Doesn't test for 100% equality, as map iteration order isn't deterministic, but should definitely be good enough.
 func TestCreateTableString_NewDateTypes(t *testing.T) {
 	table := clickhouse.Table{
-		Created: false,
-		Name:    "abc",
+		Name: "abc",
 		Cols: map[string]*clickhouse.Column{
 			"low_card_string": {
 				Name: "low_card_string",
diff --git a/platform/parsers/elastic_query_dsl/query_parser_test.go b/platform/parsers/elastic_query_dsl/query_parser_test.go
diff --git a/platform/ui/html_pages_test.go b/platform/ui/html_pages_test.go

Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,6 @@ func TestSearchOpensearch(t *testing.T) {`
`27`	`27`	`"message_____": {Name: "message_____", Type: clickhouse.NewBaseType("String")},`
`28`	`28`	`"__bytes": {Name: "__bytes", Type: clickhouse.NewBaseType("Int64")},`
`29`	`29`	`},`
`30`		`- Created: true,`
`31`	`30`	`}`
`32`	`31`
`33`	`32`	`s := &schema.StaticRegistry{Tables: map[schema.IndexName]schema.Schema{}}`
`@@ -174,7 +173,6 @@ func TestHighlighter(t *testing.T) {`
`174`	`173`	`"host_name": {Name: "host_name", Type: clickhouse.NewBaseType("String")},`
`175`	`174`	`"_timestamp": {Name: "_timestamp", Type: clickhouse.NewBaseType("DateTime64")},`
`176`	`175`	`},`
`177`		`- Created: true,`
`178`	`176`	`}`
`179`	`177`	`fields := map[schema.FieldName]schema.Field{`
`180`	`178`	`"messeage$%:;": {PropertyName: "message$%:;", InternalPropertyName: "message_____", Type: schema.QuesmaTypeText},`
Original file line number	Diff line number	Diff line change
`@@ -94,7 +94,6 @@ func testHandleTermsEnumRequest(t *testing.T, requestBody []byte, fieldName stri`
`94`	`94`	`Type: clickhouse.NewBaseType("LowCardinality(String)"),`
`95`	`95`	`},`
`96`	`96`	`},`
`97`		`- Created: true,`
`98`	`97`	`}`
`99`	`98`	`tableResolver := table_resolver.NewEmptyTableResolver()`
`100`	`99`	`managementConsole := ui.NewQuesmaManagementConsole(&config.QuesmaConfiguration{}, nil, make(<-chan logger.LogWithLevel, 50000), diag.EmptyPhoneHomeRecentStatsProvider(), nil, tableResolver)`