Improved Elasticsearch index resolution and bulk operations

Author: mertbilgic

connector.go

@@ -108,15 +108,8 @@ func (c *connector) listener(ctx *replication.ListenerContext) {
 
 	fullTableName := c.getFullTableName(msg.TableNamespace, msg.TableName)
 
-	tableName, exists := c.cfg.Elasticsearch.TableIndexMapping[fullTableName]
-	if !exists {
-		parentTableName := c.getParentTableName(fullTableName, msg.TableNamespace, msg.TableName)
-		if parentTableName != "" {
-			tableName = c.cfg.Elasticsearch.TableIndexMapping[parentTableName]
-		}
-	}
-
-	if !c.isTableInMapping(tableName) {
+	indexName := c.resolveTableToIndexName(fullTableName, msg.TableNamespace, msg.TableName)
+	if indexName == "" {
 		if err := ctx.Ack(); err != nil {
 			logger.Error("ack", "error", err)
 		}
@@ -136,29 +129,38 @@ func (c *connector) listener(ctx *replication.ListenerContext) {
 		chunks := slices.ChunkWithSize[elasticsearch.Action](actions, batchSizeLimit)
 		lastChunkIndex := len(chunks) - 1
 		for idx, chunk := range chunks {
-			c.bulk.AddActions(ctx, msg.EventTime, chunk, fullTableName, idx == lastChunkIndex)
+			c.bulk.AddActions(ctx, msg.EventTime, chunk, indexName, idx == lastChunkIndex)
 		}
 	} else {
-		c.bulk.AddActions(ctx, msg.EventTime, actions, tableName, true)
+		c.bulk.AddActions(ctx, msg.EventTime, actions, indexName, true)
 	}
 }
 
-func (c *connector) isTableInMapping(tableName string) bool {
-	if len(c.cfg.Elasticsearch.TableIndexMapping) == 0 {
-		return true
+func (c *connector) resolveTableToIndexName(fullTableName, tableNamespace, tableName string) string {
+	tableIndexMapping := c.cfg.Elasticsearch.TableIndexMapping
+	if len(tableIndexMapping) == 0 {
+		return fullTableName
 	}
 
-	if _, exists := c.cfg.Elasticsearch.TableIndexMapping[tableName]; exists {
-		return true
+	if indexName, exists := tableIndexMapping[fullTableName]; exists {
+		return indexName
 	}
 
-	t, ok := timescaledb.HyperTables.Load(tableName)
-	if ok {
-		_, exists := c.cfg.Elasticsearch.TableIndexMapping[t.(string)]
-		return exists
+	parentTableName := c.getParentTableName(fullTableName, tableNamespace, tableName)
+	if parentTableName != "" {
+		if indexName, exists := tableIndexMapping[parentTableName]; exists {
+			return indexName
+		}
 	}
 
-	return tableName != ""
+	if t, ok := timescaledb.HyperTables.Load(fullTableName); ok {
+		parentName := t.(string)
+		if indexName, exists := tableIndexMapping[parentName]; exists {
+			return indexName
+		}
+	}
+
+	return ""
 }
 
 func (c *connector) getParentTableName(fullTableName, tableNamespace, tableName string) string {

elasticsearch/bulk/bulk.go

@@ -32,7 +32,7 @@ type Indexer interface {
 		ctx *replication.ListenerContext,
 		eventTime time.Time,
 		actions []elasticsearch2.Action,
-		tableName string,
+		indexName string,
 		isLastChunk bool,
 	)
 	GetMetric() Metric
@@ -115,12 +115,12 @@ func (b *Bulk) AddActions(
 	ctx *replication.ListenerContext,
 	eventTime time.Time,
 	actions []elasticsearch2.Action,
-	tableName string,
+	indexName string,
 	isLastChunk bool,
 ) {
 	b.flushLock.Lock()
 	for i, action := range actions {
-		indexName := b.getIndexName(tableName, action.IndexName)
+		indexName := b.getIndexName(indexName, action.IndexName)
 		actions[i].IndexName = indexName
 		value := getEsActionJSON(
 			action.ID,
@@ -352,16 +352,16 @@ func joinErrors(body map[string]any) (map[string]string, error) {
 	return ivd, fmt.Errorf(sb.String())
 }
 
-func (b *Bulk) getIndexName(tableName, actionIndexName string) string {
+func (b *Bulk) getIndexName(indexName, actionIndexName string) string {
 	if actionIndexName != "" {
 		return actionIndexName
 	}
 
-	if tableName == "" {
-		panic(fmt.Sprintf("there is no index mapping for table: %s on your configuration", tableName))
+	if indexName == "" {
+		panic(fmt.Sprintf("there is no index mapping for table: %s on your configuration", indexName))
 	}
 
-	return tableName
+	return indexName
 }
 
 func (b *Bulk) handleResponse(batchActions []*elasticsearch2.Action, errs map[string]string) {