Fix resolving indices when tables are auto-discovered (#765)

Users who haven't specified any table configurations aren't able to create data views today. 1. Because `_resolve` endpoint relies on schema registry, which - as turned out - has never been populated when the index configuration has been missing. 2. Because `_field_caps` were **not** routed properly. `matchedAgainstPattern` did rely only on config, not on schema registry, so it couldn't route queries to auto discovered tables. Therefore, timestamp field list has not been expanding and data view could not have been created. This PR addresses both issues. **How it works now:** <img width="1173" alt="image" src="https://github.com/user-attachments/assets/5845efb1-ecc1-4d56-86d0-1438c9cb888e">
QuesmaOrg · Sep 13, 2024 · 1e0eb3e · 1e0eb3e
1 parent 54d168a
commit 1e0eb3e
Show file tree

Hide file tree

Showing 11 changed files with 60 additions and 31 deletions.
diff --git a/quesma/clickhouse/table_discovery.go b/quesma/clickhouse/table_discovery.go
@@ -36,6 +36,7 @@ type TableDiscovery interface {
 	LastAccessTime() time.Time
 	LastReloadTime() time.Time
 	ForceReloadCh() <-chan chan<- struct{}
+	AutodiscoveryEnabled() bool
 }
 
 type tableDiscovery struct {
@@ -100,8 +101,8 @@ func (td *tableDiscovery) TableDefinitionsFetchError() error {
 	return td.ReloadTablesError
 }
 
-func (td *tableDiscovery) TableAutodiscoveryEnabled() bool {
-	return len(td.cfg.IndexConfig) == 0
+func (td *tableDiscovery) AutodiscoveryEnabled() bool {
+	return td.cfg.IndexAutodiscoveryEnabled()
 }
 
 func (td *tableDiscovery) LastAccessTime() time.Time {
@@ -138,7 +139,7 @@ func (td *tableDiscovery) ReloadTableDefinitions() {
 		td.tableDefinitionsLastReloadUnixSec.Store(time.Now().Unix())
 		return
 	} else {
-		if td.TableAutodiscoveryEnabled() {
+		if td.AutodiscoveryEnabled() {
 			configuredTables = td.autoConfigureTables(tables, databaseName)
 		} else {
 			configuredTables = td.configureTables(tables, databaseName)

diff --git a/quesma/quesma/config/config.go b/quesma/quesma/config/config.go
@@ -351,11 +351,6 @@ func (c *QuesmaConfiguration) validateSchemaConfiguration(config IndexConfigurat
 	return err
 }
 
-//func countPrimaryKeys(config IndexConfiguration) (count int) {
-//	for _, configuration := range config.SchemaOverrides.Fields {
-//		if configuration.IsPrimaryKey {
-//			count++
-//		}
-//	}
-//	return count
-//}
+func (c *QuesmaConfiguration) IndexAutodiscoveryEnabled() bool {
+	return len(c.IndexConfig) == 0
+}
diff --git a/quesma/quesma/matchers.go b/quesma/quesma/matchers.go
@@ -8,6 +8,7 @@ import (
 	"quesma/quesma/config"
 	"quesma/quesma/mux"
 	"quesma/quesma/types"
+	"quesma/schema"
 	"quesma/tracing"
 	"strings"
 )
@@ -44,7 +45,7 @@ func matchedAgainstBulkBody(configuration *config.QuesmaConfiguration) mux.Reque
 	})
 }
 
-func matchedAgainstPattern(configuration *config.QuesmaConfiguration) mux.RequestMatcher {
+func matchedAgainstPattern(configuration *config.QuesmaConfiguration, sr schema.Registry) mux.RequestMatcher {
 	return mux.RequestMatcherFunc(func(req *mux.Request) bool {
 		indexPattern := elasticsearch.NormalizePattern(req.Params["index"])
 		if elasticsearch.IsInternalIndex(indexPattern) {
@@ -61,7 +62,13 @@ func matchedAgainstPattern(configuration *config.QuesmaConfiguration) mux.Reques
 					return false
 				}
 			}
-
+			if configuration.IndexAutodiscoveryEnabled() {
+				for tableName := range sr.AllSchemas() {
+					if config.MatchName(elasticsearch.NormalizePattern(indexPattern), string(tableName)) {
+						return true
+					}
+				}
+			}
 			for _, pattern := range indexPatterns {
 				for _, indexName := range configuration.IndexConfig {
 					if config.MatchName(elasticsearch.NormalizePattern(pattern), indexName.Name) {
@@ -73,6 +80,13 @@ func matchedAgainstPattern(configuration *config.QuesmaConfiguration) mux.Reques
 			}
 			return false
 		} else {
+			if configuration.IndexAutodiscoveryEnabled() {
+				for tableName := range sr.AllSchemas() {
+					if config.MatchName(elasticsearch.NormalizePattern(indexPattern), string(tableName)) {
+						return true
+					}
+				}
+			}
 			for _, index := range configuration.IndexConfig {
 				pattern := elasticsearch.NormalizePattern(indexPattern)
 				if config.MatchName(pattern, index.Name) {

diff --git a/quesma/quesma/router.go b/quesma/quesma/router.go
@@ -98,7 +98,7 @@ func configureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		return resolveIndexResult(sources)
 	})
 
-	router.Register(routes.IndexCountPath, and(method("GET"), matchedAgainstPattern(cfg)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
+	router.Register(routes.IndexCountPath, and(method("GET"), matchedAgainstPattern(cfg, sr)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
 		cnt, err := queryRunner.handleCount(ctx, req.Params["index"])
 		if err != nil {
 			if errors.Is(quesma_errors.ErrIndexNotExists(), err) {
@@ -137,7 +137,7 @@ func configureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		return elasticsearchQueryResult(string(responseBody), http.StatusOK), nil
 	})
 
-	router.Register(routes.IndexSearchPath, and(method("GET", "POST"), matchedAgainstPattern(cfg)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
+	router.Register(routes.IndexSearchPath, and(method("GET", "POST"), matchedAgainstPattern(cfg, sr)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
 
 		body, err := types.ExpectJSON(req.ParsedBody)
 		if err != nil {
@@ -159,7 +159,7 @@ func configureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		}
 		return elasticsearchQueryResult(string(responseBody), http.StatusOK), nil
 	})
-	router.Register(routes.IndexAsyncSearchPath, and(method("POST"), matchedAgainstPattern(cfg)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
+	router.Register(routes.IndexAsyncSearchPath, and(method("POST"), matchedAgainstPattern(cfg, sr)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
 		waitForResultsMs := 1000 // Defaults to 1 second as in docs
 		if v, ok := req.Params["wait_for_completion_timeout"]; ok {
 			if w, err := time.ParseDuration(v); err == nil {
@@ -196,7 +196,7 @@ func configureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		return elasticsearchQueryResult(string(responseBody), http.StatusOK), nil
 	})
 
-	router.Register(routes.IndexMappingPath, and(method("PUT"), matchedAgainstPattern(cfg)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
+	router.Register(routes.IndexMappingPath, and(method("PUT"), matchedAgainstPattern(cfg, sr)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
 		index := req.Params["index"]
 
 		body, err := types.ExpectJSON(req.ParsedBody)
@@ -211,7 +211,7 @@ func configureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		return putIndexResult(index)
 	})
 
-	router.Register(routes.IndexMappingPath, and(method("GET"), matchedAgainstPattern(cfg)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
+	router.Register(routes.IndexMappingPath, and(method("GET"), matchedAgainstPattern(cfg, sr)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
 		index := req.Params["index"]
 
 		foundSchema, found := sr.FindSchema(schema.TableName(index))
@@ -242,7 +242,7 @@ func configureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		return elasticsearchQueryResult(string(responseBody), http.StatusOK), nil
 	})
 
-	router.Register(routes.FieldCapsPath, and(method("GET", "POST"), matchedAgainstPattern(cfg)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
+	router.Register(routes.FieldCapsPath, and(method("GET", "POST"), matchedAgainstPattern(cfg, sr)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
 
 		responseBody, err := field_capabilities.HandleFieldCaps(ctx, cfg, sr, req.Params["index"], lm)
 		if err != nil {
@@ -257,7 +257,7 @@ func configureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		}
 		return elasticsearchQueryResult(string(responseBody), http.StatusOK), nil
 	})
-	router.Register(routes.TermsEnumPath, and(method("POST"), matchedAgainstPattern(cfg)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
+	router.Register(routes.TermsEnumPath, and(method("POST"), matchedAgainstPattern(cfg, sr)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
 		if strings.Contains(req.Params["index"], ",") {
 			return nil, errors.New("multi index terms enum is not yet supported")
 		} else {
@@ -278,7 +278,7 @@ func configureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		}
 	})
 
-	router.Register(routes.EQLSearch, and(method("GET", "POST"), matchedAgainstPattern(cfg)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
+	router.Register(routes.EQLSearch, and(method("GET", "POST"), matchedAgainstPattern(cfg, sr)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
 		body, err := types.ExpectJSON(req.ParsedBody)
 		if err != nil {
 			return nil, err
@@ -295,7 +295,7 @@ func configureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		return elasticsearchQueryResult(string(responseBody), http.StatusOK), nil
 	})
 
-	router.Register(routes.IndexPath, and(method("PUT"), matchedAgainstPattern(cfg)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
+	router.Register(routes.IndexPath, and(method("PUT"), matchedAgainstPattern(cfg, sr)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
 		index := req.Params["index"]
 
 		body, err := types.ExpectJSON(req.ParsedBody)
@@ -315,7 +315,7 @@ func configureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		return putIndexResult(index)
 	})
 
-	router.Register(routes.IndexPath, and(method("GET"), matchedAgainstPattern(cfg)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
+	router.Register(routes.IndexPath, and(method("GET"), matchedAgainstPattern(cfg, sr)), func(ctx context.Context, req *mux.Request) (*mux.Result, error) {
 		index := req.Params["index"]
 
 		foundSchema, found := sr.FindSchema(schema.TableName(index))

diff --git a/quesma/quesma/router_test.go b/quesma/quesma/router_test.go
@@ -6,6 +6,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"quesma/quesma/config"
 	"quesma/quesma/mux"
+	"quesma/schema"
 	"testing"
 )
 
@@ -131,8 +132,7 @@ func Test_matchedAgainstPattern(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 
 			req := &mux.Request{Params: map[string]string{"index": tt.pattern}, Body: tt.body}
-
-			assert.Equalf(t, tt.want, matchedAgainstPattern(&tt.configuration).Matches(req), "matchedAgainstPattern(%v)", tt.configuration)
+			assert.Equalf(t, tt.want, matchedAgainstPattern(&tt.configuration, schema.StaticRegistry{}).Matches(req), "matchedAgainstPattern(%v)", tt.configuration)
 		})
 	}
 }

diff --git a/quesma/quesma/schema_transformer_test.go b/quesma/quesma/schema_transformer_test.go
@@ -20,6 +20,8 @@ func (f fixedTableProvider) TableDefinitions() map[string]schema.Table {
 	return f.tables
 }
 
+func (f fixedTableProvider) AutodiscoveryEnabled() bool { return false }
+
 func Test_ipRangeTransform(t *testing.T) {
 	const isIPAddressInRangePrimitive = "isIPAddressInRange"
 	const CASTPrimitive = "CAST"

diff --git a/quesma/quesma/search.go b/quesma/quesma/search.go
@@ -270,7 +270,7 @@ func (q *QueryRunner) executePlan(ctx context.Context, plan *model.ExecutionPlan
 }
 
 func (q *QueryRunner) handleSearchCommon(ctx context.Context, indexPattern string, body types.JSON, optAsync *AsyncQuery, queryLanguage QueryLanguage) ([]byte, error) {
-	sources, sourcesElastic, sourcesClickhouse := ResolveSources(indexPattern, q.cfg, q.im)
+	sources, sourcesElastic, sourcesClickhouse := ResolveSources(indexPattern, q.cfg, q.im, q.schemaRegistry)
 
 	switch sources {
 	case sourceBoth:

diff --git a/quesma/quesma/source_resolver.go b/quesma/quesma/source_resolver.go
@@ -6,6 +6,7 @@ import (
 	"quesma/elasticsearch"
 	"quesma/logger"
 	"quesma/quesma/config"
+	"quesma/schema"
 	"quesma/util"
 	"slices"
 	"strings"
@@ -18,7 +19,7 @@ const (
 	sourceNone          = "none"
 )
 
-func ResolveSources(indexPattern string, cfg *config.QuesmaConfiguration, im elasticsearch.IndexManagement) (string, []string, []string) {
+func ResolveSources(indexPattern string, cfg *config.QuesmaConfiguration, im elasticsearch.IndexManagement, sr schema.Registry) (string, []string, []string) {
 	if elasticsearch.IsIndexPattern(indexPattern) {
 		matchesElastic := []string{}
 		matchesClickhouse := []string{}
@@ -29,6 +30,13 @@ func ResolveSources(indexPattern string, cfg *config.QuesmaConfiguration, im ela
 					matchesElastic = append(matchesElastic, indexName)
 				}
 			}
+			if cfg.IndexAutodiscoveryEnabled() {
+				for tableName := range sr.AllSchemas() {
+					if config.MatchName(elasticsearch.NormalizePattern(indexPattern), string(tableName)) {
+						matchesClickhouse = append(matchesClickhouse, string(tableName))
+					}
+				}
+			}
 
 			for indexName, indexConfig := range cfg.IndexConfig {
 				if util.IndexPatternMatches(pattern, indexName) && !indexConfig.Disabled {

diff --git a/quesma/quesma/source_resolver_test.go b/quesma/quesma/source_resolver_test.go
@@ -6,6 +6,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"quesma/elasticsearch"
 	"quesma/quesma/config"
+	"quesma/schema"
 	"quesma/util"
 	"testing"
 )
@@ -87,7 +88,7 @@ func TestResolveSources(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name+tt.args.indexPattern, func(t *testing.T) {
-			got, _, _ := ResolveSources(tt.args.indexPattern, &tt.args.cfg, tt.args.im)
+			got, _, _ := ResolveSources(tt.args.indexPattern, &tt.args.cfg, tt.args.im, schema.StaticRegistry{})
 			assert.Equalf(t, tt.want, got, "ResolveSources(%v, %v, %v)", tt.args.indexPattern, tt.args.cfg, tt.args.im)
 		})
 	}

diff --git a/quesma/schema/registry.go b/quesma/schema/registry.go
@@ -26,6 +26,7 @@ type (
 	}
 	TableProvider interface {
 		TableDefinitions() map[string]Table
+		AutodiscoveryEnabled() bool
 	}
 	Table struct {
 		Columns map[string]Column
@@ -39,6 +40,14 @@ type (
 func (s *schemaRegistry) loadSchemas() (map[TableName]Schema, error) {
 	definitions := s.dataSourceTableProvider.TableDefinitions()
 	schemas := make(map[TableName]Schema)
+	if s.dataSourceTableProvider.AutodiscoveryEnabled() {
+		for tableName := range definitions {
+			fields := make(map[FieldName]Field)
+			existsInDataSource := s.populateSchemaFromTableDefinition(definitions, tableName, fields)
+			schemas[TableName(tableName)] = NewSchema(fields, existsInDataSource)
+		}
+		return schemas, nil
+	}
 
 	for indexName, indexConfiguration := range *s.indexConfiguration {
 		fields := make(map[FieldName]Field)

diff --git a/quesma/schema/registry_test.go b/quesma/schema/registry_test.go
@@ -335,6 +335,5 @@ type fixedTableProvider struct {
 	tables map[string]schema.Table
 }
 
-func (f fixedTableProvider) TableDefinitions() map[string]schema.Table {
-	return f.tables
-}
+func (f fixedTableProvider) TableDefinitions() map[string]schema.Table { return f.tables }
+func (f fixedTableProvider) AutodiscoveryEnabled() bool                { return false }