Implement Elasticsearch functions in GO (#15160)

### What problem does this PR solve? Implement Elasticsearch functions in GO (except for Search) ### Type of change - [x] Refactoring
2026-07-16 12:47:19 +08:00 · 2026-05-25 19:15:07 +08:00
parent 7d200d5bd7
commit af85aa9c7b
21 changed files with 2418 additions and 758 deletions
--- a/internal/engine/elasticsearch/chunk.go
+++ b/internal/engine/elasticsearch/chunk.go
--- a/internal/engine/elasticsearch/client.go
+++ b/internal/engine/elasticsearch/client.go
@@ -17,11 +17,16 @@
 package elasticsearch

 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
+	"os"
+	"path/filepath"
 	"ragflow/internal/server"
+	"ragflow/internal/utility"
 	"time"

 	"github.com/elastic/go-elasticsearch/v8"
@@ -81,6 +86,16 @@ func NewEngine(cfg interface{}) (*elasticsearchEngine, error) {
 		config: esConfig,
 	}

+	// Create two index templates for different index types
+	// Template for chunk indices (ragflow_*) - priority 1
+	if err := engine.CreateIndexTemplate(context.Background(), "ragflow_mapping", "ragflow_*", "mapping.json", 1); err != nil {
+		return nil, fmt.Errorf("failed to create chunk index template: %w", err)
+	}
+	// Template for doc_meta indices (ragflow_doc_meta_*) - priority 2 (higher than ragflow_*)
+	if err := engine.CreateIndexTemplate(context.Background(), "ragflow_doc_meta_mapping", "ragflow_doc_meta_*", "doc_meta_es_mapping.json", 2); err != nil {
+		return nil, fmt.Errorf("failed to create doc_meta index template: %w", err)
+	}
+
 	return engine, nil
 }

@@ -109,6 +124,82 @@ func (e *elasticsearchEngine) Close() error {
 	return nil
 }

