WIP

Author: nablaone

platform/config/config.go

@@ -57,6 +57,8 @@ type QuesmaConfiguration struct {
 	DefaultIngestOptimizers     map[string]OptimizerConfiguration
 	DefaultQueryOptimizers      map[string]OptimizerConfiguration
 	MapFieldsDiscoveringEnabled bool
+
+	IndexNameRewriteRules []IndexNameRewriteRule // rules for rewriting index names, e.g. "index_name" -> "index_name_v2"
 }
 
 func NewQuesmaConfigurationIndexConfigOnly(indexConfig map[string]IndexConfiguration) QuesmaConfiguration {

platform/config/config_v2.go

@@ -12,6 +12,7 @@ import (
 	"github.com/rs/zerolog"
 	"log"
 	"reflect"
+	"regexp"
 	"slices"
 	"strings"
 )
@@ -126,8 +127,15 @@ type (
 		IndexConfig    IndicesConfigs `koanf:"indexes"`
 		// DefaultTargetConnectorType is used in V2 code only
 		DefaultTargetConnectorType string //it is not serialized to maintain configuration BWC, so it's basically just populated from '*' config in `config_v2.go`
+
+		IndexNameRewriteRules map[string]IndexNameRewriteRule `koanf:"indexNameRewriteRules"`
 	}
 	IndicesConfigs map[string]IndexConfiguration
+
+	IndexNameRewriteRule struct {
+		From string `koanf:"from"` // pattern to match
+		To   string `koanf:"to"`   // replacement string
+	}
 )
 
 func (p *QuesmaProcessorConfig) IsFieldMapSyntaxEnabled(indexName string) bool {
@@ -415,6 +423,18 @@ func (c *QuesmaNewConfiguration) definedProcessorNames() []string {
 	return names
 }
 
+func (c *QuesmaNewConfiguration) validateRewriteRules(rules map[string]IndexNameRewriteRule) error {
+
+	for name, rule := range rules {
+		_, err := regexp.Compile(rule.From)
+		if err != nil {
+			return fmt.Errorf("index name rewrite rule '%s' has an invalid 'from' regex: %w", name, err)
+		}
+	}
+
+	return nil
+}
+
 func (c *QuesmaNewConfiguration) validateProcessor(p Processor) error {
 	if len(p.Name) == 0 {
 		return fmt.Errorf("processor must have a non-empty name")
@@ -433,12 +453,23 @@ func (c *QuesmaNewConfiguration) validateProcessor(p Processor) error {
 						return fmt.Errorf("configuration of index %s must have at most two targets (query processor)", indexName)
 					}
 				}
+
+				if p.Config.IndexNameRewriteRules != nil || len(p.Config.IndexNameRewriteRules) > 0 {
+					return fmt.Errorf("index name rewrite rules are not supported in query processor configuration, use the ingest processor for this purpose")
+				}
+
 			} else {
 				if _, ok := indexConfig.Target.([]interface{}); ok {
 					if len(indexConfig.Target.([]interface{})) > 2 {
 						return fmt.Errorf("configuration of index %s must have at most two targets (ingest processor)", indexName)
 					}
 				}
+
+				err := c.validateRewriteRules(p.Config.IndexNameRewriteRules)
+				if err != nil {
+					return err
+				}
+
 			}
 			targets, errTarget := c.getTargetsExtendedConfig(indexConfig.Target)
 			if errTarget != nil {

platform/config/config_v2_util.go

@@ -344,6 +344,25 @@ func (c *QuesmaConfiguration) translateAndAddDualPipeline(confNew *QuesmaNewConf
 		c.DefaultIngestOptimizers = nil
 	}
 
+	if ingestProcessor.Config.IndexNameRewriteRules != nil {
+
+		if len(ingestProcessor.Config.IndexNameRewriteRules) > 0 {
+
+			var names []string
+			for name := range ingestProcessor.Config.IndexNameRewriteRules {
+				names = append(names, name)
+			}
+
+			orderedRules := make([]IndexNameRewriteRule, len(names))
+			for _, name := range names {
+				if rule, ok := ingestProcessor.Config.IndexNameRewriteRules[name]; ok {
+					orderedRules = append(orderedRules, rule)
+				}
+			}
+			c.IndexNameRewriteRules = orderedRules
+		}
+	}
+
 	// safe to call per validation earlier
 	if targts, ok := ingestProcessor.Config.IndexConfig[DefaultWildcardIndexName].Target.([]interface{}); ok {
 		conn := confNew.GetBackendConnectorByName(targts[0].(string))

platform/functionality/bulk/bulk.go

@@ -75,7 +75,7 @@ func Write(ctx context.Context, defaultIndex *string, bulk types.NDJSON, ip *ing
 
 	// The returned results should be in the same order as the input request, however splitting the bulk might change the order.
 	// Therefore, each BulkRequestEntry has a corresponding pointer to the result entry, allowing us to freely split and reshuffle the bulk.
-	results, clickhouseBulkEntries, elasticRequestBody, elasticBulkEntries, err := SplitBulk(ctx, defaultIndex, bulk, maxBulkSize, tableResolver)
+	results, clickhouseBulkEntries, elasticRequestBody, elasticBulkEntries, err := SplitBulk(ctx, defaultIndex, bulk, maxBulkSize, tableResolver, ip.GetIndexNameRewriter())
 	if err != nil {
 		return []BulkItem{}, err
 	}
@@ -118,7 +118,7 @@ func Write(ctx context.Context, defaultIndex *string, bulk types.NDJSON, ip *ing
 	return nonEmptyResults, nil
 }
 
-func SplitBulk(ctx context.Context, defaultIndex *string, bulk types.NDJSON, maxBulkSize int, tableResolver table_resolver.TableResolver) ([]BulkItem, map[string][]BulkRequestEntry, []byte, []BulkRequestEntry, error) {
+func SplitBulk(ctx context.Context, defaultIndex *string, bulk types.NDJSON, maxBulkSize int, tableResolver table_resolver.TableResolver, rewriter ingest.IndexNameRewriter) ([]BulkItem, map[string][]BulkRequestEntry, []byte, []BulkRequestEntry, error) {
 	results := make([]BulkItem, maxBulkSize)
 
 	clickhouseBulkEntries := make(map[string][]BulkRequestEntry, maxBulkSize)
@@ -129,6 +129,8 @@ func SplitBulk(ctx context.Context, defaultIndex *string, bulk types.NDJSON, max
 		index := op.GetIndex()
 		operation := op.GetOperation()
 
+		index = rewriter.RewriteIndex(index)
+
 		entryWithResponse := BulkRequestEntry{
 			operation: operation,
 			index:     index,

platform/ingest/index_name_rewriter.go

@@ -0,0 +1,74 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package ingest
+
+import (
+	"fmt"
+	"github.com/QuesmaOrg/quesma/platform/config"
+	"github.com/QuesmaOrg/quesma/platform/logger"
+	"regexp"
+)
+
+type IndexNameRewriter interface {
+	RewriteIndex(indexName string) string
+}
+
+type nopIndexNameRewriter struct {
+}
+
+func (n *nopIndexNameRewriter) RewriteIndex(indexName string) string {
+	// no-op rewriter, returns the index name as is
+	return indexName
+}
+
+type rewriteRule struct {
+	Pattern     *regexp.Regexp
+	Replacement string
+}
+
+func (r *rewriteRule) String() string {
+	return fmt.Sprintf("RewriteRule `%s` -> `%s`", r.Pattern.String(), r.Replacement)
+}
+
+type indexNameRegexpRewriter struct {
+	rules []rewriteRule
+}
+
+func NewIndexNameRewriter(cfg *config.QuesmaConfiguration) IndexNameRewriter {
+
+	if len(cfg.IndexNameRewriteRules) == 0 {
+		logger.Info().Msgf("No index name rewrite rules configured, using no-op rewriter")
+		// if no rewrite rules are configured, return a no-op rewriter
+		return &nopIndexNameRewriter{}
+	}
+
+	var rules []rewriteRule
+
+	for _, rule := range cfg.IndexNameRewriteRules {
+		if rule.From == "" || rule.To == "" {
+			continue // skip invalid rules
+		}
+		pattern, err := regexp.Compile(rule.From)
+		if err != nil {
+			logger.Error().Msgf("Unable to compile regexp for index name rewrite: %s", rule.From)
+			continue // skip invalid regex patterns
+		}
+		r := rewriteRule{
+			Pattern:     pattern,
+			Replacement: rule.To,
+		}
+		rules = append(rules, r)
+		logger.Info().Msgf("Added index name rewrite rule: %s", r.String())
+	}
+
+	return &indexNameRegexpRewriter{rules: rules}
+}
+
+func (i *indexNameRegexpRewriter) RewriteIndex(indexName string) string {
+
+	rewritten := indexName
+	for _, rule := range i.rules {
+		rewritten = rule.Pattern.ReplaceAllString(rewritten, rule.Replacement)
+	}
+	return rewritten
+}

platform/ingest/processor.go

@@ -69,6 +69,8 @@ type (
 		virtualTableStorage       persistence.JSONDatabase
 		tableResolver             table_resolver.TableResolver
 
+		indexNameRewriter IndexNameRewriter
+
 		errorLogCounter atomic.Int64
 	}
 	TableMap  = util.SyncMap[string, *chLib.Table]
@@ -1062,9 +1064,14 @@ func (ip *IngestProcessor) Ping() error {
 	return ip.chDb.Ping()
 }
 
+func (ip *IngestProcessor) GetIndexNameRewriter() IndexNameRewriter {
+	return ip.indexNameRewriter
+}
+
 func NewIngestProcessor(cfg *config.QuesmaConfiguration, chDb quesma_api.BackendConnector, phoneHomeClient diag.PhoneHomeClient, loader chLib.TableDiscovery, schemaRegistry schema.Registry, virtualTableStorage persistence.JSONDatabase, tableResolver table_resolver.TableResolver) *IngestProcessor {
 	ctx, cancel := context.WithCancel(context.Background())
-	return &IngestProcessor{ctx: ctx, cancel: cancel, chDb: chDb, tableDiscovery: loader, cfg: cfg, phoneHomeClient: phoneHomeClient, schemaRegistry: schemaRegistry, virtualTableStorage: virtualTableStorage, tableResolver: tableResolver}
+	indexRewriter := NewIndexNameRewriter(cfg)
+	return &IngestProcessor{ctx: ctx, cancel: cancel, chDb: chDb, tableDiscovery: loader, cfg: cfg, phoneHomeClient: phoneHomeClient, schemaRegistry: schemaRegistry, virtualTableStorage: virtualTableStorage, tableResolver: tableResolver, indexNameRewriter: indexRewriter}
 }
 
 func NewOnlySchemaFieldsCHConfig(clusterName string) *chLib.ChTableConfig {