Fixing quesma common table dynamic mapping case, second approach (#885)

This is a second approach (1 -
#878) of fixing a problem where
mappings contains more fields than ingest and sql queries are based on
this knowledge. In this approach we are not altering table to have all
fields that were added by dynamic mapping but we based on field origin
information and building queries according to that.

<img width="1706" alt="image"
src="https://github.com/user-attachments/assets/40e52e56-7da5-4346-b138-8f5b738513c4">

Author: pdelewski

quesma/ingest/processor.go

@@ -641,6 +641,15 @@ func (ip *IngestProcessor) processInsertQuery(ctx context.Context,
 		ignoredFields := ip.getIgnoredFields(tableName)
 		columnsFromJson := JsonToColumns("", jsonData[0], 1,
 			tableConfig, tableFormatter, ignoredFields)
+
+		fieldOrigins := make(map[schema.FieldName]schema.FieldSource)
+
+		for _, column := range columnsFromJson {
+			fieldOrigins[schema.FieldName(column.ClickHouseColumnName)] = schema.FieldSourceIngest
+		}
+
+		ip.schemaRegistry.UpdateFieldsOrigins(schema.TableName(tableName), fieldOrigins)
+
 		// This comes externally from (configuration)
 		// So we need to convert that separately
 		columnsFromSchema := SchemaToColumns(findSchemaPointer(ip.schemaRegistry, tableName), tableFormatter, tableName, ip.schemaRegistry.GetFieldEncodings())

quesma/quesma/schema_transformer.go

@@ -463,7 +463,10 @@ func (s *SchemaCheckPass) applyWildcardExpansion(indexSchema schema.Schema, quer
 
 		cols := make([]string, 0, len(indexSchema.Fields))
 		for _, col := range indexSchema.Fields {
-			cols = append(cols, col.InternalPropertyName.AsString())
+			// Take only fields that are ingested
+			if col.Origin == schema.FieldSourceIngest {
+				cols = append(cols, col.InternalPropertyName.AsString())
+			}
 		}
 		sort.Strings(cols)
 
@@ -491,7 +494,10 @@ func (s *SchemaCheckPass) applyFullTextField(indexSchema schema.Schema, query *m
 
 	for _, field := range indexSchema.Fields {
 		if field.Type.IsFullText() {
-			fullTextFields = append(fullTextFields, field.InternalPropertyName.AsString())
+			// Take only fields that are ingested
+			if field.Origin == schema.FieldSourceIngest {
+				fullTextFields = append(fullTextFields, field.InternalPropertyName.AsString())
+			}
 		}
 	}
 

quesma/schema/registry.go

@@ -16,6 +16,7 @@ type (
 	Registry interface {
 		AllSchemas() map[TableName]Schema
 		FindSchema(name TableName) (Schema, bool)
+		UpdateFieldsOrigins(name TableName, fields map[FieldName]FieldSource)
 		UpdateDynamicConfiguration(name TableName, table Table)
 		UpdateFieldEncodings(encodings map[FieldEncodingKey]EncodedFieldName)
 		GetFieldEncodings() map[FieldEncodingKey]EncodedFieldName
@@ -34,6 +35,8 @@ type (
 		dynamicConfiguration    map[string]Table
 		fieldEncodingsLock      sync.RWMutex
 		fieldEncodings          map[FieldEncodingKey]EncodedFieldName
+		fieldOriginsLock        sync.RWMutex
+		fieldOrigins            map[TableName]map[FieldName]FieldSource
 	}
 	typeAdapter interface {
 		Convert(string) (QuesmaType, bool)
@@ -95,6 +98,7 @@ func (s *schemaRegistry) loadSchemas() (map[TableName]Schema, error) {
 		s.populateAliases(indexConfiguration, fields, aliases)
 		s.removeIgnoredFields(indexConfiguration, fields, aliases)
 		s.removeGeoPhysicalFields(fields)
+		s.populateFieldsOrigins(indexName, fields)
 		if tableDefinition, ok := definitions[indexName]; ok {
 			schemas[TableName(indexName)] = NewSchemaWithAliases(fields, aliases, existsInDataSource, tableDefinition.DatabaseName)
 		} else {
@@ -117,7 +121,7 @@ func (s *schemaRegistry) populateSchemaFromDynamicConfiguration(indexName string
 			continue
 		}
 
-		fields[FieldName(column.Name)] = Field{PropertyName: FieldName(column.Name), InternalPropertyName: FieldName(column.Name), Type: columnType}
+		fields[FieldName(column.Name)] = Field{PropertyName: FieldName(column.Name), InternalPropertyName: FieldName(column.Name), Type: columnType, Origin: FieldSourceMapping}
 	}
 }
 
@@ -246,7 +250,7 @@ func (s *schemaRegistry) populateSchemaFromTableDefinition(definitions map[strin
 					fields[propertyName] = Field{PropertyName: propertyName, InternalPropertyName: FieldName(column.Name), InternalPropertyType: column.Type, Type: QuesmaTypeKeyword}
 				}
 			} else {
-				fields[propertyName] = Field{PropertyName: propertyName, InternalPropertyName: FieldName(column.Name), InternalPropertyType: column.Type, Type: existing.Type}
+				fields[propertyName] = Field{PropertyName: propertyName, InternalPropertyName: FieldName(column.Name), InternalPropertyType: column.Type, Type: existing.Type, Origin: existing.Origin}
 			}
 		}
 	}
@@ -275,3 +279,25 @@ func (s *schemaRegistry) removeGeoPhysicalFields(fields map[FieldName]Field) {
 		}
 	}
 }
+
+func (s *schemaRegistry) populateFieldsOrigins(indexName string, fields map[FieldName]Field) {
+	s.fieldOriginsLock.RLock()
+	if fieldOrigins, ok := s.fieldOrigins[TableName(indexName)]; ok {
+		for fieldName, field := range fields {
+			if origin, ok := fieldOrigins[field.InternalPropertyName]; ok {
+				field.Origin = origin
+				fields[fieldName] = field
+			}
+		}
+	}
+	s.fieldOriginsLock.RUnlock()
+}
+
+func (s *schemaRegistry) UpdateFieldsOrigins(name TableName, fields map[FieldName]FieldSource) {
+	s.fieldOriginsLock.Lock()
+	defer s.fieldOriginsLock.Unlock()
+	if s.fieldOrigins == nil {
+		s.fieldOrigins = make(map[TableName]map[FieldName]FieldSource)
+	}
+	s.fieldOrigins[name] = fields
+}

quesma/schema/registry_test.go

@@ -319,7 +319,7 @@ func Test_schemaRegistry_UpdateDynamicConfiguration(t *testing.T) {
 		"message":    {PropertyName: "message", InternalPropertyName: "message", Type: schema.QuesmaTypeKeyword, InternalPropertyType: "String"},
 		"event_date": {PropertyName: "event_date", InternalPropertyName: "event_date", Type: schema.QuesmaTypeTimestamp, InternalPropertyType: "DateTime64"},
 		"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.QuesmaTypeLong, InternalPropertyType: "Int64"},
-		"new_column": {PropertyName: "new_column", InternalPropertyName: "new_column", Type: schema.QuesmaTypeText}},
+		"new_column": {PropertyName: "new_column", InternalPropertyName: "new_column", Type: schema.QuesmaTypeText, Origin: schema.FieldSourceMapping}},
 		true, "")
 	resultSchema, resultFound = s.FindSchema(schema.TableName(tableName))
 	assert.True(t, resultFound, "schema not found")

quesma/schema/schema.go

@@ -6,6 +6,14 @@ import (
 	"strings"
 )
 
+// FieldSource is an enum that represents the source of a field in the schema
+type FieldSource int
+
+const (
+	FieldSourceIngest FieldSource = iota
+	FieldSourceMapping
+)
+
 type (
 	Schema struct {
 		Fields             map[FieldName]Field
@@ -23,6 +31,7 @@ type (
 		InternalPropertyName FieldName
 		InternalPropertyType string
 		Type                 QuesmaType
+		Origin               FieldSource
 	}
 	TableName string
 	FieldName string

quesma/schema/static_registry.go

@@ -46,3 +46,7 @@ func (e *StaticRegistry) GetFieldEncodings() map[FieldEncodingKey]EncodedFieldNa
 	}
 	return e.FieldEncodings
 }
+
+func (e *StaticRegistry) UpdateFieldsOrigins(name TableName, fields map[FieldName]FieldSource) {
+
+}