`)
buffer.Html(fmt.Sprintf(
diff --git a/quesma/quesma/ui/management_console.go b/quesma/quesma/ui/management_console.go
index e54dec9b0..ac89e1bf8 100644
--- a/quesma/quesma/ui/management_console.go
+++ b/quesma/quesma/ui/management_console.go
@@ -8,6 +8,7 @@ import (
"quesma/elasticsearch"
"quesma/quesma/types"
"quesma/schema"
+ "quesma/table_resolver"
"quesma/telemetry"
"quesma/util"
@@ -100,13 +101,14 @@ type (
phoneHomeAgent telemetry.PhoneHomeAgent
schemasProvider SchemasProvider
totalUnsupportedQueries int
+ tableResolver table_resolver.TableResolver
}
SchemasProvider interface {
AllSchemas() map[schema.TableName]schema.Schema
}
)
-func NewQuesmaManagementConsole(cfg *config.QuesmaConfiguration, logManager *clickhouse.LogManager, indexManager elasticsearch.IndexManagement, logChan <-chan logger.LogWithLevel, phoneHomeAgent telemetry.PhoneHomeAgent, schemasProvider SchemasProvider) *QuesmaManagementConsole {
+func NewQuesmaManagementConsole(cfg *config.QuesmaConfiguration, logManager *clickhouse.LogManager, indexManager elasticsearch.IndexManagement, logChan <-chan logger.LogWithLevel, phoneHomeAgent telemetry.PhoneHomeAgent, schemasProvider SchemasProvider, indexRegistry table_resolver.TableResolver) *QuesmaManagementConsole {
return &QuesmaManagementConsole{
queryDebugPrimarySource: make(chan *QueryDebugPrimarySource, 10),
queryDebugSecondarySource: make(chan *QueryDebugSecondarySource, 10),
@@ -124,6 +126,7 @@ func NewQuesmaManagementConsole(cfg *config.QuesmaConfiguration, logManager *cli
indexManagement: indexManager,
phoneHomeAgent: phoneHomeAgent,
schemasProvider: schemasProvider,
+ tableResolver: indexRegistry,
}
}
diff --git a/quesma/quesma/ui/table_resolver.go b/quesma/quesma/ui/table_resolver.go
new file mode 100644
index 000000000..b771fc1e6
--- /dev/null
+++ b/quesma/quesma/ui/table_resolver.go
@@ -0,0 +1,121 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package ui
+
+import (
+ "quesma/quesma/ui/internal/builder"
+ "strings"
+)
+
+func (qmc *QuesmaManagementConsole) generateTableResolver() []byte {
+
+ buffer := newBufferWithHead()
+ buffer.Write(generateTopNavigation("table_resolver"))
+
+ buffer.Html(`
`)
+
+ buffer.Html("Table Resolver
")
+
+ buffer.Html("Ask Quesma
")
+
+ buffer.Html("Ask Quesma to resolve an pattern:")
+ buffer.Html(`
`)
+ buffer.Html(`")
+
+ buffer.Html(``)
+
+ buffer.Html(`
`)
+
+ buffer.Html("Recent decisions
")
+
+ pipelines := qmc.tableResolver.Pipelines()
+
+ buffer.Html(``)
+ buffer.Html(``)
+ buffer.Html(`| Pattern | `)
+ for _, pipeline := range pipelines {
+ buffer.Html(``).Text(pipeline).Html(` | `)
+ }
+ buffer.Html(`
`)
+
+ decisions := qmc.tableResolver.RecentDecisions()
+
+ for _, decision := range decisions {
+ buffer.Html(``)
+ buffer.Html(`| `).Text(decision.Pattern).Html(` | `)
+
+ for _, pipeline := range pipelines {
+ buffer.Html(``)
+ if decision.Decisions[pipeline] != nil {
+ buffer.Text(decision.Decisions[pipeline].String())
+ } else {
+ buffer.Text("n/a")
+ }
+ buffer.Html(` | `)
+ }
+
+ buffer.Html(`
`)
+ }
+ buffer.Html(`
`)
+
+ buffer.Html(`")
+
+ buffer.Html("\n\n\n")
+ return buffer.Bytes()
+
+}
+
+func (qmc *QuesmaManagementConsole) generateTableResolverAnswer(prompt string) []byte {
+ var buffer builder.HtmlBuffer
+
+ prompt = strings.TrimSpace(prompt)
+
+ if prompt == "" {
+ return buffer.Bytes()
+ }
+
+ patterns := strings.Split(prompt, " ")
+
+ pipelines := qmc.tableResolver.Pipelines()
+
+ buffer.Html("
Quesma's decision")
+
+ buffer.Html(`
`)
+ buffer.Html(``)
+ buffer.Html(`| Pattern | `)
+ for _, pipeline := range pipelines {
+ buffer.Html(``).Text(pipeline).Html(` | `)
+ }
+ buffer.Html(`
`)
+
+ for _, pattern := range patterns {
+
+ pattern = strings.TrimSpace(pattern)
+
+ buffer.Html(``)
+ buffer.Html(`| `).Text(pattern).Html(` | `)
+
+ for _, pipeline := range pipelines {
+ decision := qmc.tableResolver.Resolve(pipeline, pattern)
+ buffer.Html(``)
+ if decision != nil {
+ buffer.Text(decision.String())
+ } else {
+ buffer.Text("n/a")
+ }
+ buffer.Html(` | `)
+ }
+ buffer.Html(`
`)
+ }
+
+ buffer.Html(`
`)
+
+ return buffer.Bytes()
+}
diff --git a/quesma/table_resolver/empty.go b/quesma/table_resolver/empty.go
new file mode 100644
index 000000000..2cca8e229
--- /dev/null
+++ b/quesma/table_resolver/empty.go
@@ -0,0 +1,33 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package table_resolver
+
+type EmptyTableResolver struct {
+ Decisions map[string]*Decision
+ RecentDecisionList []PatternDecisions
+ PipelinesList []string
+}
+
+func NewEmptyTableResolver() *EmptyTableResolver {
+ return &EmptyTableResolver{
+ Decisions: make(map[string]*Decision),
+ }
+}
+
+func (r *EmptyTableResolver) Resolve(pipeline string, indexPattern string) *Decision {
+ return r.Decisions[indexPattern]
+}
+
+func (r *EmptyTableResolver) RecentDecisions() []PatternDecisions {
+ return r.RecentDecisionList
+}
+
+func (r *EmptyTableResolver) Pipelines() []string {
+ return r.PipelinesList
+}
+
+func (r *EmptyTableResolver) Start() {
+}
+
+func (r *EmptyTableResolver) Stop() {
+}
diff --git a/quesma/table_resolver/model.go b/quesma/table_resolver/model.go
new file mode 100644
index 000000000..eb691906f
--- /dev/null
+++ b/quesma/table_resolver/model.go
@@ -0,0 +1,116 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package table_resolver
+
+import (
+ "fmt"
+ "quesma/logger"
+ "strings"
+)
+
+type Decision struct {
+ // obvious fields
+ IsClosed bool
+ Err error
+ IsEmpty bool
+
+ // which connector to use, and how
+ UseConnectors []ConnectorDecision
+
+ // who made the decision and why
+ Message string
+ ResolverName string
+}
+
+func (d *Decision) String() string {
+
+ var lines []string
+
+ if d.IsClosed {
+ lines = append(lines, "Returns a closed index message.")
+ }
+
+ if d.IsEmpty {
+ lines = append(lines, "Returns an empty result.")
+ }
+
+ if d.Err != nil {
+ lines = append(lines, fmt.Sprintf("Returns error: '%v'.", d.Err))
+ }
+
+ for _, connector := range d.UseConnectors {
+ lines = append(lines, connector.Message())
+ }
+
+ lines = append(lines, fmt.Sprintf("%s (%s).", d.Message, d.ResolverName))
+
+ return strings.Join(lines, " ")
+}
+
+type ConnectorDecision interface {
+ Message() string
+}
+
+type ConnectorDecisionElastic struct {
+ // TODO instance of elastic connector
+}
+
+func (*ConnectorDecisionElastic) Message() string {
+ return "Pass to Elasticsearch."
+}
+
+type ConnectorDecisionClickhouse struct {
+ // TODO instance of clickhouse connector
+
+ ClickhouseTableName string
+ ClickhouseTables []string
+ IsCommonTable bool
+}
+
+func (d *ConnectorDecisionClickhouse) Message() string {
+ var lines []string
+
+ lines = append(lines, "Pass to clickhouse.")
+ if len(d.ClickhouseTableName) > 0 {
+ lines = append(lines, fmt.Sprintf("Table: '%s' .", d.ClickhouseTableName))
+ }
+ if d.IsCommonTable {
+ lines = append(lines, "Common table.")
+ }
+ if len(d.ClickhouseTables) > 0 {
+ lines = append(lines, fmt.Sprintf("Indexes: %v.", d.ClickhouseTables))
+ }
+
+ return strings.Join(lines, " ")
+}
+
+// PatternDecisions is a struct that holds the pattern and the decisions made for that pattern
+type PatternDecisions struct {
+ Pattern string
+ Decisions map[string]*Decision
+}
+
+type TableResolver interface {
+ Start()
+ Stop()
+
+ Resolve(pipeline string, indexPattern string) *Decision
+
+ Pipelines() []string
+ RecentDecisions() []PatternDecisions
+}
+
+// TODO will be removed in the next PR,
+// right now it is used to mark places where we must refactor the code
+func TODO(place string, decision *Decision) {
+ var trace bool
+ if trace {
+ logger.Debug().Msgf("TODO: use table_resolver decision here %s : %v", place, decision.String())
+ }
+}
+
+// TODO hardcoded pipeline names
+const (
+ QueryPipeline = "Query"
+ IngestPipeline = "Ingest"
+)
diff --git a/quesma/table_resolver/rules.go b/quesma/table_resolver/rules.go
new file mode 100644
index 000000000..f8a021b05
--- /dev/null
+++ b/quesma/table_resolver/rules.go
@@ -0,0 +1,279 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package table_resolver
+
+import (
+ "fmt"
+ "quesma/common_table"
+ "quesma/elasticsearch"
+ "quesma/end_user_errors"
+ "quesma/quesma/config"
+ "quesma/util"
+)
+
+// TODO these rules may be incorrect and incomplete
+// They will be fixed int the next iteration.
+
+func patternIsNotAllowed(input parsedPattern) *Decision {
+ if !input.isPattern {
+ return nil
+ }
+ return &Decision{
+ Message: "Pattern is not allowed.",
+ Err: fmt.Errorf("pattern is not allowed"),
+ }
+}
+
+func makeIsDisabledInConfig(cfg map[string]config.IndexConfiguration, pipeline string) func(input parsedPattern) *Decision {
+
+ return func(input parsedPattern) *Decision {
+
+ if !input.isPattern {
+ idx, ok := cfg[input.source]
+ if ok {
+ if len(getTargets(idx, pipeline)) == 0 {
+ return &Decision{
+ IsClosed: true,
+ Message: "Index is disabled in config.",
+ }
+ }
+ }
+ }
+
+ return nil
+ }
+}
+
+func resolveInternalElasticName(pattern parsedPattern) *Decision {
+
+ if elasticsearch.IsInternalIndex(pattern.source) {
+ return &Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionElastic{}},
+ Message: "It's kibana internals",
+ }
+ }
+
+ return nil
+}
+
+func makeElasticIsDefault(cfg map[string]config.IndexConfiguration) func(input parsedPattern) *Decision {
+
+ return func(input parsedPattern) *Decision {
+ return &Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionElastic{}},
+ Message: "Elastic is default.",
+ }
+ }
+}
+
+func (r *tableRegistryImpl) singleIndex(indexConfig map[string]config.IndexConfiguration, pipeline string) func(input parsedPattern) *Decision {
+
+ return func(input parsedPattern) *Decision {
+
+ if input.isPattern {
+ return nil
+ }
+
+ if cfg, ok := indexConfig[input.source]; ok {
+ if !cfg.UseCommonTable {
+
+ targets := getTargets(cfg, pipeline)
+
+ switch len(targets) {
+
+ case 0:
+ return &Decision{
+ Message: "Disabled in the config.",
+ IsClosed: true,
+ }
+
+ case 1:
+
+ decision := &Decision{
+ Message: "Enabled in the config. ",
+ }
+
+ var targetDecision ConnectorDecision
+
+ // FIXME this
+ switch targets[0] {
+
+ case "elasticsearch":
+ targetDecision = &ConnectorDecisionElastic{}
+ case "clickhouse":
+ targetDecision = &ConnectorDecisionClickhouse{
+ ClickhouseTableName: input.source,
+ ClickhouseTables: []string{input.source},
+ }
+ default:
+ return &Decision{
+ Message: "Unsupported configuration",
+ Err: end_user_errors.ErrSearchCondition.New(fmt.Errorf("unsupported target: %s", targets[0])),
+ }
+ }
+ decision.UseConnectors = append(decision.UseConnectors, targetDecision)
+
+ return decision
+ case 2:
+
+ // check targets and decide
+ // TODO what about A/B testing ?
+
+ return &Decision{
+ Message: "Enabled in the config. Physical table will be used.",
+
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionClickhouse{
+ ClickhouseTableName: input.source,
+ ClickhouseTables: []string{input.source},
+ }},
+ }
+
+ default:
+ return &Decision{
+ Message: "Unsupported configuration",
+ Err: end_user_errors.ErrSearchCondition.New(fmt.Errorf("too many backend connector")),
+ }
+ }
+ }
+ }
+
+ // TODO autodiscovery ?
+
+ return nil
+ }
+}
+
+func (r *tableRegistryImpl) makeCheckIfPatternMatchesAllConnectors() func(input parsedPattern) *Decision {
+
+ return func(input parsedPattern) *Decision {
+ if input.isPattern {
+
+ matchedElastic := []string{}
+ matchedClickhouse := []string{}
+
+ for _, pattern := range input.parts {
+
+ for indexName := range r.elasticIndexes {
+ if util.IndexPatternMatches(pattern, indexName) {
+ matchedElastic = append(matchedElastic, indexName)
+ }
+ }
+
+ for tableName := range r.clickhouseIndexes {
+ if util.IndexPatternMatches(pattern, tableName) {
+ matchedClickhouse = append(matchedClickhouse, tableName)
+ }
+ }
+ }
+
+ nElastic := len(matchedElastic)
+ nClickhouse := len(matchedClickhouse)
+
+ switch {
+
+ case nElastic > 0 && nClickhouse > 0:
+ return &Decision{
+ Err: end_user_errors.ErrSearchCondition.New(fmt.Errorf("index pattern [%s] resolved to both elasticsearch indices: [%s] and clickhouse tables: [%s]", input.parts, matchedElastic, matchedClickhouse)),
+ Message: "Both Elastic and Clickhouse matched.",
+ }
+
+ case nElastic > 0 && nClickhouse == 0:
+
+ return &Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionElastic{}},
+ Message: "Only Elastic matched.",
+ }
+
+ case nElastic == 0 && nClickhouse > 0:
+ // it will be resolved by sth else later
+ return nil
+
+ case nElastic == 0 && nClickhouse == 0:
+
+ // TODO we should return emtpy result here
+ // or pass to another tableResolver
+ return &Decision{
+ IsEmpty: true,
+ Message: "No indexes matched. Checked both connectors.",
+ }
+ }
+ }
+
+ return nil
+ }
+
+}
+
+func (r *tableRegistryImpl) makeCommonTableResolver(cfg map[string]config.IndexConfiguration) func(input parsedPattern) *Decision {
+
+ return func(input parsedPattern) *Decision {
+
+ if input.isPattern {
+
+ // TODO at this point we shouldn't have elastic indexes?
+ for _, pattern := range input.parts {
+ for indexName := range r.elasticIndexes {
+ if util.IndexPatternMatches(pattern, indexName) {
+
+ // TODO what about config ?
+ // TODO ?
+ return &Decision{
+ Err: end_user_errors.ErrSearchCondition.New(fmt.Errorf("index parsedPattern [%s] resolved to elasticsearch indices", input.parts)),
+ Message: "We're not supporting common tables for Elastic.",
+ }
+ }
+ }
+ }
+
+ matchedIndexes := []string{}
+
+ for _, pattern := range input.parts {
+ for indexName, index := range r.clickhouseIndexes {
+
+ // TODO what about config ?
+ // what if index uses common table but is't
+ if util.IndexPatternMatches(pattern, indexName) && index.isVirtual {
+ matchedIndexes = append(matchedIndexes, indexName)
+ }
+ }
+ }
+
+ if len(matchedIndexes) == 0 {
+ return &Decision{
+ IsEmpty: true,
+ Message: "No indexes found.",
+ }
+ }
+
+ // HERE
+ return &Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionClickhouse{
+ IsCommonTable: true,
+ ClickhouseTableName: common_table.TableName,
+ ClickhouseTables: matchedIndexes,
+ }},
+ Message: "Common table will be used. Querying multiple indexes.",
+ }
+ }
+
+ if input.source == common_table.TableName {
+ return &Decision{
+ Err: fmt.Errorf("common table is not allowed to be queried directly"),
+ Message: "It's internal table. Not allowed to be queried directly.",
+ }
+ }
+
+ if idxConfig, ok := cfg[input.source]; ok && idxConfig.UseCommonTable {
+ return &Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionClickhouse{
+ ClickhouseTableName: common_table.TableName,
+ ClickhouseTables: []string{input.source},
+ IsCommonTable: true,
+ }},
+ Message: "Common table will be used.",
+ }
+ }
+
+ return nil
+ }
+}
diff --git a/quesma/table_resolver/table_resolver.go b/quesma/table_resolver/table_resolver.go
new file mode 100644
index 000000000..c5a092e9e
--- /dev/null
+++ b/quesma/table_resolver/table_resolver.go
@@ -0,0 +1,318 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package table_resolver
+
+import (
+ "context"
+ "fmt"
+ "quesma/clickhouse"
+ "quesma/elasticsearch"
+ "quesma/logger"
+ "quesma/quesma/config"
+ "quesma/quesma/recovery"
+ "sort"
+ "strings"
+ "sync"
+ "time"
+)
+
+type tableResolver interface {
+ resolve(indexPattern string) *Decision
+}
+
+// parsedPattern stores the parsed index pattern
+type parsedPattern struct {
+ // source
+ source string
+
+ // parsed data
+ isPattern bool
+ parts []string
+}
+
+type basicResolver struct {
+ name string
+ resolver func(pattern parsedPattern) *Decision
+}
+
+type compoundResolver struct {
+ decisionLadder []basicResolver
+}
+
+func (ir *compoundResolver) resolve(indexName string) *Decision {
+ patterns := strings.Split(indexName, ",")
+
+ input := parsedPattern{
+ source: indexName,
+ isPattern: len(patterns) > 1 || strings.Contains(indexName, "*"),
+ parts: patterns,
+ }
+
+ for _, resolver := range ir.decisionLadder {
+ decision := resolver.resolver(input)
+
+ if decision != nil {
+ decision.ResolverName = resolver.name
+ return decision
+ }
+ }
+ return &Decision{
+ Message: "Could not resolve pattern. This is a bug.",
+ Err: fmt.Errorf("could not resolve index"), // TODO better error
+ }
+}
+
+// HACK: we should have separate config for each pipeline
+// now we have a single config for both, but with different fields
+func getTargets(indexConf config.IndexConfiguration, pipeline string) []string {
+ switch pipeline {
+ case IngestPipeline:
+ return indexConf.IngestTarget
+ case QueryPipeline:
+ return indexConf.QueryTarget
+ default:
+ return []string{}
+ }
+}
+
+// table represents a table or an index discovered in the connector (clickhouse or elastic or ...)
+type table struct {
+ name string
+ isVirtual bool
+}
+
+type pipelineResolver struct {
+ pipelineName string
+
+ resolver tableResolver
+ recentDecisions map[string]*Decision
+}
+
+type tableRegistryImpl struct {
+ m sync.Mutex
+ ctx context.Context
+ cancel context.CancelFunc
+
+ tableDiscovery clickhouse.TableDiscovery
+ elasticIndexResolver elasticsearch.IndexResolver
+
+ elasticIndexes map[string]table
+ clickhouseIndexes map[string]table
+
+ pipelineResolvers map[string]*pipelineResolver
+}
+
+func (r *tableRegistryImpl) Resolve(pipeline string, indexPattern string) *Decision {
+ r.m.Lock()
+ defer r.m.Unlock()
+
+ res, exists := r.pipelineResolvers[pipeline]
+ if !exists {
+ // proper error handling
+ return nil
+ }
+
+ if decision, ok := res.recentDecisions[indexPattern]; ok {
+ return decision
+ }
+
+ decision := res.resolver.resolve(indexPattern)
+ res.recentDecisions[indexPattern] = decision
+
+ logger.Debug().Msgf("Decision for pipeline '%s', pattern '%s': %s", pipeline, indexPattern, decision.String())
+
+ return decision
+}
+
+func (r *tableRegistryImpl) updateIndexes() {
+ r.m.Lock()
+ defer r.m.Unlock()
+
+ defer func() {
+ for _, res := range r.pipelineResolvers {
+ res.recentDecisions = make(map[string]*Decision)
+ }
+ }()
+
+ logger.Info().Msgf("Index registry updating state.")
+
+ // TODO how to interact with the table discovery ?
+ // right now we enforce the reload of the table definitions
+ // schema registry is doing the same
+ // we should inject list of tables into the resolver
+ r.tableDiscovery.ReloadTableDefinitions()
+
+ tableMap := r.tableDiscovery.TableDefinitions()
+ clickhouseIndexes := make(map[string]table)
+
+ tableMap.Range(func(name string, tableDef *clickhouse.Table) bool {
+ clickhouseIndexes[name] = table{
+ name: name,
+ isVirtual: tableDef.VirtualTable,
+ }
+ return true
+ })
+
+ r.clickhouseIndexes = clickhouseIndexes
+ logger.Info().Msgf("Clickhouse tables updated: %v", clickhouseIndexes)
+
+ elasticIndexes := make(map[string]table)
+ sources, ok, err := r.elasticIndexResolver.Resolve("*")
+ if err != nil {
+ logger.Error().Msgf("Could not resolve indexes from Elastic: %v", err)
+ return
+ }
+ if !ok {
+ logger.Error().Msg("Could not resolve indexes from Elastic")
+ return
+ }
+
+ for _, index := range sources.Indices {
+ elasticIndexes[index.Name] = table{
+ name: index.Name,
+ }
+ }
+
+ logger.Info().Msgf("Elastic tables updated: %v", elasticIndexes)
+ r.elasticIndexes = elasticIndexes
+}
+
+func (r *tableRegistryImpl) updateState() {
+ r.updateIndexes()
+}
+
+func (r *tableRegistryImpl) Stop() {
+ r.cancel()
+ logger.Info().Msg("Table resolver stopped.")
+}
+
+func (r *tableRegistryImpl) Start() {
+ go func() {
+ defer recovery.LogPanic()
+ logger.Info().Msg("Table resolve started.")
+
+ for {
+ select {
+ case <-r.ctx.Done():
+ return
+ case <-time.After(1 * time.Minute):
+ r.updateState()
+ }
+ }
+ }()
+}
+
+func (r *tableRegistryImpl) RecentDecisions() []PatternDecisions {
+ r.m.Lock()
+ defer r.m.Unlock()
+
+ var patternsMap = make(map[string]bool)
+
+ for _, res := range r.pipelineResolvers {
+ for p := range res.recentDecisions {
+ patternsMap[p] = true
+ }
+ }
+
+ var patterns []string
+ for p := range patternsMap {
+ patterns = append(patterns, p)
+ }
+
+ sort.Strings(patterns)
+
+ var res []PatternDecisions
+ for _, p := range patterns {
+
+ pd := PatternDecisions{
+ Pattern: p,
+ Decisions: make(map[string]*Decision),
+ }
+ for _, resolver := range r.pipelineResolvers {
+ if decision, ok := resolver.recentDecisions[p]; ok {
+ pd.Decisions[resolver.pipelineName] = decision
+ }
+ }
+ res = append(res, pd)
+ }
+
+ return res
+}
+
+func (r *tableRegistryImpl) Pipelines() []string {
+
+ r.m.Lock()
+ defer r.m.Unlock()
+
+ var res []string
+
+ for name := range r.pipelineResolvers {
+ res = append(res, name)
+ }
+ sort.Strings(res)
+
+ return res
+}
+
+func NewTableResolver(quesmaConf config.QuesmaConfiguration, discovery clickhouse.TableDiscovery, elasticResolver elasticsearch.IndexResolver) TableResolver {
+ ctx, cancel := context.WithCancel(context.Background())
+
+ indexConf := quesmaConf.IndexConfig
+
+ res := &tableRegistryImpl{
+ ctx: ctx,
+ cancel: cancel,
+
+ tableDiscovery: discovery,
+ elasticIndexResolver: elasticResolver,
+ pipelineResolvers: make(map[string]*pipelineResolver),
+ }
+
+ // TODO Here we should read the config and create resolver for each pipeline defined.
+ // TODO We should use the pipeline name as a key in the map.
+
+ ingestResolver := &pipelineResolver{
+ pipelineName: IngestPipeline,
+
+ resolver: &compoundResolver{
+ decisionLadder: []basicResolver{
+ {"patternIsNotAllowed", patternIsNotAllowed},
+ {"kibanaInternal", resolveInternalElasticName},
+ {"disabled", makeIsDisabledInConfig(indexConf, QueryPipeline)},
+
+ {"singleIndex", res.singleIndex(indexConf, IngestPipeline)},
+ {"commonTable", res.makeCommonTableResolver(indexConf)},
+
+ {"elasticAsDefault", makeElasticIsDefault(indexConf)},
+ },
+ },
+ recentDecisions: make(map[string]*Decision),
+ }
+
+ res.pipelineResolvers[IngestPipeline] = ingestResolver
+
+ queryResolver := &pipelineResolver{
+ pipelineName: QueryPipeline,
+
+ resolver: &compoundResolver{
+ decisionLadder: []basicResolver{
+ // checking if we can handle the parsedPattern
+ {"kibanaInternal", resolveInternalElasticName},
+ {"searchAcrossConnectors", res.makeCheckIfPatternMatchesAllConnectors()},
+ {"disabled", makeIsDisabledInConfig(indexConf, QueryPipeline)},
+
+ {"singleIndex", res.singleIndex(indexConf, QueryPipeline)},
+ {"commonTable", res.makeCommonTableResolver(indexConf)},
+
+ // default action
+ {"elasticAsDefault", makeElasticIsDefault(indexConf)},
+ },
+ },
+ recentDecisions: make(map[string]*Decision),
+ }
+
+ res.pipelineResolvers[QueryPipeline] = queryResolver
+ // update the state ASAP
+ res.updateState()
+ return res
+}
diff --git a/quesma/table_resolver/table_resolver_test.go b/quesma/table_resolver/table_resolver_test.go
new file mode 100644
index 000000000..66635ffd1
--- /dev/null
+++ b/quesma/table_resolver/table_resolver_test.go
@@ -0,0 +1,277 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package table_resolver
+
+import (
+ "fmt"
+ "github.com/k0kubun/pp"
+ "github.com/stretchr/testify/assert"
+ "quesma/clickhouse"
+ "quesma/common_table"
+ "quesma/elasticsearch"
+ "quesma/end_user_errors"
+ "quesma/quesma/config"
+ "reflect"
+ "strings"
+ "testing"
+)
+
+func TestTableResolver(t *testing.T) {
+
+ indexConf := map[string]config.IndexConfiguration{
+ "index1": {
+ QueryTarget: []string{"clickhouse"},
+ IngestTarget: []string{"clickhouse"},
+ },
+ "index2": {
+ UseCommonTable: true,
+ QueryTarget: []string{"clickhouse"},
+ IngestTarget: []string{"clickhouse"},
+ },
+ "index3": {
+ QueryTarget: []string{"elasticsearch"},
+ IngestTarget: []string{"elasticsearch"},
+ },
+ "closed": {
+ QueryTarget: []string{},
+ IngestTarget: []string{},
+ },
+ }
+
+ cfg := config.QuesmaConfiguration{IndexConfig: indexConf}
+
+ tests := []struct {
+ name string
+ pipeline string
+ pattern string
+ elasticIndexes []string
+ clickhouseIndexes []string
+ virtualTables []string
+ expected Decision
+ }{
+ {
+ name: "elastic fallback",
+ pipeline: IngestPipeline,
+ pattern: "some-index",
+ expected: Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionElastic{}},
+ },
+ },
+ {
+ name: "all",
+ pipeline: QueryPipeline,
+ pattern: "*",
+ expected: Decision{
+ IsEmpty: true,
+ },
+ },
+ {
+ name: "empty *",
+ pipeline: QueryPipeline,
+ pattern: "*",
+ clickhouseIndexes: []string{"index1", "index2"},
+ expected: Decision{
+ IsEmpty: true,
+ },
+ },
+ {
+ name: "query all, indices in both connectors",
+ pipeline: QueryPipeline,
+ pattern: "*",
+ clickhouseIndexes: []string{"index1", "index2"},
+ elasticIndexes: []string{"index3"},
+ expected: Decision{
+ Err: end_user_errors.ErrSearchCondition.New(fmt.Errorf("")),
+ },
+ },
+ {
+ name: "ingest with a pattern",
+ pipeline: IngestPipeline,
+ pattern: "*",
+ clickhouseIndexes: []string{"index1", "index2"},
+ elasticIndexes: []string{"index3"},
+ expected: Decision{
+ Err: fmt.Errorf("pattern is not allowed"),
+ },
+ },
+ {
+ name: "query closed index",
+ pipeline: QueryPipeline,
+ pattern: "closed",
+ clickhouseIndexes: []string{"closed"},
+ expected: Decision{
+ IsClosed: true,
+ },
+ },
+ {
+ name: "ingest closed index",
+ pipeline: QueryPipeline,
+ pattern: "closed",
+ clickhouseIndexes: []string{"closed"},
+ expected: Decision{
+ IsClosed: true,
+ },
+ },
+ {
+ name: "ingest to index1",
+ pipeline: IngestPipeline,
+ pattern: "index1",
+ clickhouseIndexes: []string{"index1"},
+ expected: Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionClickhouse{
+ ClickhouseTableName: "index1",
+ ClickhouseTables: []string{"index1"}},
+ },
+ },
+ },
+ {
+ name: "query from index1",
+ pipeline: QueryPipeline,
+ pattern: "index1",
+ clickhouseIndexes: []string{"index1"},
+ expected: Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionClickhouse{
+ ClickhouseTableName: "index1",
+ ClickhouseTables: []string{"index1"}},
+ },
+ },
+ },
+ {
+ name: "ingest to index2",
+ pipeline: IngestPipeline,
+ pattern: "index2",
+ clickhouseIndexes: []string{"index2"},
+ expected: Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionClickhouse{
+ ClickhouseTableName: common_table.TableName,
+ ClickhouseTables: []string{"index2"},
+ IsCommonTable: true,
+ }},
+ },
+ },
+ {
+ name: "query from index2",
+ pipeline: QueryPipeline,
+ pattern: "index2",
+ clickhouseIndexes: []string{"index2"},
+ expected: Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionClickhouse{
+ ClickhouseTableName: common_table.TableName,
+ ClickhouseTables: []string{"index2"},
+ IsCommonTable: true,
+ }},
+ },
+ },
+ {
+ name: "ingest to index3",
+ pipeline: IngestPipeline,
+ pattern: "index3",
+ elasticIndexes: []string{"index3"},
+ expected: Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionElastic{}},
+ },
+ },
+ {
+ name: "query from index3",
+ pipeline: QueryPipeline,
+ pattern: "index3",
+ elasticIndexes: []string{"index3"},
+ expected: Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionElastic{}},
+ },
+ },
+ {
+ name: "query pattern",
+ pipeline: QueryPipeline,
+ pattern: "index*",
+ virtualTables: []string{"index2"},
+ expected: Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionClickhouse{
+ ClickhouseTableName: common_table.TableName,
+ ClickhouseTables: []string{"index2"},
+ IsCommonTable: true,
+ }},
+ },
+ },
+ {
+ name: "query kibana internals",
+ pipeline: QueryPipeline,
+ pattern: ".kibana",
+ expected: Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionElastic{}},
+ },
+ },
+ {
+ name: "ingest kibana internals",
+ pipeline: IngestPipeline,
+ pattern: ".kibana",
+ expected: Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionElastic{}},
+ },
+ },
+ {
+ name: "ingest not configured index",
+ pipeline: IngestPipeline,
+ pattern: "not-configured",
+ expected: Decision{
+ UseConnectors: []ConnectorDecision{&ConnectorDecisionElastic{}},
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+
+ tableDiscovery := clickhouse.NewEmptyTableDiscovery()
+
+ for _, index := range tt.clickhouseIndexes {
+ tableDiscovery.TableMap.Store(index, &clickhouse.Table{
+ Name: index,
+ })
+ }
+
+ for _, index := range tt.virtualTables {
+ tableDiscovery.TableMap.Store(index, &clickhouse.Table{
+ Name: index,
+ VirtualTable: true,
+ })
+ }
+
+ elasticResolver := elasticsearch.NewEmptyIndexResolver()
+
+ sources := elasticsearch.Sources{
+ Indices: make([]elasticsearch.Index, 0),
+ }
+
+ for _, index := range tt.elasticIndexes {
+ sources.Indices = append(sources.Indices, elasticsearch.Index{
+ Name: index,
+ })
+ }
+ elasticResolver.Indexes["*"] = sources
+
+ resolver := NewTableResolver(cfg, tableDiscovery, elasticResolver)
+
+ decision := resolver.Resolve(tt.pipeline, tt.pattern)
+
+ assert.NotNil(t, decision)
+ if tt.expected.Err != nil {
+ if !strings.Contains(decision.Err.Error(), tt.expected.Err.Error()) {
+ t.Errorf("Error is not an instance of the expected error: got %v, expected %v", decision.Err, tt.expected.Err)
+ }
+ } else {
+ assert.Nil(t, decision.Err)
+ }
+ assert.Equal(t, tt.expected.IsClosed, decision.IsClosed, "expected %v, got %v", tt.expected.IsClosed, decision.IsClosed)
+ assert.Equal(t, tt.expected.IsEmpty, decision.IsEmpty, "expected %v, got %v", tt.expected.IsEmpty, decision.IsEmpty)
+
+ if !reflect.DeepEqual(tt.expected.UseConnectors, decision.UseConnectors) {
+ pp.Println(tt.expected)
+ pp.Println(decision)
+ t.Errorf("UseConnectors didn't match, expected %v, got %v", tt.expected.UseConnectors, decision.UseConnectors)
+ }
+
+ })
+ }
+
+}