+// CreateIndexTemplate creates an index template with the specified mapping
+// The template will be automatically applied to any new index matching the pattern
+func (e *elasticsearchEngine) CreateIndexTemplate(ctx context.Context, templateName, indexPattern, mappingFileName string, priority ...int) error {
+	if templateName == "" || indexPattern == "" {
+		return fmt.Errorf("template name and index pattern cannot be empty")
+	}
+
+	p := 1
+	if len(priority) > 0 {
+		p = priority[0]
+	}
+
+	if mappingFileName == "" {
+		mappingFileName = "mapping.json"
+	}
+
+	// Read mapping from file
+	mappingPath := filepath.Join(utility.GetProjectRoot(), "conf", mappingFileName)
+	data, err := os.ReadFile(mappingPath)
+	if err != nil {
+		return fmt.Errorf("failed to read mapping file: %w", err)
+	}
+
+	var mapping map[string]interface{}
+	if err := json.Unmarshal(data, &mapping); err != nil {
+		return fmt.Errorf("failed to parse mapping file: %w", err)
+	}
+
+	// Separate settings and mappings from the mapping file
+	templateSettings := mapping["settings"]
+	templateMappings := mapping["mappings"]
+
+	// Build template body with proper structure
+	templateBody := map[string]interface{}{
+		"index_patterns": []string{indexPattern},
+		"priority": p, // Configurable priority to override existing templates
+		"template": map[string]interface{}{
+			"settings": templateSettings,
+			"mappings": templateMappings,
+		},
+	}
+
+	templateBytes, err := json.Marshal(templateBody)
+	if err != nil {
+		return fmt.Errorf("failed to marshal template: %w", err)
+	}
+
+	// Create or update template
+	req := esapi.IndicesPutIndexTemplateRequest{
+		Name: templateName,
+		Body: bytes.NewReader(templateBytes),
+	}
+
+	res, err := req.Do(ctx, e.client)
+	if err != nil {
+		return fmt.Errorf("failed to create index template: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.IsError() {
+		bodyBytes, _ := io.ReadAll(res.Body)
+		return fmt.Errorf("failed to create index template: %s, body: %s", res.Status(), string(bodyBytes))
+	}
+
+	var result map[string]interface{}
+	if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
+		return fmt.Errorf("failed to parse response: %w", err)
+	}
+
+	if acknowledged, ok := result["acknowledged"].(bool); !ok || !acknowledged {
+		return fmt.Errorf("index template creation not acknowledged")
+	}
+
+	return nil
+}
+
 // GetClusterStats gets Elasticsearch cluster statistics
 // Reference: curl -XGET "http://{es_host}/_cluster/stats" -H "kbn-xsrf: reporting"
 func (e *elasticsearchEngine) GetClusterStats() (map[string]interface{}, error) {
--- a/internal/engine/elasticsearch/metadata.go
+++ b/internal/engine/elasticsearch/metadata.go
@@ -21,6 +21,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"strings"

 	"github.com/elastic/go-elasticsearch/v8/esapi"
@@ -32,6 +33,17 @@ import (
 // CreateMetadataStore creates the document metadata index
 func (e *elasticsearchEngine) CreateMetadataStore(ctx context.Context, tenantID string) error {
 	indexName := buildMetadataIndexName(tenantID)
+
+	// Check if index already exists
+	exists, err := e.indexExists(ctx, indexName)
+	if err != nil {
+		return fmt.Errorf("failed to check index existence: %w", err)
+	}
+	if exists {
+		return nil
+	}
+
+	// Index will be created with mapping from index template (ragflow_doc_meta_mapping)
 	req := esapi.IndicesCreateRequest{
 		Index: indexName,
 	}
@@ -41,15 +53,29 @@ func (e *elasticsearchEngine) CreateMetadataStore(ctx context.Context, tenantID
 	}
 	defer res.Body.Close()
 	if res.IsError() {
-		return fmt.Errorf("elasticsearch returned error: %s", res.Status())
+		bodyBytes, _ := io.ReadAll(res.Body)
+		return fmt.Errorf("elasticsearch returned error: %s, body: %s", res.Status(), string(bodyBytes))
 	}
+
+	// Parse response
+	var result map[string]interface{}
+	if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
+		return fmt.Errorf("failed to parse response: %w", err)
+	}
+
+	acknowledged, ok := result["acknowledged"].(bool)
+	if !ok || !acknowledged {
+		return fmt.Errorf("metadata index creation not acknowledged")
+	}
+
 	return nil
 }

 // InsertMetadata inserts documents into tenant's metadata index
+// If a document with the same id and kb_id already exists, it will be updated with the new value
 func (e *elasticsearchEngine) InsertMetadata(ctx context.Context, metadata []map[string]interface{}, tenantID string) ([]string, error) {
 	indexName := buildMetadataIndexName(tenantID)
-	common.Info("Inserting metadata into Elasticsearch index", zap.String("index_name", indexName), zap.String("tenant_id", tenantID), zap.Int("doc_count", len(metadata)))
+	common.Info("ElasticsearchConnection.InsertMetadata called", zap.String("index_name", indexName), zap.String("tenant_id", tenantID), zap.Int("doc_count", len(metadata)))

 	if len(metadata) == 0 {
 		return []string{}, nil
@@ -75,28 +101,34 @@ func (e *elasticsearchEngine) InsertMetadata(ctx context.Context, metadata []map
 	// Build bulk request body
 	var buf bytes.Buffer
 	for _, doc := range metadata {
-		// Action line - index operation
-		action := map[string]interface{}{
+		docIDRaw, hasID := doc["id"]
+		kbIDRaw, hasKBID := doc["kb_id"]
+		docID, idOK := docIDRaw.(string)
+		kbID, kbOK := kbIDRaw.(string)
+		if !hasID || !hasKBID || !idOK || !kbOK || strings.TrimSpace(docID) == "" || strings.TrimSpace(kbID) == "" {
+			common.Warn("Skipping metadata document without id or kb_id")
+			continue
+		}
+
+		// Action line: use json.Marshal to properly escape string values
+		compositeID := fmt.Sprintf("%d:%s|%d:%s", len(docID), docID, len(kbID), kbID)
+		action, err := json.Marshal(map[string]interface{}{
 			"index": map[string]interface{}{
 				"_index": indexName,
+				"_id":    compositeID,
 			},
-		}
-		actionBytes, err := json.Marshal(action)
+		})
 		if err != nil {
 			common.Error("Failed to marshal bulk action", err)
 			return nil, fmt.Errorf("failed to marshal bulk action: %w", err)
 		}
-		buf.Write(actionBytes)
+		buf.Write(action)
 		buf.WriteByte('\n')

-		// Document line - meta_fields is stored as-is (ES can handle nested objects)
-		docBytes, err := json.Marshal(doc)
-		if err != nil {
-			common.Error("Failed to marshal document", err)
-			return nil, fmt.Errorf("failed to marshal document: %w", err)
+		// Document line
+		if err := json.NewEncoder(&buf).Encode(doc); err != nil {
+			return nil, fmt.Errorf("failed to encode document: %w", err)
 		}
-		buf.Write(docBytes)
-		buf.WriteByte('\n')
 	}

 	// Execute bulk request
@@ -113,8 +145,9 @@ func (e *elasticsearchEngine) InsertMetadata(ctx context.Context, metadata []map
 	defer res.Body.Close()

 	if res.IsError() {
-		common.Sugar.Errorw("Elasticsearch bulk request returned error", "status", res.Status())
-		return nil, fmt.Errorf("elasticsearch bulk request returned error: %s", res.Status())
+		bodyBytes, _ := io.ReadAll(res.Body)
+		common.Sugar.Errorw("Elasticsearch bulk request returned error", "status", res.Status(), "body", string(bodyBytes))
+		return nil, fmt.Errorf("elasticsearch bulk request returned error: %s, body: %s", res.Status(), string(bodyBytes))
 	}

 	// Parse bulk response to check for errors
@@ -129,14 +162,15 @@ func (e *elasticsearchEngine) InsertMetadata(ctx context.Context, metadata []map
 		common.Warn("Bulk request had some errors")
 	}

-	common.Info("Successfully inserted metadata into Elasticsearch index", zap.String("index_name", indexName), zap.Int("doc_count", len(metadata)))
+	common.Info("ElasticsearchConnection.InsertMetadata result", zap.String("index_name", indexName), zap.Int("count", len(metadata)))
 	return []string{}, nil
 }

 // UpdateMetadata updates document metadata in tenant's metadata index
+// The metaFields map will fully replace the existing meta_fields
 func (e *elasticsearchEngine) UpdateMetadata(ctx context.Context, docID string, datasetID string, metaFields map[string]interface{}, tenantID string) error {
 	indexName := buildMetadataIndexName(tenantID)
-	common.Info("Updating metadata in Elasticsearch index", zap.String("index_name", indexName), zap.String("docID", docID), zap.String("datasetID", datasetID))
+	common.Info("ElasticsearchConnection.UpdateMetadata called", zap.String("index_name", indexName), zap.String("docID", docID), zap.String("datasetID", datasetID))

 	// Check if index exists
 	exists, err := e.indexExists(ctx, indexName)
@@ -193,14 +227,16 @@ func (e *elasticsearchEngine) UpdateMetadata(ctx context.Context, docID string,
 		return fmt.Errorf("elasticsearch update by query returned error: %s", res.Status())
 	}

-	common.Info("Successfully updated metadata in Elasticsearch index", zap.String("index_name", indexName), zap.String("docID", docID))
+	common.Info("ElasticsearchConnection.UpdateMetadata completes", zap.String("index_name", indexName), zap.String("docID", docID))
 	return nil
 }

 // DeleteMetadata deletes metadata from tenant's metadata index by condition
+// The condition is a map used to build an ES query (e.g., map["kb_id"]="xxx")
+// Returns the number of deleted documents
 func (e *elasticsearchEngine) DeleteMetadata(ctx context.Context, condition map[string]interface{}, tenantID string) (int64, error) {
 	indexName := buildMetadataIndexName(tenantID)
-	common.Info("Deleting metadata from Elasticsearch index", zap.String("index_name", indexName), zap.Any("condition", condition))
+	common.Info("ElasticsearchConnection.DeleteMetadata called", zap.String("index_name", indexName), zap.Any("condition", condition))

 	// Check if index exists
 	exists, err := e.indexExists(ctx, indexName)
@@ -258,10 +294,205 @@ func (e *elasticsearchEngine) DeleteMetadata(ctx context.Context, condition map[
 		deleted = int64(d)
 	}

-	common.Info("Successfully deleted metadata", zap.String("index_name", indexName), zap.Int64("deleted_count", deleted))
+	common.Info("ElasticsearchConnection.DeleteMetadata completes", zap.String("index_name", indexName), zap.Int64("deleted_count", deleted))
 	return deleted, nil
 }

+// DeleteMetadataKeys deletes specific metadata keys from a document's meta_fields.
+// If deleting those keys leaves no metadata entries, the metadata document is removed.
+func (e *elasticsearchEngine) DeleteMetadataKeys(ctx context.Context, docID string, datasetID string, keys []string, tenantID string) error {
+	indexName := buildMetadataIndexName(tenantID)
+	common.Info("ElasticsearchConnection.DeleteMetadataKeys called", zap.String("index_name", indexName), zap.String("docID", docID), zap.Any("keys", keys))
+
+	// Check if index exists
+	exists, err := e.indexExists(ctx, indexName)
+	if err != nil {
+		return fmt.Errorf("failed to check index existence: %w", err)
+	}
+	if !exists {
+		return fmt.Errorf("index '%s' does not exist", indexName)
+	}
+
+	// Build the document ID for query (no escaping needed for ES term queries)
+	docIDTerm := docID
+	datasetIDTerm := datasetID
+
+	// Build query to find the document
+	query := map[string]interface{}{
+		"bool": map[string]interface{}{
+			"must": []map[string]interface{}{
+				{"term": map[string]interface{}{"id": docIDTerm}},
+				{"term": map[string]interface{}{"kb_id": datasetIDTerm}},
+			},
+		},
+	}
+
+	// First, get the current meta_fields to check if it will be empty after deletion
+	getReq := map[string]interface{}{
+		"query": query,
+		"_source": []string{"meta_fields"},
+		"size": 1,
+	}
+
+	getBytes, err := json.Marshal(getReq)
+	if err != nil {
+		return fmt.Errorf("failed to marshal get request: %w", err)
+	}
+
+	// Use esapi.SearchRequest directly
+	getSearchReq := esapi.SearchRequest{
+		Index: []string{indexName},
+		Body:  bytes.NewReader(getBytes),
+	}
+
+	getRes, err := getSearchReq.Do(ctx, e.client)
+	if err != nil {
+		return fmt.Errorf("failed to get current metadata: %w", err)
+	}
+	defer getRes.Body.Close()
+
+	if getRes.IsError() {
+		return fmt.Errorf("elasticsearch get request returned error: %s", getRes.Status())
+	}
+
+	var getResult map[string]interface{}
+	if err := json.NewDecoder(getRes.Body).Decode(&getResult); err != nil {
+		return fmt.Errorf("failed to parse get response: %w", err)
+	}
+
+	hits, ok := getResult["hits"].(map[string]interface{})
+	if !ok {
+		return fmt.Errorf("invalid get response format")
+	}
+	hitsArray, ok := hits["hits"].([]interface{})
+	if !ok || len(hitsArray) == 0 {
+		return fmt.Errorf("document not found: %s", docID)
+	}
+
+	// Check current meta_fields
+	firstHit, ok := hitsArray[0].(map[string]interface{})
+	if !ok {
+		return fmt.Errorf("invalid hit format")
+	}
+	source, ok := firstHit["_source"].(map[string]interface{})
+	if !ok {
+		return fmt.Errorf("invalid source format")
+	}
+	metaFieldsVal, hasMetaFields := source["meta_fields"]
+
+	var currentMetaFields map[string]interface{}
+	if hasMetaFields && metaFieldsVal != nil {
+		switch v := metaFieldsVal.(type) {
+		case map[string]interface{}:
+			currentMetaFields = v
+		case string:
+			if unmarshalErr := json.Unmarshal([]byte(v), &currentMetaFields); unmarshalErr != nil {
+				common.Warn("Failed to parse meta_fields JSON", zap.String("docID", docID), zap.Error(unmarshalErr))
+				currentMetaFields = make(map[string]interface{})
+			}
+		}
+	}
+
+	// If no current meta_fields or already empty, nothing to delete
+	if currentMetaFields == nil || len(currentMetaFields) == 0 {
+		common.Info("No metadata fields to delete from document", zap.String("docID", docID))
+		return nil
+	}
+
+	// Calculate which keys will be removed
+	keysToRemove := make(map[string]bool)
+	for _, k := range keys {
+		keysToRemove[k] = true
+	}
+
+	// Check if any keys actually exist and would be removed
+	hasKeysToRemove := false
+	for k := range currentMetaFields {
+		if keysToRemove[k] {
+			hasKeysToRemove = true
+			break
+		}
+	}
+
+	if !hasKeysToRemove {
+		common.Info("No matching keys to delete from document", zap.String("docID", docID))
+		return nil
+	}
+
+	// Count remaining keys after deletion (keys that are NOT being removed)
+	remainingKeys := 0
+	for k := range currentMetaFields {
+		if !keysToRemove[k] {
+			remainingKeys++
+		}
+	}
+
+	// If no other keys would remain after deletion, delete the document directly
+	if remainingKeys == 0 {
+		common.Info("All metadata keys would be deleted, removing document from index", zap.String("docID", docID))
+
+		// Build condition for deletion using docID and datasetID
+		condition := map[string]interface{}{
+			"id":    docIDTerm,
+			"kb_id": datasetIDTerm,
+		}
+
+		// Use existing DeleteMetadata method which handles the deletion properly
+		_, err := e.DeleteMetadata(ctx, condition, tenantID)
+		if err != nil {
+			return fmt.Errorf("failed to delete document: %w", err)
+		}
+
+		common.Info("Successfully removed document with empty meta_fields", zap.String("docID", docID))
+		return nil
+	}
+
+	// Some keys will remain, so remove only the specified keys
+	keysParam := make([]string, len(keys))
+	for i, k := range keys {
+		keysParam[i] = k
+	}
+
+	// Build update script that removes keys from meta_fields map
+	scriptSource := "for(int i=0;i<params.keys.length;i++){if(ctx._source.meta_fields.containsKey(params.keys[i])){ctx._source.meta_fields.remove(params.keys[i])}}"
+
+	updateReq := map[string]interface{}{
+		"query": query,
+		"script": map[string]interface{}{
+			"source": scriptSource,
+			"params": map[string]interface{}{
+				"keys": keysParam,
+			},
+		},
+	}
+
+	updateBytes, err := json.Marshal(updateReq)
+	if err != nil {
+		return fmt.Errorf("failed to marshal update request: %w", err)
+	}
+
+	req := esapi.UpdateByQueryRequest{
+		Index: []string{indexName},
+		Body:  bytes.NewReader(updateBytes),
+	}
+
+	res, err := req.Do(ctx, e.client)
+	if err != nil {
+		common.Error("Failed to execute update by query", err)
+		return fmt.Errorf("failed to execute update by query: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.IsError() {
+		common.Sugar.Errorw("Elasticsearch update by query returned error", "status", res.Status())
+		return fmt.Errorf("elasticsearch update by query returned error: %s", res.Status())
+	}
+
+	common.Info("ElasticsearchConnection.DeleteMetadataKeys completes", zap.String("index_name", indexName), zap.String("docID", docID))
+
+	return nil
+}
+
 // DropMetadataStore drops a metadata index from Elasticsearch
 func (e *elasticsearchEngine) DropMetadataStore(ctx context.Context, tenantID string) error {
 	indexName := buildMetadataIndexName(tenantID)
--- a/internal/engine/engine.go
+++ b/internal/engine/engine.go
@@ -47,6 +47,7 @@ type DocEngine interface {
 	InsertMetadata(ctx context.Context, metadata []map[string]interface{}, tenantID string) ([]string, error)
 	UpdateMetadata(ctx context.Context, docID string, datasetID string, metaFields map[string]interface{}, tenantID string) error
 	DeleteMetadata(ctx context.Context, condition map[string]interface{}, tenantID string) (int64, error)
+	DeleteMetadataKeys(ctx context.Context, docID string, datasetID string, keys []string, tenantID string) error
 	DropMetadataStore(ctx context.Context, tenantID string) error
 	MetadataStoreExists(ctx context.Context, tenantID string) (bool, error)

--- a/internal/engine/infinity/metadata.go
+++ b/internal/engine/infinity/metadata.go
@@ -228,7 +228,7 @@ func (e *infinityEngine) UpdateMetadata(ctx context.Context, docID string, datas
 			if metaFieldsData, exists := qr.Data["meta_fields"]; exists && len(metaFieldsData) > 0 {
 				existingMetaFieldsVal := metaFieldsData[0]

-				// Parse existing meta_fields if it's a string
+				// Parse existing meta_fields if it's a string or []uint8
 				var existingMetaFields map[string]interface{}
 				if existingMetaFieldsVal != nil {
 					switch v := existingMetaFieldsVal.(type) {
@@ -237,6 +237,16 @@ func (e *infinityEngine) UpdateMetadata(ctx context.Context, docID string, datas
 							common.Warn(fmt.Sprintf("Failed to parse existing meta_fields: %v", err))
 							existingMetaFields = make(map[string]interface{})
 						}
+					case []uint8:
+						// Handle raw bytes from Infinity - Infinity prefixes JSON with 4 bytes (likely length), skip them
+						decoded := v
+						if len(decoded) > 4 {
+							decoded = decoded[4:]
+						}
+						if err := json.Unmarshal(decoded, &existingMetaFields); err != nil {
+							common.Warn(fmt.Sprintf("Failed to parse existing meta_fields from []uint8: %v", err))
+							existingMetaFields = make(map[string]interface{})
+						}
 					case map[string]interface{}:
 						existingMetaFields = v
 					}
@@ -285,6 +295,7 @@ func (e *infinityEngine) UpdateMetadata(ctx context.Context, docID string, datas
 }

 // DeleteMetadata deletes metadata from tenant's metadata table by condition
+// Returns the number of deleted documents.
 func (e *infinityEngine) DeleteMetadata(ctx context.Context, condition map[string]interface{}, tenantID string) (int64, error) {
 	tableName := buildMetadataTableName(tenantID)

@@ -341,6 +352,143 @@ func (e *infinityEngine) DeleteMetadata(ctx context.Context, condition map[strin
 	return delResp.DeletedRows, nil
 }

+// DeleteMetadataKeys deletes specific metadata keys from a document's meta_fields.
+// If deleting those keys leaves no metadata entries, the metadata row is removed.
+func (e *infinityEngine) DeleteMetadataKeys(ctx context.Context, docID string, datasetID string, keys []string, tenantID string) error {
+	tableName := buildMetadataTableName(tenantID)
+	common.Info("InfinityConnection.DeleteMetadataKeys called", zap.String("tableName", tableName), zap.String("docID", docID), zap.Any("keys", keys))
+
+	db, err := e.client.conn.GetDatabase(e.client.dbName)
+	if err != nil {
+		return fmt.Errorf("failed to get database: %w", err)
+	}
+
+	table, err := db.GetTable(tableName)
+	if err != nil {
+		return fmt.Errorf("failed to get metadata table %s: %w", tableName, err)
+	}
+
+	// Build filter to find the document
+	escapedDocID := strings.ReplaceAll(docID, "'", "''")
+	escapedDatasetID := strings.ReplaceAll(datasetID, "'", "''")
+	filter := fmt.Sprintf("id = '%s' AND kb_id = '%s'", escapedDocID, escapedDatasetID)
+
+	// Query existing metadata to get current meta_fields
+	queryTable := table.Output([]string{"id", "kb_id", "meta_fields"}).Filter(filter).Limit(1).Offset(0)
+	result, err := queryTable.ToResult()
+	if err != nil {
+		return fmt.Errorf("failed to query existing metadata: %w", err)
+	}
+
+	qr, ok := result.(*infinity.QueryResult)
+	if !ok || qr == nil || len(qr.Data["id"]) == 0 {
+		return fmt.Errorf("document not found: %s", docID)
+	}
+
+	// Get existing meta_fields
+	var existingMetaFields map[string]interface{}
+	if metaFieldsData, exists := qr.Data["meta_fields"]; exists && len(metaFieldsData) > 0 {
+		if metaFieldsData[0] != nil {
+			switch v := metaFieldsData[0].(type) {
+			case string:
+				if err := json.Unmarshal([]byte(v), &existingMetaFields); err != nil {
+					common.Warn("Failed to parse meta_fields JSON", zap.String("docID", docID), zap.Error(err))
+					existingMetaFields = make(map[string]interface{})
+				}
+			case []uint8:
+				// Handle raw bytes from Infinity - Infinity prefixes JSON with 4 bytes (likely length), skip them
+				decoded := v
+				if len(decoded) > 4 {
+					decoded = decoded[4:]
+				}
+				if err := json.Unmarshal(decoded, &existingMetaFields); err != nil {
+					common.Warn("Failed to parse meta_fields JSON from []uint8", zap.String("docID", docID), zap.String("err", err.Error()))
+					existingMetaFields = make(map[string]interface{})
+				}
+			case map[string]interface{}:
+				existingMetaFields = v
+			default:
+				common.Debug("meta_fields unexpected type", zap.String("type", fmt.Sprintf("%T", metaFieldsData[0])), zap.Any("value", metaFieldsData[0]))
+			}
+		}
+	} else {
+		common.Debug("meta_fields not found in qr.Data or empty", zap.Any("exists", exists))
+	}
+
+	if existingMetaFields == nil {
+		existingMetaFields = make(map[string]interface{})
+	}
+
+	// Build set of keys to remove
+	keysToRemove := make(map[string]bool)
+	for _, k := range keys {
+		keysToRemove[k] = true
+	}
+
+	// Check if any keys actually exist and would be removed
+	hasKeysToRemove := false
+	for k := range existingMetaFields {
+		if keysToRemove[k] {
+			hasKeysToRemove = true
+			break
+		}
+	}
+
+	if !hasKeysToRemove {
+		common.Info(
+			"No matching keys to delete from document",
+			zap.String("docID", docID),
+			zap.Int("existingMetaFieldCount", len(existingMetaFields)),
+			zap.Int("keysCount", len(keys)),
+		)
+		return nil
+	}
+
+	// Count remaining keys after deletion (keys that are NOT being removed)
+	remainingKeys := 0
+	for k := range existingMetaFields {
+		if !keysToRemove[k] {
+			remainingKeys++
+		}
+	}
+
+	// If no other keys would remain after deletion, delete the document directly
+	if remainingKeys == 0 {
+		common.Info("All metadata keys would be deleted, removing document from index", zap.String("docID", docID))
+
+		// Use existing DeleteMetadata method which handles the deletion properly
+		condition := map[string]interface{}{
+			"id":    docID,
+			"kb_id": datasetID,
+		}
+		_, err := e.DeleteMetadata(ctx, condition, tenantID)
+		if err != nil {
+			return fmt.Errorf("failed to delete document: %w", err)
+		}
+
+		common.Info("Successfully removed document with empty meta_fields", zap.String("docID", docID))
+		return nil
+	}
+
+	// Some keys will remain, so remove only the specified keys
+	for _, key := range keys {
+		delete(existingMetaFields, key)
+	}
+
+	// Update with the modified metadata
+	updatedFields := map[string]interface{}{
+		"meta_fields": utility.ConvertMapToJSONString(existingMetaFields),
+	}
+
+	_, err = table.Update(filter, updatedFields)
+	if err != nil {
+		return fmt.Errorf("failed to delete metadata keys: %w", err)
+	}
+
+	common.Info("InfinityConnection.DeleteMetadataKeys completed", zap.String("tableName", tableName), zap.String("docID", docID))
+	return nil
+}
+
 // DropMetadataStore drops a metadata table from Infinity
 func (e *infinityEngine) DropMetadataStore(ctx context.Context, tenantID string) error {
 	tableName := buildMetadataTableName(tenantID)