Refine handling of POST /api/v1/datasets/search in GO (#15583)

### What problem does this PR solve? Refine handling of POST /api/v1/datasets/search in GO ### Type of change - [x] Refactoring
2026-08-07 08:01:13 +08:00 · 2026-06-08 11:49:37 +08:00
parent 074c331cdf
commit c960dc2a4c
70 changed files with 8580 additions and 1915 deletions
--- a/internal/engine/infinity/chunk.go
+++ b/internal/engine/infinity/chunk.go
@@ -30,12 +30,14 @@ import (
 	"sort"
 	"strconv"
 	"strings"
-	"unicode"

 	infinity "github.com/infiniflow/infinity-go-sdk"
 	"go.uber.org/zap"
 )

+// ChinesePunctRegex splits on comma, semicolon, Chinese punctuations, and newlines
+var ChinesePunctRegex = regexp.MustCompile(`[,，;；、\r\n]+`)
+
 // CreateChunkStore creates a chunk table in Infinity
 // baseName is the table name prefix (e.g., "ragflow_<tenant_id>")
 // The full table name is built as "{baseName}_{datasetID}"
@@ -47,7 +49,7 @@ func (e *infinityEngine) CreateChunkStore(ctx context.Context, baseName, dataset
 	var tableName string
 	var mappingFile string

-    tableName = buildChunkTableName(baseName, datasetID)
+	tableName = buildChunkTableName(baseName, datasetID)
 	if datasetID == "skill" {
 		mappingFile = "skill_infinity_mapping.json"
 		common.Info("Creating skill index table", zap.String("tableName", tableName), zap.String("mappingFile", mappingFile))
@@ -574,32 +576,7 @@ func (e *infinityEngine) DeleteChunks(ctx context.Context, condition map[string]
 // It supports three matching types: MatchTextExpr (full-text), MatchDenseExpr (vector), and FusionExpr (combined).
 // If no match expressions are provided, Search relies solely on filter (e.g., doc_id, available_int) to find results.
 func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (*types.SearchResult, error) {
-	common.Debug("Search in Infinity started", zap.Any("indexNames", req.IndexNames))
-	if common.IsDebugEnabled() {
-		// Format match expressions for logging
-		var matchExprsStr string
-		for i, expr := range req.MatchExprs {
-			switch e := expr.(type) {
-			case *types.MatchTextExpr:
-				matchExprsStr += fmt.Sprintf("    [%d] MatchTextExpr: fields=%v, matchingText=%s, topN=%d, extraOptions=%v\n", i, e.Fields, e.MatchingText, e.TopN, e.ExtraOptions)
-			case *types.MatchDenseExpr:
-				matchExprsStr += fmt.Sprintf("    [%d] MatchDenseExpr: vectorColumn=%s, vectorSize=%d, topN=%d, extraOptions=%v\n", i, e.VectorColumnName, len(e.EmbeddingData), e.TopN, e.ExtraOptions)
-			case *types.FusionExpr:
-				matchExprsStr += fmt.Sprintf("    [%d] FusionExpr: method=%s, topN=%d, fusionParams=%v\n", i, e.Method, e.TopN, e.FusionParams)
-			default:
-				matchExprsStr += fmt.Sprintf("    [%d] unknown type\n", i)
-			}
-		}
-		common.Debug(fmt.Sprintf("Search request:\n"+
-			"    indexNames=%v\n"+
-			"    KbIDs=%v\n"+
-			"    offset=%d, limit=%d\n"+
-			"    SelectFields=%v\n"+
-			"    Filter=%v\n"+
-			"    MatchExprs:\n%s    orderBy=%v\n"+
-			"    RankFeature=%v",
-			req.IndexNames, req.KbIDs, req.Offset, req.Limit, req.SelectFields, req.Filter, matchExprsStr, req.OrderBy, req.RankFeature))
-	}
+	types.LogSearchRequest("Infinity", req)

 	if len(req.IndexNames) == 0 {
 		return nil, fmt.Errorf("index names cannot be empty")
@@ -621,13 +598,8 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 		return nil, fmt.Errorf("failed to get database: %w", err)
 	}

-	isMetadataTable := false
 	isSkillIndex := false
 	for _, idx := range req.IndexNames {
-		if strings.HasPrefix(idx, "ragflow_doc_meta_") {
-			isMetadataTable = true
-			break
-		}
 		if strings.HasPrefix(idx, "skill_") {
 			isSkillIndex = true
 			break
@@ -635,9 +607,7 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 	}

 	var outputColumns []string
-	if isMetadataTable {
-		outputColumns = []string{"id", "kb_id", "meta_fields"}
-	} else if isSkillIndex {
+	if isSkillIndex {
 		outputColumns = []string{
 			"skill_id", "space_id", "folder_id", "name", "tags", "description", "content",
 			"version", "status", "create_time", "update_time",
@@ -716,23 +686,31 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 		outputColumns = append(outputColumns, "row_id()")
 	}

+	// Strip score pseudo-columns when there's no match expression — Infinity
+	// rejects SCORE()/SCORE_FACTORS() without MATCH TEXT/TENSOR/Fusion with
+	// "InfinityException(3013)". This protects callers (e.g. the no-match
+	// fallback in retrieval.go) that reuse a SelectFields list containing
+	// "_score" across both matched and unmatched queries.
+	if !hasTextMatch && !hasVectorMatch {
+		filtered := outputColumns[:0]
+		for _, c := range outputColumns {
+			switch c {
+			case "_score", "SCORE", "score()", "similarity()":
+				continue
+			}
+			filtered = append(filtered, c)
+		}
+		outputColumns = filtered
+	}
+
 	outputColumns = convertSelectFields(outputColumns, isSkillIndex)
 	if hasVectorMatch && matchDense != nil && matchDense.VectorColumnName != "" {
 		outputColumns = append(outputColumns, matchDense.VectorColumnName)
 	}

 	var filterParts []string
-	if isMetadataTable && len(req.KbIDs) > 0 && req.KbIDs[0] != "" {
-		kbIDs := req.KbIDs
-		if len(kbIDs) == 1 {
-			filterParts = append(filterParts, fmt.Sprintf("kb_id = '%s'", kbIDs[0]))
-		} else {
-			kbIDStr := strings.Join(kbIDs, "', '")
-			filterParts = append(filterParts, fmt.Sprintf("kb_id IN ('%s')", kbIDStr))
-		}
-	}

-	if !isMetadataTable && (hasTextMatch || hasVectorMatch) {
+	if hasTextMatch || hasVectorMatch {
 		if req.Filter != nil {
 			if availInt, ok := req.Filter["available_int"]; ok {
 				filterParts = append(filterParts, fmt.Sprintf("available_int=%v", availInt))
@@ -756,13 +734,10 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (

 	// Build filter string from req.Filter
 	if req.Filter != nil {
-		filterCopy := req.Filter
-		if !isMetadataTable {
-			filterCopy = make(map[string]interface{})
-			for k, v := range req.Filter {
-				if k != "kb_id" {
-					filterCopy[k] = v
-				}
+		filterCopy := make(map[string]interface{})
+		for k, v := range req.Filter {
+			if k != "kb_id" {
+				filterCopy[k] = v
 			}
 		}

@@ -807,8 +782,13 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 			}
 		}

-		minMatch := 0.3
-
+		// minMatch comes from matchText.ExtraOptions when set (Python parity).
+		// Mirrors rag/utils/infinity_conn.py which reads
+		// matchExpr.extra_options.get("minimum_should_match", 0.0) — for the
+		// English (non-Chinese) path, the Go Question() builder omits
+		// minimum_should_match, so the default is 0.0 to match Python's
+		// effective 0% threshold for English queries.
+		minMatch := 0.0
 		var questionText string
 		var vectorData []float64
 		textTopN := pageSize
@@ -820,6 +800,19 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 				if oq, ok := matchText.ExtraOptions["original_query"].(string); ok {
 					originalQuery = oq
 				}
+				if v, ok := matchText.ExtraOptions["minimum_should_match"]; ok {
+					switch x := v.(type) {
+					case float64:
+						minMatch = x
+					case int:
+						minMatch = float64(x)
+					case string:
+						s := strings.TrimSuffix(x, "%")
+						if pct, err := strconv.Atoi(s); err == nil {
+							minMatch = float64(pct) / 100
+						}
+					}
+				}
 			}
 		}
 		if matchDense != nil {
@@ -933,13 +926,21 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 					}
 				}

-				if hasTextMatch && fusionExpr == nil {
+				if hasTextMatch {
 					fieldsStr := strings.Join(convertedFields, ",")
 					filterFulltext := fmt.Sprintf("filter_fulltext('%s', '%s')", fieldsStr, questionText)
 					denseFilterStr = fmt.Sprintf("(%s) AND %s", denseFilterStr, filterFulltext)
 				}
+				threshold := "0.0"
+				if matchDense != nil && matchDense.ExtraOptions != nil {
+					if sim, ok := matchDense.ExtraOptions["similarity"].(float64); ok {
+						threshold = fmt.Sprintf("%g", sim)
+					} else if s, ok := matchDense.ExtraOptions["threshold"].(string); ok {
+						threshold = s
+					}
+				}
 				extraOptions := map[string]string{
-					"threshold": utility.FloatToString(0.0),
+					"threshold": threshold,
 					"filter":    denseFilterStr,
 				}

@@ -1031,7 +1032,7 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 			// Skill index uses different schema
 			// so we skip the document-specific field mappings
 			if !isSkillIndex {
-				GetFields(searchChunks, nil)
+				applyFieldMappings(searchChunks)
 			} else {
 				// For skill index, only handle ROW_ID -> row_id() mapping
 				for _, chunk := range searchChunks {
@@ -1229,23 +1230,10 @@ func (e *infinityEngine) GetChunk(ctx context.Context, tableName, chunkID string
 	return chunk, nil
 }

-// GetFields applies field mappings to chunks and returns a dict keyed by chunk ID.
-// Equivalent to Python's get_fields() in infinity_conn.py.
-// When fields is nil/empty, returns all fields from chunks.
-func GetFields(chunks []map[string]interface{}, fields []string) map[string]map[string]interface{} {
-	result := make(map[string]map[string]interface{})
-	if len(chunks) == 0 {
-		return result
-	}
-
-	// If fields is provided, create a set for lookup
-	fieldSet := make(map[string]bool)
-	for _, f := range fields {
-		fieldSet[f] = true
-	}
-
+// applyFieldMappings applies field mappings to chunks (side-effect only).
+// Used by Search() to mutate chunks with derived fields before returning.
+func applyFieldMappings(chunks []map[string]interface{}) {
 	for _, chunk := range chunks {
-		// Apply field mappings
 		// docnm -> docnm_kwd, title_tks, title_sm_tks
 		if val, ok := chunk["docnm"].(string); ok {
 			chunk["docnm_kwd"] = val
@@ -1253,12 +1241,12 @@ func GetFields(chunks []map[string]interface{}, fields []string) map[string]map[
 			chunk["title_sm_tks"] = val
 		}

-		// important_keywords -> important_kwd (split by comma), important_tks
+		// important_keywords -> important_kwd (split by comma/semicolon/Chinese punctuations), important_tks
 		if val, ok := chunk["important_keywords"].(string); ok {
 			if val == "" {
 				chunk["important_kwd"] = []interface{}{}
 			} else {
-				parts := strings.Split(val, ",")
+				parts := ChinesePunctRegex.Split(val, -1)
 				chunk["important_kwd"] = parts
 			}
 			chunk["important_tks"] = val
@@ -1313,7 +1301,7 @@ func GetFields(chunks []map[string]interface{}, fields []string) map[string]map[
 			"important_kwd": true, "question_kwd": true,
 		}
 		arrayFields := []string{
-			"doc_type_kwd", "important_kwd", "important_tks", "question_tks",
+			"important_kwd", "important_tks", "question_tks",
 			"question_kwd", "authors_tks", "authors_sm_tks", "title_tks",
 			"title_sm_tks", "content_ltks", "content_sm_ltks", "tag_kwd",
 		}
@@ -1341,27 +1329,302 @@ func GetFields(chunks []map[string]interface{}, fields []string) map[string]map[
 			chunk["row_id()"] = val
 			delete(chunk, "ROW_ID")
 		}
+	}
+}

-		// Build result map keyed by id
-		if id, ok := chunk["id"].(string); ok {
-			fieldMap := make(map[string]interface{})
-			for field, value := range chunk {
-				if len(fieldSet) == 0 || fieldSet[field] {
-					fieldMap[field] = value
+// GetFields extracts the requested fields from Infinity search results
+func (e *infinityEngine) GetFields(chunks []map[string]interface{}, fields []string) map[string]map[string]interface{} {
+	result := make(map[string]map[string]interface{})
+
+	// Python: if not fields, return {}
+	if len(fields) == 0 {
+		return result
+	}
+
+	if len(chunks) == 0 {
+		return result
+	}
+
+	// Build field set for lookup (Python lines 713-715)
+	fieldsAll := make(map[string]bool)
+	for _, f := range fields {
+		fieldsAll[f] = true
+	}
+	fieldsAll["id"] = true
+
+	// noneColumns is rebuilt per chunk inside the loop below. The
+	// per-chunk "missing → nil" map MUST be fresh for every iteration; if
+	// it's reused, the first chunk that contains a field removes it from
+	// the shared set, and later chunks missing that same field silently
+	// stop getting the nil placeholder, producing inconsistent shapes
+	// per document.
+
+	// Check if important_kwd is needed (for empty_count handling)
+	needImportantKwdEmptyCount := fieldsAll["important_kwd"]
+
+	for _, chunk := range chunks {
+		// Build column map for case-insensitive lookup (Python line 747)
+		columnMap := make(map[string]string)
+		for k := range chunk {
+			columnMap[strings.ToLower(k)] = k
+		}
+
+		// Apply field mappings first (to get derived fields)
+		// docnm -> docnm_kwd, title_tks, title_sm_tks (Python lines 716-719)
+		// Note: Python checks "docnm" in res.columns regardless of whether fields were requested
+		if val, ok := chunk["docnm"].(string); ok {
+			if fieldsAll["docnm_kwd"] {
+				chunk["docnm_kwd"] = val
+			}
+			if fieldsAll["title_tks"] {
+				chunk["title_tks"] = val
+			}
+			if fieldsAll["title_sm_tks"] {
+				chunk["title_sm_tks"] = val
+			}
+		}
+
+		// important_keywords -> important_kwd (split by comma), important_tks (Python lines 720-732)
+		// Python: v.split(",") if v else [] — empty string yields empty list
+		if fieldsAll["important_kwd"] || fieldsAll["important_tks"] {
+			if val, ok := chunk["important_keywords"].(string); ok && val != "" {
+				if fieldsAll["important_kwd"] {
+					if needImportantKwdEmptyCount {
+						// Check for important_kwd_empty_count (Python lines 722-728)
+						if emptyCountVal, hasEmptyCount := chunk["important_kwd_empty_count"]; hasEmptyCount {
+							tokens := strings.Split(val, ",")
+							var emptyCount int
+							switch v := emptyCountVal.(type) {
+							case float64:
+								emptyCount = int(v)
+							case int:
+								emptyCount = v
+							case string:
+								emptyCount, _ = strconv.Atoi(v)
+							}
+							kwdList := make([]interface{}, 0, len(tokens)+emptyCount)
+							for _, t := range tokens {
+								kwdList = append(kwdList, t)
+							}
+							for i := 0; i < emptyCount; i++ {
+								kwdList = append(kwdList, "")
+							}
+							chunk["important_kwd"] = kwdList
+						} else {
+							parts := strings.Split(val, ",")
+							kwdList := make([]interface{}, len(parts))
+							for i, p := range parts {
+								kwdList[i] = p
+							}
+							chunk["important_kwd"] = kwdList
+						}
+					} else {
+						parts := strings.Split(val, ",")
+						kwdList := make([]interface{}, len(parts))
+						for i, p := range parts {
+							kwdList[i] = p
+						}
+						chunk["important_kwd"] = kwdList
+					}
+				}
+				if fieldsAll["important_tks"] {
+					chunk["important_tks"] = val
+				}
+			} else {
+				if fieldsAll["important_kwd"] {
+					chunk["important_kwd"] = []interface{}{}
+				}
+				if fieldsAll["important_tks"] {
+					chunk["important_tks"] = []interface{}{}
 				}
 			}
-			result[id] = fieldMap
+		}
+
+		// questions -> question_kwd (split by newline), question_tks (Python lines 733-737)
+		// Python: v.splitlines() — empty string yields empty list
+		if fieldsAll["question_kwd"] || fieldsAll["question_tks"] {
+			if val, ok := chunk["questions"].(string); ok && val != "" {
+				if fieldsAll["question_kwd"] {
+					parts := strings.Split(val, "\n")
+					qList := make([]interface{}, len(parts))
+					for i, p := range parts {
+						qList[i] = p
+					}
+					chunk["question_kwd"] = qList
+				}
+				if fieldsAll["question_tks"] {
+					chunk["question_tks"] = val
+				}
+			} else {
+				if fieldsAll["question_kwd"] {
+					chunk["question_kwd"] = []interface{}{}
+				}
+				if fieldsAll["question_tks"] {
+					chunk["question_tks"] = []interface{}{}
+				}
+			}
+		}
+
+		// content -> content_with_weight, content_ltks, content_sm_ltks (Python lines 738-741)
+		if fieldsAll["content_with_weight"] || fieldsAll["content_ltks"] || fieldsAll["content_sm_ltks"] {
+			if val, ok := chunk["content"].(string); ok {
+				if fieldsAll["content_with_weight"] {
+					chunk["content_with_weight"] = val
+				}
+				if fieldsAll["content_ltks"] {
+					chunk["content_ltks"] = val
+				}
+				if fieldsAll["content_sm_ltks"] {
+					chunk["content_sm_ltks"] = val
+				}
+			}
+		}
+
+		// authors -> authors_tks, authors_sm_tks (Python lines 742-745)
+		if fieldsAll["authors_tks"] || fieldsAll["authors_sm_tks"] {
+			if val, ok := chunk["authors"].(string); ok {
+				if fieldsAll["authors_tks"] {
+					chunk["authors_tks"] = val
+				}
+				if fieldsAll["authors_sm_tks"] {
+					chunk["authors_sm_tks"] = val
+				}
+			}
+		}
+
+		// Post-process fields matching Python lines 758-780
+		// This single loop processes all column transformations in Python order
+		kwdNoSplit := map[string]bool{
+			"knowledge_graph_kwd": true, "docnm_kwd": true,
+			"important_kwd": true, "question_kwd": true,
+		}
+		for field, val := range chunk {
+			fieldLower := strings.ToLower(field)
+
+			// field_keyword: split by "###" (Python lines 760-761)
+			needsSplit := false
+			if fieldLower == "source_id" {
+				needsSplit = true
+			} else if strings.HasSuffix(fieldLower, "_kwd") && !kwdNoSplit[fieldLower] {
+				needsSplit = true
+			}
+			if needsSplit {
+				if strVal, ok := val.(string); ok && strings.Contains(strVal, "###") {
+					parts := strings.Split(strVal, "###")
+					var filtered []interface{}
+					for _, p := range parts {
+						if p != "" {
+							filtered = append(filtered, p)
+						}
+					}
+					chunk[field] = filtered
+				}
+				continue
+			}
+
+			// _feas: JSON parse (Python lines 762-763)
+			if strings.HasSuffix(fieldLower, "_feas") {
+				if strVal, ok := val.(string); ok && strVal != "" {
+					var parsed interface{}
+					if err := json.Unmarshal([]byte(strVal), &parsed); err == nil {
+						chunk[field] = parsed
+					}
+				} else {
+					chunk[field] = map[string]interface{}{}
+				}
+				continue
+			}
+
+			// chunk_data: JSON parse (Python lines 764-766)
+			if fieldLower == "chunk_data" {
+				if strVal, ok := val.(string); ok && strVal != "" {
+					var parsed interface{}
+					if err := json.Unmarshal([]byte(strVal), &parsed); err == nil {
+						chunk[field] = parsed
+					}
+				} else if val == nil {
+					// Keep nil
+				}
+				continue
+			}
+
+			// position_int: hex decode with grouping by 5 (Python lines 767-776)
+			if fieldLower == "position_int" && fieldsAll[fieldLower] {
+				// If already converted to slice by applyFieldMappings, skip
+				if _, isSlice := val.([]interface{}); isSlice {
+					continue
+				}
+				// applyFieldMappings returns [][]int, check that too
+				if _, isIntSlice := val.([][]int); isIntSlice {
+					continue
+				}
+				if strVal, ok := val.(string); ok && strVal != "" {
+					chunk[field] = utility.ConvertHexToPositionIntArray(strVal)
+				} else {
+					chunk[field] = []interface{}{}
+				}
+				continue
+			}
+
+			// page_num_int, top_int: hex decode (Python lines 777-778)
+			if (fieldLower == "page_num_int" || fieldLower == "top_int") && fieldsAll[fieldLower] {
+				// If already converted to slice by applyFieldMappings, skip
+				if _, isSlice := val.([]interface{}); isSlice {
+					continue
+				}
+				// applyFieldMappings returns []int, check that too
+				if _, isIntSlice := val.([]int); isIntSlice {
+					continue
+				}
+				if strVal, ok := val.(string); ok && strVal != "" {
+					chunk[field] = utility.ConvertHexToIntArray(strVal)
+				} else {
+					chunk[field] = []interface{}{}
+				}
+				continue
+			}
+		}
+
+		// Handle row_id mapping (Python lines 748-750)
+		if fieldsAll["row_id()"] {
+			if lowerKey, ok := columnMap["row_id"]; ok {
+				chunk["row_id()"] = chunk[lowerKey]
+			}
+		}
+
+		// Delete base columns after mapping (Python lines 781-783)
+		for _, col := range []string{"docnm", "important_keywords", "questions", "content", "authors"} {
+			delete(chunk, col)
+		}
+
+		// Build result map keyed by id
+		if idVal, ok := chunk["id"].(string); ok {
+			fieldMap := make(map[string]interface{})
+			// Rebuild noneColumns for this chunk so that fields missing
+			// from THIS chunk get a nil placeholder. Reusing a set across
+			// chunks would let the first chunk's contents permanently
+			// remove keys, leaving later chunks with inconsistent shapes.
+			noneColumns := make(map[string]bool, len(fieldsAll))
+			for f := range fieldsAll {
+				noneColumns[strings.ToLower(f)] = true
+			}
+			for field, value := range chunk {
+				if fieldsAll[field] {
+					fieldMap[field] = value
+					delete(noneColumns, strings.ToLower(field))
+				}
+			}
+			// Set none_columns to None (Python lines 784-785)
+			for col := range noneColumns {
+				fieldMap[col] = nil
+			}
+			result[idVal] = fieldMap
 		}
 	}

 	return result
 }

-// GetFields is a method wrapper for infinityEngine to satisfy DocEngine interface
-func (e *infinityEngine) GetFields(chunks []map[string]interface{}, fields []string) map[string]map[string]interface{} {
-	return GetFields(chunks, fields)
-}
-
 // GetAggregation aggregates chunk values by field name.
 // Input: [{"docnm_kwd": "docA"}, {"docnm_kwd": "docA"}, {"docnm_kwd": "docB"}]
 //
@@ -1466,12 +1729,8 @@ func (e *infinityEngine) GetAggregation(chunks []map[string]interface{}, fieldNa
 	return result
 }

-// GetDocIDs extracts document IDs from search results.
-// Extracts "id" field from each chunk and returns as a list.
-func (e *infinityEngine) GetDocIDs(chunks []map[string]interface{}) []string {
-	if len(chunks) == 0 {
-		return nil
-	}
+// GetChunkIDs extracts chunk IDs from Infinity search results.
+func (e *infinityEngine) GetChunkIDs(chunks []map[string]interface{}) []string {
 	ids := make([]string, 0, len(chunks))
 	for _, chunk := range chunks {
 		if id, ok := chunk["id"].(string); ok {
@@ -1489,95 +1748,81 @@ func (e *infinityEngine) GetHighlight(chunks []map[string]interface{}, keywords
 		return result
 	}

-	// Check if field exists
-	hasField := false
-	for _, chunk := range chunks {
-		if _, ok := chunk[fieldName]; ok {
-			hasField = true
-			break
-		}
-	}
-	if !hasField {
-		// Try alternative field names
-		if fieldName == "content_with_weight" {
-			if _, ok := chunks[0]["content"]; ok {
-				fieldName = "content"
-				hasField = true
-			}
-		}
-	}
-	if !hasField {
-		return result
-	}
-
-	emTag := regexp.MustCompile(`<em>[^<>]+</em>`)
-
-	for _, chunk := range chunks {
-		id := ""
-		if idVal, ok := chunk["id"].(string); ok {
-			id = idVal
-		}
-
-		txt, ok := chunk[fieldName].(string)
-		if !ok || txt == "" {
-			continue
-		}
-
-		// Check if already highlighted
-		if emTag.MatchString(txt) {
-			result[id] = txt
-			continue
-		}
-
-		// Replace newlines with spaces
-		txt = regexp.MustCompile(`[\r\n]`).ReplaceAllString(txt, " ")
-
-		// Split by sentence delimiters
-		delimiters := regexp.MustCompile(`[.?!;\n]`)
-		segments := delimiters.Split(txt, -1)
-
-		var highlightedSegments []string
-		for _, segment := range segments {
-			// Check if segment is English or contains keywords
-			englishCount := 0
-			totalCount := 0
-			for _, r := range segment {
-				if unicode.IsLetter(r) {
-					totalCount++
-					if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') {
-						englishCount++
-					}
-				}
-			}
-			isEnglish := totalCount > 0 && float64(englishCount)/float64(totalCount) > 0.5
-			segmentToCheck := segment
-			if isEnglish {
-				// For English: match whole words with boundaries
-				for _, kw := range keywords {
-					re := regexp.MustCompile(`(^|[ .?/'\"\(\)!,:;-])` + regexp.QuoteMeta(kw) + `([ .?/'\"\(\)!,:;-]|$)`)
-					segmentToCheck = re.ReplaceAllString(segmentToCheck, "$1<em>"+kw+"</em>$2")
-				}
-			} else {
-				// For non-English: simple substring match
-				for _, kw := range keywords {
-					segmentToCheck = strings.ReplaceAll(segmentToCheck, kw, "<em>"+kw+"</em>")
-				}
-			}
-			if strings.Contains(segmentToCheck, "<em>") {
-				highlightedSegments = append(highlightedSegments, segmentToCheck)
-			}
-		}
-
-		if len(highlightedSegments) > 0 {
-			result[id] = strings.Join(highlightedSegments, "...")
-		}
-	}
-
+	// For Infinity, scores are already returned in search results (_score column)
+	// So GetScores just extracts scores from chunks, mimicking Python's approach
 	return result
 }

+// KNNScores for Infinity - since Infinity normalizes scores during fusion,
+// we just need to return a result structure that GetScores can parse.
+// This matches Python's approach where Infinity doesn't use the two-pass KNN.
+func (e *infinityEngine) KNNScores(ctx context.Context, chunks []map[string]interface{}, queryVector []float64, topK int) (map[string]interface{}, error) {
+	if len(chunks) == 0 {
+		return nil, nil
+	}
+
+	// Build a result structure that GetScores can parse
+	// For Infinity, scores are already in _score field from the first search
+	result := make(map[string]interface{})
+	hitList := make([]interface{}, 0, len(chunks))
+	for _, chunk := range chunks {
+		if id, ok := chunk["id"].(string); ok {
+			hit := map[string]interface{}{
+				"_id":    id,
+				"_score": chunk["_score"],
+			}
+			hitList = append(hitList, hit)
+		}
+	}
+	result["hits"] = map[string]interface{}{
+		"hits": hitList,
+	}
+	return result, nil
+}
+
+// GetScores extracts similarity scores from KNN search result.
+// For Infinity, it parses the result from KNNScores and extracts _score values.
+func (e *infinityEngine) GetScores(knnResult map[string]interface{}) map[string]float64 {
+	scores := make(map[string]float64)
+	hits, ok := knnResult["hits"].(map[string]interface{})
+	if !ok {
+		return scores
+	}
+	hitList, ok := hits["hits"].([]interface{})
+	if !ok {
+		return scores
+	}
+	for _, h := range hitList {
+		hit, ok := h.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		docID, ok := hit["_id"].(string)
+		if !ok {
+			continue
+		}
+		scoreVal := hit["_score"]
+		if scoreVal == nil {
+			scores[docID] = 0.0
+			continue
+		}
+		score, ok := scoreVal.(float64)
+		if !ok {
+			scores[docID] = 0.0
+			continue
+		}
+		scores[docID] = score
+	}
+	return scores
+}
+
 // convertSelectFields converts field names to Infinity format
 // isSkillIndex indicates if this is a skill index (uses skill_id instead of id)
+//
+// Does NOT mutate the input slice — callers (e.g. retrieval.go) reuse the same
+// SelectFields list both for Search() and GetFields(); mutating it would
+// replace logical names like "content_with_weight" with their Infinity column
+// names ("content"), breaking GetFields's field-presence checks.
 func convertSelectFields(output []string, isSkillIndex ...bool) []string {
 	fieldMapping := map[string]string{
 		"docnm_kwd":           "docnm",
@@ -1599,20 +1844,24 @@ func convertSelectFields(output []string, isSkillIndex ...bool) []string {
 		skillIndex = isSkillIndex[0]
 	}

+	// Copy + map without mutating the caller's slice.
+	mapped := make([]string, len(output))
 	needEmptyCount := false
 	for i, field := range output {
 		if field == "important_kwd" {
 			needEmptyCount = true
 		}
 		if newField, ok := fieldMapping[field]; ok {
-			output[i] = newField
+			mapped[i] = newField
+		} else {
+			mapped[i] = field
 		}
 	}

 	// Remove duplicates
 	seen := make(map[string]bool)
 	result := []string{}
-	for _, f := range output {
+	for _, f := range mapped {
 		if f != "" && !seen[f] {
 			seen[f] = true
 			result = append(result, f)
@@ -2036,4 +2285,4 @@ func (e *infinityEngine) DropChunkStore(ctx context.Context, baseName, datasetID
 // ChunkStoreExists checks if a chunk table exists in Infinity
 func (e *infinityEngine) ChunkStoreExists(ctx context.Context, baseName, datasetID string) (bool, error) {
 	return e.tableExists(ctx, buildChunkTableName(baseName, datasetID))
-}
+}
--- a/internal/engine/infinity/common.go
+++ b/internal/engine/infinity/common.go
@@ -243,6 +243,17 @@ func buildFilterFromCondition(condition map[string]interface{}, tableColumns map
 			}
 			continue
 		}
+		if strListVal, ok := v.([]string); ok {
+			var inVals []string
+			for _, item := range strListVal {
+				item = strings.ReplaceAll(item, "'", "''")
+				inVals = append(inVals, fmt.Sprintf("'%s'", item))
+			}
+			if len(inVals) > 0 {
+				conditions = append(conditions, fmt.Sprintf("%s IN (%s)", k, strings.Join(inVals, ", ")))
+			}
+			continue
+		}

 		// Handle exists condition
 		if k == "exists" {
--- a/internal/engine/infinity/meta_filter.go
+++ b/internal/engine/infinity/meta_filter.go
@@ -0,0 +1,582 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package infinity
+
+import (
+	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+
+	infinity "github.com/infiniflow/infinity-go-sdk"
+)
+
+// Key pattern for validation
+var keyPattern = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`)
+
+// Supported operators
+var supportedOperators = map[string]bool{
+	"=":          true,
+	"≠":          true,
+	">":          true,
+	"<":          true,
+	"≥":          true,
+	"≤":          true,
+	"in":         true,
+	"not in":     true,
+	"contains":   true,
+	"not contains": true,
+	"start with": true,
+	"end with":   true,
+	"empty":      true,
+	"not empty":  true,
+}
+
+// Range operators mapping
+var rangeOps = map[string]string{
+	">":  ">",
+	"<":  "<",
+	"≥":  ">=",
+	"≤":  "<=",
+}
+
+// MetaFilterTranslator translates filter clauses to Infinity SQL
+type MetaFilterTranslator struct{}
+
+// NewMetaFilterTranslator creates a new translator
+func NewMetaFilterTranslator() *MetaFilterTranslator {
+	return &MetaFilterTranslator{}
+}
+
+// Translate translates a single filter dict into Infinity SQL filter string
+func (t *MetaFilterTranslator) Translate(flt map[string]interface{}) (string, error) {
+	op, _ := flt["op"].(string)
+	key, _ := flt["key"].(string)
+	value := flt["value"]
+
+	if key == "" {
+		return "", fmt.Errorf("filter is missing a string key")
+	}
+	if !keyPattern.MatchString(key) {
+		return "", fmt.Errorf("invalid key format (must be identifier-like)")
+	}
+	if !supportedOperators[op] {
+		return "", fmt.Errorf("unknown operator %q", op)
+	}
+
+	switch op {
+	case "empty":
+		return t.translateEmpty(key), nil
+	case "not empty":
+		return t.translateNotEmpty(key), nil
+	case "=":
+		return t.translateEqual(key, value, flt), nil
+	case "≠":
+		return t.translateNotEqual(key, value, flt), nil
+	case ">", "<", "≥", "≤":
+		return t.translateRange(key, op, value, flt), nil
+	case "in":
+		return t.translateIn(key, value, flt), nil
+	case "not in":
+		return t.translateNotIn(key, value, flt), nil
+	case "contains":
+		return t.translateContains(key, value, flt)
+	case "not contains":
+		return t.translateNotContains(key, value, flt), nil
+	case "start with":
+		return t.translateStartWith(key, value, flt), nil
+	case "end with":
+		return t.translateEndWith(key, value, flt), nil
+	}
+
+	return "", fmt.Errorf("no handler for operator %q", op)
+}
+
+func (t *MetaFilterTranslator) translateEmpty(key string) string {
+	return fmt.Sprintf("JSON_EXTRACT_STRING(meta_fields, '$.%s') = '\"\"'", key)
+}
+
+func (t *MetaFilterTranslator) translateNotEmpty(key string) string {
+	return fmt.Sprintf("JSON_EXTRACT_STRING(meta_fields, '$.%s') != '\"\"'", key)
+}
+
+func (t *MetaFilterTranslator) translateEqual(key string, value interface{}, flt map[string]interface{}) string {
+	coerced := coerceScalar(value, flt)
+	if s, ok := coerced.(string); ok {
+		escaped := escapeSQLString(s)
+		return fmt.Sprintf("JSON_CONTAINS(meta_fields, '$.%s', '\"%s\"')", key, escaped)
+	}
+	return fmt.Sprintf("JSON_CONTAINS(meta_fields, '$.%s', %v)", key, coerced)
+}
+
+func (t *MetaFilterTranslator) translateNotEqual(key string, value interface{}, flt map[string]interface{}) string {
+	coerced := coerceScalar(value, flt)
+	if s, ok := coerced.(string); ok {
+		escaped := escapeSQLString(s)
+		return fmt.Sprintf("NOT JSON_CONTAINS(meta_fields, '$.%s', '\"%s\"')", key, escaped)
+	}
+	return fmt.Sprintf("NOT JSON_CONTAINS(meta_fields, '$.%s', %v)", key, coerced)
+}
+
+func (t *MetaFilterTranslator) translateRange(key string, op string, value interface{}, flt map[string]interface{}) string {
+	coerced := coerceRangeValue(value, flt)
+	sqlOp := rangeOps[op]
+	if s, ok := coerced.(string); ok {
+		escaped := escapeSQLString(s)
+		return fmt.Sprintf("JSON_EXTRACT_STRING(meta_fields, '$.%s') %s '%s'", key, sqlOp, escaped)
+	}
+	return fmt.Sprintf("JSON_EXTRACT_DOUBLE(meta_fields, '$.%s') %s %v", key, sqlOp, coerced)
+}
+
+func (t *MetaFilterTranslator) translateIn(key string, value interface{}, flt map[string]interface{}) string {
+	members := csvOrList(value, flt)
+
+	var stringParts, numParts []string
+	for _, m := range members {
+		coerced := coerceRangeValue(m, flt)
+		if num, ok := coerceToFloat(coerced); ok {
+			numParts = append(numParts, fmt.Sprintf("JSON_CONTAINS(meta_fields, '$.%s', %v)", key, num))
+		} else 		if s, ok := coerced.(string); ok {
+			escaped := escapeSQLString(s)
+			stringParts = append(stringParts, fmt.Sprintf("JSON_CONTAINS(meta_fields, '$.%s', '\"%s\"')", key, escaped))
+		}
+	}
+
+	var conditions []string
+	if len(stringParts) > 0 {
+		conditions = append(conditions, "("+strings.Join(stringParts, " OR ")+")")
+	}
+	if len(numParts) > 0 {
+		conditions = append(conditions, "("+strings.Join(numParts, " OR ")+")")
+	}
+
+	return "(" + strings.Join(conditions, " OR ") + ")"
+}
+
+func (t *MetaFilterTranslator) translateNotIn(key string, value interface{}, flt map[string]interface{}) string {
+	members := csvOrList(value, flt)
+
+	var stringParts, numParts []string
+	for _, m := range members {
+		coerced := coerceRangeValue(m, flt)
+		if num, ok := coerceToFloat(coerced); ok {
+			numParts = append(numParts, fmt.Sprintf("NOT JSON_CONTAINS(meta_fields, '$.%s', %v)", key, num))
+		} else if s, ok := coerced.(string); ok {
+			escaped := escapeSQLString(s)
+			stringParts = append(stringParts, fmt.Sprintf("NOT JSON_CONTAINS(meta_fields, '$.%s', '\"%s\"')", key, escaped))
+		}
+	}
+
+	var conditions []string
+	if len(stringParts) > 0 {
+		conditions = append(conditions, "("+strings.Join(stringParts, " AND ")+")")
+	}
+	if len(numParts) > 0 {
+		conditions = append(conditions, "("+strings.Join(numParts, " AND ")+")")
+	}
+
+	return strings.Join(conditions, " AND ")
+}
+
+func (t *MetaFilterTranslator) translateContains(key string, value interface{}, flt map[string]interface{}) (string, error) {
+	// Python guard: if not value and value != 0 -> raise ValueError.
+	// Returning "" here would let the empty fragment slip into the
+	// joined SQL (e.g. "( AND other_condition)"), so we surface the
+	// error instead and let the caller decide how to respond.
+	//
+	// isEmptyValue mirrors Python's `not value` truthiness check so
+	// nil, "", empty slices, and empty maps are all caught — a plain
+	// fmt.Sprintf("%v", ...) == "" test misses those last two.
+	if isEmptyValue(value) && !isNumericZero(value) {
+		return "", fmt.Errorf("contains value is empty: %v", flt)
+	}
+	coerced := coerceRangeValue(value, flt)
+	if num, ok := coerceToFloat(coerced); ok {
+		return fmt.Sprintf("JSON_CONTAINS(meta_fields, '$.%s', %v)", key, num), nil
+	}
+	escaped := escapeSQLString(fmt.Sprintf("%v", value))
+	return fmt.Sprintf("JSON_CONTAINS(meta_fields, '$.%s', '\"%s\"')", key, escaped), nil
+}
+
+func (t *MetaFilterTranslator) translateNotContains(key string, value interface{}, flt map[string]interface{}) string {
+	text := coerceString(value, flt)
+	escaped := escapeSQLString(text)
+	return fmt.Sprintf("NOT JSON_CONTAINS(meta_fields, '$.%s', '\"%s\"')", key, escaped)
+}
+
+func (t *MetaFilterTranslator) translateStartWith(key string, value interface{}, flt map[string]interface{}) string {
+	text := coerceString(value, flt)
+	escaped := escapeSQLString(escapeLikeWildcards(text))
+	return fmt.Sprintf("JSON_EXTRACT_STRING(meta_fields, '$.%s') LIKE '%s%%'", key, escaped)
+}
+
+func (t *MetaFilterTranslator) translateEndWith(key string, value interface{}, flt map[string]interface{}) string {
+	text := coerceString(value, flt)
+	escaped := escapeSQLString(escapeLikeWildcards(text))
+	return fmt.Sprintf("JSON_EXTRACT_STRING(meta_fields, '$.%s') LIKE '%%%s'", key, escaped)
+}
+
+// PlanPushdown translates every filter
+func PlanPushdown(filters []map[string]interface{}, logic string) ([]string, error) {
+	if logic != "and" && logic != "or" {
+		return nil, fmt.Errorf("unknown logic %q", logic)
+	}
+
+	translator := NewMetaFilterTranslator()
+	var result []string
+	for _, flt := range filters {
+		translated, err := translator.Translate(flt)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, translated)
+	}
+	return result, nil
+}
+
+// BuildInfinityFilter builds the full WHERE clause
+func BuildInfinityFilter(filters []map[string]interface{}, logic string) (string, error) {
+	if len(filters) == 0 {
+		return "1=1", nil
+	}
+
+	fragments, err := PlanPushdown(filters, logic)
+	if err != nil {
+		return "", err
+	}
+
+	joiner := " AND "
+	if logic == "or" {
+		joiner = " OR "
+	}
+
+	return "(" + strings.Join(fragments, joiner) + ")", nil
+}
+
+// IsPushdownSupported checks if all filters can be pushed down
+func IsPushdownSupported(filters []map[string]interface{}) bool {
+	for _, flt := range filters {
+		op, _ := flt["op"].(string)
+		if !supportedOperators[op] {
+			return false
+		}
+		key, _ := flt["key"].(string)
+		if key == "" || !keyPattern.MatchString(key) {
+			return false
+		}
+	}
+	return true
+}
+
+// ExtractDocIDs extracts doc IDs from Infinity result
+func ExtractDocIDs(result interface{}) []string {
+	var docIDs []string
+
+	// Try to handle different result types from Infinity SDK
+	switch v := result.(type) {
+	case map[string]interface{}:
+		if idData, ok := v["id"].([]interface{}); ok {
+			for _, id := range idData {
+				if idStr, ok := id.(string); ok {
+					docIDs = append(docIDs, idStr)
+				}
+			}
+		}
+	case *infinity.QueryResult:
+		if v == nil {
+			break
+		}
+		if data, ok := v.Data["id"]; ok {
+			for _, id := range data {
+				if idStr, ok := id.(string); ok {
+					docIDs = append(docIDs, idStr)
+				}
+			}
+		}
+	}
+
+	return docIDs
+}
+
+// coerceScalar handles scalar comparison values.
+// Mirrors Python's ast.literal_eval: tries int first, then float, then string.
+func coerceScalar(value interface{}, flt map[string]interface{}) interface{} {
+	if value == nil {
+		return nil
+	}
+
+	s := strings.TrimSpace(fmt.Sprintf("%v", value))
+
+	// Try to parse as int first (Python ast.literal_eval preserves int vs float)
+	if parsed, err := strconv.ParseInt(s, 10, 64); err == nil {
+		return parsed
+	}
+	// Then try float
+	if parsed, err := strconv.ParseFloat(s, 64); err == nil {
+		return parsed
+	}
+
+	return s
+}
+
+// coerceRangeValue handles range comparison values.
+// Mirrors Python: tries int first, then float, then string.
+func coerceRangeValue(value interface{}, flt map[string]interface{}) interface{} {
+	if value == nil {
+		return nil
+	}
+
+	s := strings.TrimSpace(fmt.Sprintf("%v", value))
+
+	// Try to parse as int first
+	if parsed, err := strconv.ParseInt(s, 10, 64); err == nil {
+		return parsed
+	}
+	// Then try float
+	if parsed, err := strconv.ParseFloat(s, 64); err == nil {
+		return parsed
+	}
+
+	return s
+}
+
+// coerceString ensures value is a non-empty string
+func coerceString(value interface{}, flt map[string]interface{}) string {
+	if value == nil {
+		return ""
+	}
+	s := fmt.Sprintf("%v", value)
+	if s == "" {
+		return ""
+	}
+	return s
+}
+
+// csvOrList handles in/not in values
+func csvOrList(value interface{}, flt map[string]interface{}) []interface{} {
+	if value == nil {
+		return nil
+	}
+
+	var members []interface{}
+
+	switch v := value.(type) {
+	case []interface{}:
+		members = v
+	case string:
+		trimmed := strings.TrimSpace(v)
+		if strings.HasPrefix(trimmed, "[") && strings.HasSuffix(trimmed, "]") {
+			parsed := parseJSONArray(trimmed)
+			if parsed != nil {
+				members = parsed
+			}
+		} else {
+			parts := strings.Split(v, ",")
+			for _, p := range parts {
+				trimmed := strings.TrimSpace(p)
+				if trimmed != "" {
+					members = append(members, strings.ToLower(trimmed))
+				}
+			}
+		}
+	default:
+		members = []interface{}{v}
+	}
+
+	if len(members) == 0 {
+		return nil
+	}
+
+	result := make([]interface{}, len(members))
+	for i, m := range members {
+		if s, ok := m.(string); ok {
+			result[i] = strings.ToLower(strings.TrimSpace(s))
+		} else {
+			result[i] = m
+		}
+	}
+
+	return result
+}
+
+// escapeSQLString escapes SQL string
+func escapeSQLString(s string) string {
+	return strings.ReplaceAll(s, "'", "''")
+}
+
+// escapeLikeWildcards escapes LIKE wildcards
+func escapeLikeWildcards(text string) string {
+	text = strings.ReplaceAll(text, "\\", "\\\\")
+	text = strings.ReplaceAll(text, "%", "\\%")
+	text = strings.ReplaceAll(text, "_", "\\_")
+	return text
+}
+
+// coerceToFloat tries to convert interface{} to float64
+func coerceToFloat(value interface{}) (float64, bool) {
+	switch v := value.(type) {
+	case float64:
+		return v, true
+	case float32:
+		return float64(v), true
+	case int:
+		return float64(v), true
+	case int64:
+		return float64(v), true
+	case string:
+		if f, err := strconv.ParseFloat(v, 64); err == nil {
+			return f, true
+		}
+	}
+	return 0, false
+}
+
+// isNumericZero checks if value is numeric zero (0, 0.0, etc.)
+func isNumericZero(value interface{}) bool {
+	switch v := value.(type) {
+	case int:
+		return v == 0
+	case int64:
+		return v == 0
+	case float64:
+		return v == 0
+	case float32:
+		return v == 0
+	default:
+		return false
+	}
+}
+
+// isEmptyValue mirrors Python's `not value` truthiness for the small
+// set of types we receive in filter dicts. nil, empty strings, and
+// zero-length slices/maps are all considered "empty" — calling
+// fmt.Sprintf("%v", ...) on an empty slice or map produces "[]" or
+// "map[]", neither of which is the empty string, so a stringly-typed
+// guard would miss them.
+func isEmptyValue(value interface{}) bool {
+	if value == nil {
+		return true
+	}
+	switch v := value.(type) {
+	case string:
+		return v == ""
+	case []string:
+		return len(v) == 0
+	case []interface{}:
+		return len(v) == 0
+	case map[string]interface{}:
+		return len(v) == 0
+	}
+	return false
+}
+
+// parseJSONArray parses a simple JSON array string
+func parseJSONArray(s string) []interface{} {
+	s = strings.TrimSpace(s)
+	if len(s) < 2 {
+		return nil
+	}
+	s = s[1 : len(s)-1]
+
+	var result []interface{}
+	parts := splitJSONParts(s)
+	for _, p := range parts {
+		p = strings.TrimSpace(p)
+		if p == "" {
+			continue
+		}
+		if len(p) >= 2 {
+			if (p[0] == '"' && p[len(p)-1] == '"') || (p[0] == '\'' && p[len(p)-1] == '\'') {
+				p = p[1 : len(p)-1]
+			}
+		}
+		result = append(result, p)
+	}
+	return result
+}
+
+// splitJSONParts splits JSON array parts. It tracks the actual quote
+// character that opened the current quoted region so a double-quoted
+// string isn't terminated by a stray single quote inside it (e.g. an
+// apostrophe) and a single-quoted string isn't split by a comma inside
+// a double-quoted neighbour. Naive `inQuote = !inQuote` was wrong on
+// both counts.
+//
+// Quotes preceded by an odd number of backslashes are treated as
+// escaped literals (JSON `\"` / `\'`) so the comma inside a string like
+// `"a\"b,c"` doesn't trigger a spurious split.
+func splitJSONParts(s string) []string {
+	var parts []string
+	var current strings.Builder
+	var quoteChar rune
+	depth := 0
+
+	runes := []rune(s)
+	for i := 0; i < len(runes); i++ {
+		c := runes[i]
+		if c == '\'' || c == '"' {
+			// Inside a quoted string, count the consecutive backslashes
+			// immediately before this rune. An odd count means this
+			// quote is escaped (e.g. JSON `\"`); an even count (incl. 0)
+			// means it really does toggle the quote state.
+			if quoteChar != 0 {
+				bs := 0
+				for j := i - 1; j >= 0 && runes[j] == '\\'; j-- {
+					bs++
+				}
+				if bs%2 == 1 {
+					current.WriteRune(c)
+					continue
+				}
+			}
+			if quoteChar == 0 {
+				quoteChar = c
+			} else if c == quoteChar {
+				quoteChar = 0
+			}
+			current.WriteRune(c)
+			continue
+		}
+		switch c {
+		case '[', '{':
+			if quoteChar == 0 {
+				depth++
+			}
+			current.WriteRune(c)
+		case ']', '}':
+			if quoteChar == 0 {
+				depth--
+			}
+			current.WriteRune(c)
+		case ',':
+			if quoteChar == 0 && depth == 0 {
+				parts = append(parts, current.String())
+				current.Reset()
+			} else {
+				current.WriteRune(c)
+			}
+		default:
+			current.WriteRune(c)
+		}
+	}
+
+	if current.Len() > 0 {
+		parts = append(parts, current.String())
+	}
+
+	return parts
+}
--- a/internal/engine/infinity/meta_filter_test.go
+++ b/internal/engine/infinity/meta_filter_test.go
@@ -0,0 +1,290 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package infinity
+
+import (
+	"reflect"
+	"testing"
+)
+
+// TestSplitJSONParts pins down the quote-aware split. The previous
+// implementation only tracked an `inQuote` bool toggled on `'`, so
+// double-quoted strings containing commas were split incorrectly and
+// single quotes inside double-quoted strings (e.g. apostrophes) were
+// also mis-handled.
+func TestSplitJSONParts(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want []string
+	}{
+		{
+			name: "empty",
+			in:   "",
+			want: nil,
+		},
+		{
+			name: "single_quoted_basic",
+			in:   `'a', 'b', 'c'`,
+			want: []string{`'a'`, ` 'b'`, ` 'c'`},
+		},
+		{
+			name: "double_quoted_with_commas_inside",
+			in:   `"a,b", "c"`,
+			want: []string{`"a,b"`, ` "c"`},
+		},
+		{
+			name: "apostrophe_inside_double_quoted",
+			// "don't", "we, can't" — the apostrophes and the comma in
+			// "we, can't" must not break the double-quoted regions.
+			in:   `"don't", "we, can't"`,
+			want: []string{`"don't"`, ` "we, can't"`},
+		},
+		{
+			name: "nested_brackets_inside_quotes_do_not_count",
+			in:   `"{a,b}", [1, 2]`,
+			want: []string{`"{a,b}"`, ` [1, 2]`},
+		},
+		{
+			name: "comma_inside_brackets_outside_quotes_splits",
+			in:   `[1, 2], [3, 4]`,
+			want: []string{`[1, 2]`, ` [3, 4]`},
+		},
+		{
+			name: "trailing_comma_drops_empty_part",
+			in:   `"a", "b",`,
+			want: []string{`"a"`, ` "b"`},
+		},
+		{
+			// Regression: previously the parser toggled quote state on
+			// every '"', so the comma inside the first element split
+			// the array into three pieces and corrupted in/not in filters.
+			name: "escaped_double_quote_does_not_split",
+			in:   `"a\"b,c", "d"`,
+			want: []string{`"a\"b,c"`, ` "d"`},
+		},
+		{
+			// `\\\"` = escaped backslash + escaped quote. The backslash
+			// is a literal, the quote is escaped, so no split.
+			name: "escaped_backslash_then_escaped_quote",
+			in:   `"a\\\"b", "c"`,
+			want: []string{`"a\\\"b"`, ` "c"`},
+		},
+		{
+			// Symmetric handling for single-quoted regions.
+			name: "escaped_single_quote_does_not_split",
+			in:   `'a\'b,c', 'd'`,
+			want: []string{`'a\'b,c'`, ` 'd'`},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := splitJSONParts(tt.in)
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("splitJSONParts(%q):\n got  %#v\n want %#v", tt.in, got, tt.want)
+			}
+		})
+	}
+}
+
+// TestParseJSONArrayDoubleQuotedWithCommas is the regression case from
+// the bug report: ["a,b", "c"] must yield two elements, not three.
+func TestParseJSONArrayDoubleQuotedWithCommas(t *testing.T) {
+	got := parseJSONArray(`["a,b", "c"]`)
+	want := []interface{}{"a,b", "c"}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("parseJSONArray([\"a,b\", \"c\"]):\n got  %#v\n want %#v", got, want)
+	}
+}
+
+// TestTranslateContainsEmptyValueErrors pins down the empty-value path
+// in translateContains. The Python reference raises ValueError for
+// `if not value and value != 0`; the Go port must mirror that instead
+// of returning "" (which would join into malformed SQL like
+// "( AND other_condition)" once BuildInfinityFilter is called).
+func TestTranslateContainsEmptyValueErrors(t *testing.T) {
+	tr := NewMetaFilterTranslator()
+	cases := []struct {
+		name string
+		flt  map[string]interface{}
+	}{
+		{
+			name: "nil_value",
+			flt:  map[string]interface{}{"op": "contains", "key": "author", "value": nil},
+		},
+		{
+			name: "empty_string",
+			flt:  map[string]interface{}{"op": "contains", "key": "author", "value": ""},
+		},
+		{
+			name: "empty_slice",
+			flt:  map[string]interface{}{"op": "contains", "key": "author", "value": []string{}},
+		},
+		{
+			name: "empty_map",
+			flt:  map[string]interface{}{"op": "contains", "key": "author", "value": map[string]interface{}{}},
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := tr.Translate(tc.flt)
+			if err == nil {
+				t.Fatalf("expected error for empty value, got %q", got)
+			}
+			if got != "" {
+				t.Errorf("expected empty SQL on error, got %q", got)
+			}
+		})
+	}
+}
+
+// TestTranslateContainsNumericZeroStaysGuarded confirms the
+// "value == 0 is not empty" branch inherited from the Python
+// reference: numeric 0 is a real value to search for, not an
+// empty guard rail.
+func TestTranslateContainsNumericZeroStaysGuarded(t *testing.T) {
+	tr := NewMetaFilterTranslator()
+	flt := map[string]interface{}{"op": "contains", "key": "year", "value": 0}
+	got, err := tr.Translate(flt)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	want := `JSON_CONTAINS(meta_fields, '$.year', 0)`
+	if got != want {
+		t.Errorf("got %q, want %q", got, want)
+	}
+}
+
+// TestBuildInfinityFilterRejectsEmptyContains guards the downstream
+// effect: an empty-value contains must propagate as an error out of
+// BuildInfinityFilter, not silently produce "( AND ...)" SQL.
+func TestBuildInfinityFilterRejectsEmptyContains(t *testing.T) {
+	filters := []map[string]interface{}{
+		{"op": "contains", "key": "author", "value": ""},
+		{"op": "=", "key": "year", "value": 2026},
+	}
+	_, err := BuildInfinityFilter(filters, "and")
+	if err == nil {
+		t.Fatal("expected error from BuildInfinityFilter, got nil")
+	}
+}
+
+// TestTotalHitsFromInfinityExtraInfo pins down the parser that decodes
+// the JSON payload Infinity returns in QueryResult.ExtraInfo when the
+// total_hits_count option is set. The shape isn't part of the public
+// SDK contract, so we accept several spellings; getting the
+// (total > cap) trigger wrong would let the push-down silently drop
+// docs and the caller would never fall back to the in-memory path.
+func TestTotalHitsFromInfinityExtraInfo(t *testing.T) {
+	tests := []struct {
+		name    string
+		extra   string
+		wantVal int64
+		wantOk  bool
+	}{
+		{
+			name:    "empty",
+			extra:   "",
+			wantVal: 0,
+			wantOk:  false,
+		},
+		{
+			name:    "invalid_json",
+			extra:   "not json",
+			wantVal: 0,
+			wantOk:  false,
+		},
+		{
+			name:    "total_hits_count_under_cap",
+			extra:   `{"total_hits_count": 42}`,
+			wantVal: 42,
+			wantOk:  true,
+		},
+		{
+			name:    "total_hits_count_at_cap",
+			extra:   `{"total_hits_count": 10000}`,
+			wantVal: 10000,
+			wantOk:  true,
+		},
+		{
+			name:    "total_hits_count_over_cap_triggers_fallback",
+			extra:   `{"total_hits_count": 50000}`,
+			wantVal: 50000,
+			wantOk:  true,
+		},
+		{
+			name:    "camelCase_alias",
+			extra:   `{"totalHitsCount": 7}`,
+			wantVal: 7,
+			wantOk:  true,
+		},
+		{
+			name:    "short_alias",
+			extra:   `{"total": 9}`,
+			wantVal: 9,
+			wantOk:  true,
+		},
+		{
+			name:    "no_recognized_key",
+			extra:   `{"unrelated": 1}`,
+			wantVal: 0,
+			wantOk:  false,
+		},
+		{
+			name:    "negative_value_rejected",
+			extra:   `{"total_hits_count": -1}`,
+			wantVal: 0,
+			wantOk:  false,
+		},
+		{
+			name:    "non_integer_value_rejected",
+			extra:   `{"total_hits_count": "many"}`,
+			wantVal: 0,
+			wantOk:  false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotVal, gotOk := totalHitsFromInfinityExtraInfo(tt.extra)
+			if gotOk != tt.wantOk {
+				t.Errorf("ok mismatch: got %v want %v (extra=%q)", gotOk, tt.wantOk, tt.extra)
+			}
+			if gotVal != tt.wantVal {
+				t.Errorf("value mismatch: got %d want %d (extra=%q)", gotVal, tt.wantVal, tt.extra)
+			}
+		})
+	}
+}
+
+// TestInfinityPushdownCapSemantics documents the overflow contract for
+// the Infinity path: the query is built with .Limit(metaPushdownMaxSize)
+// and .Option({total_hits_count: true}); when the parsed total exceeds
+// the cap, FilterDocIdsByMetaPushdown must return nil so the caller
+// falls back to the in-memory meta_filter rather than returning a
+// truncated slice as a definitive answer.
+func TestInfinityPushdownCapSemantics(t *testing.T) {
+	extra := `{"total_hits_count": 12345}`
+	total, ok := totalHitsFromInfinityExtraInfo(extra)
+	if !ok {
+		t.Fatal("expected total to be parseable from a well-formed Infinity ExtraInfo payload")
+	}
+	if total <= int64(metaPushdownMaxSize) {
+		t.Fatalf("expected total > cap, got total=%d cap=%d", total, metaPushdownMaxSize)
+	}
+}
--- a/internal/engine/infinity/metadata.go
+++ b/internal/engine/infinity/metadata.go
@@ -26,6 +26,8 @@ import (

 	infinity "github.com/infiniflow/infinity-go-sdk"
 	"ragflow/internal/common"
+	"ragflow/internal/dao"
+	"ragflow/internal/engine/types"
 	"ragflow/internal/utility"

 	"go.uber.org/zap"
@@ -110,6 +112,17 @@ func (e *infinityEngine) CreateMetadataStore(ctx context.Context, tenantID strin
 		return fmt.Errorf("Failed to create secondary index on kb_id: %w", err)
 	}

+	// Create secondary index on meta_fields for metadata filter queries
+	_, err = table.CreateIndex(
+		fmt.Sprintf("idx_%s_meta_fields", tableName),
+		infinity.NewIndexInfo("meta_fields", infinity.IndexTypeSecondary, nil),
+		infinity.ConflictTypeIgnore,
+		"",
+	)
+	if err != nil {
+		return fmt.Errorf("Failed to create secondary index on meta_fields: %w", err)
+	}
+
 	return nil
 }

@@ -188,8 +201,31 @@ func (e *infinityEngine) InsertMetadata(ctx context.Context, metadata []map[stri
 }

 // UpdateMetadata updates or inserts document metadata in tenant's metadata table.
-// If a row with the given docID and datasetID exists, it merges the new metadata with existing.
-// If no row exists, it inserts a new row.
+//
+// "Updates" here means MERGE, not replace. The supplied metaFields are
+// overlaid on top of the row's existing meta_fields map: keys already
+// present are overwritten with the new value, keys not in the input
+// are preserved, and brand-new keys are added. If no row exists for
+// (docID, datasetID), one is inserted containing exactly metaFields.
+//
+// Examples (existing row → input → resulting meta_fields):
+//
+//	{character:["曹操","孙权"], year:2025}
+//	  + {author:["John","Tom"], category:"tech"}
+//	  = {character:["曹操","孙权"], year:2025, author:["John","Tom"], category:"tech"}
+//
+//	{character:["曹操","孙权"], year:2025}
+//	  + {year:2025}
+//	  = {character:["曹操","孙权"], year:2025}    // year value unchanged, character preserved
+//
+//	(empty / row absent) + {author:"Tom"} = {author:"Tom"}
+//
+// Note: this is at odds with the SET-METADATA CLI's name, which a
+// reader naturally parses as "replace". The merge semantics exist so
+// that user-driven metadata edits compose with auto-extracted fields
+// produced by the LLM extraction pipeline. See the CLI parser in
+// internal/cli/user_parser.go (parseSetMeta) for the user-facing
+// surface that drives this engine method.
 func (e *infinityEngine) UpdateMetadata(ctx context.Context, docID string, datasetID string, metaFields map[string]interface{}, tenantID string) error {
 	tableName := buildMetadataTableName(tenantID)
 	common.Info("InfinityConnection.UpdateMetadata called", zap.String("tableName", tableName), zap.String("docID", docID), zap.String("datasetID", datasetID))
@@ -499,4 +535,314 @@ func (e *infinityEngine) DropMetadataStore(ctx context.Context, tenantID string)
 func (e *infinityEngine) MetadataStoreExists(ctx context.Context, tenantID string) (bool, error) {
 	tableName := buildMetadataTableName(tenantID)
 	return e.tableExists(ctx, tableName)
-}
+}
+
+// SearchMetadata executes search specifically for metadata tables
+// This is separate from Search() which handles only chunk tables
+func (e *infinityEngine) SearchMetadata(ctx context.Context, req *types.SearchMetadataRequest) (*types.SearchMetadataResult, error) {
+	tenantID := req.TenantID
+	common.Debug("SearchMetadata in Infinity started", zap.String("tenantID", tenantID))
+
+	// Validate inputs
+	if tenantID == "" {
+		return nil, fmt.Errorf("tenantID cannot be empty")
+	}
+
+	// Build table name from tenantID
+	tableName := buildMetadataTableName(tenantID)
+
+	exists, err := e.tableExists(ctx, tableName)
+	if err != nil {
+		common.Warn("Infinity SearchMetadata table existence check failed", zap.String("table", tableName), zap.Error(err))
+		return nil, fmt.Errorf("failed to check metadata table existence: %w", err)
+	}
+	if !exists {
+		common.Debug("Infinity SearchMetadata table absent, returning empty result", zap.String("table", tableName))
+		// Return an empty (non-nil) slice — Python returns `[]`, and a
+		// nil slice is read by callers as "fall back to in-memory". A
+		// zero-match against an absent table is a definitive answer,
+		// not a missing-data condition.
+		return &types.SearchMetadataResult{
+			MetadataRecords: []map[string]interface{}{},
+			Total:           0,
+		}, nil
+	}
+
+	// Build search request for metadata - simpler than chunk search, no match expressions
+	searchReq := &types.SearchRequest{
+		IndexNames:   []string{tableName},
+		Offset:       req.Offset,
+		Limit:        req.Limit,
+		SelectFields: req.SelectFields,
+		Filter:       req.Filter,
+		MatchExprs:   nil, // No match expressions for metadata
+		OrderBy:      req.OrderBy,
+		RankFeature:  nil,
+	}
+
+	result, err := e.Search(ctx, searchReq)
+	if err != nil {
+		return nil, err
+	}
+	return &types.SearchMetadataResult{
+		MetadataRecords: result.Chunks,
+		Total:           result.Total,
+	}, nil
+}
+
+// parseLengthPrefixedJSON parses Infinity's length-prefixed JSON format
+// (a sequence of [4-byte little-endian length][JSON] records) and returns
+// each parsed JSON object. This is the same on-the-wire format that the
+// service layer's ParseAllLengthPrefixedJSON understands; duplicated here
+// to keep the engine package free of service-layer dependencies.
+//
+// The format is what Infinity's SDK returns for VARCHAR/TEXT columns
+// when a query matches multiple rows: instead of giving us a list of
+// per-row byte arrays, it concatenates all rows' values into a single
+// blob, prefixing each with a 4-byte little-endian length.
+//
+// Returns nil if `data` is too short to be valid, or if no JSON
+// objects could be extracted.
+func parseLengthPrefixedJSON(data []byte) []map[string]interface{} {
+	if len(data) < 4 {
+		return nil
+	}
+	var results []map[string]interface{}
+	offset := 0
+	for offset+4 <= len(data) {
+		// Read 4-byte length (little-endian)
+		length := uint32(data[offset]) |
+			uint32(data[offset+1])<<8 |
+			uint32(data[offset+2])<<16 |
+			uint32(data[offset+3])<<24
+		if length == 0 || offset+4+int(length) > len(data) {
+			// Length invalid; bail out.
+			break
+		}
+		jsonStart := offset + 4
+		jsonEnd := jsonStart + int(length)
+		var result map[string]interface{}
+		if err := json.Unmarshal(data[jsonStart:jsonEnd], &result); err == nil {
+			results = append(results, result)
+			offset = jsonEnd
+			continue
+		}
+		break
+	}
+	return results
+}
+
+// realignMetaFieldsColumn fixes a column-oriented data-frame
+// misalignment that happens when Infinity's SDK returns the
+// `meta_fields` column for a multi-row query as a single
+// length-prefixed byte array instead of one entry per row. After the
+// column→row loop has run, the first matching chunk holds the entire
+// concatenated blob and the rest are missing the field. This function
+// splits the blob into per-row JSON objects and reattaches them in
+// order to the chunks that need them.
+//
+// Safe no-op when:
+//   - there are no chunks
+//   - the `meta_fields` column is already aligned (one byte array per
+//     chunk), so a length-prefixed parse of any single value yields
+//     exactly one object
+//   - the byte array doesn't parse as length-prefixed JSON
+func realignMetaFieldsColumn(chunks []map[string]interface{}) {
+	if len(chunks) < 2 {
+		return
+	}
+	firstVal, ok := chunks[0]["meta_fields"]
+	if !ok {
+		return
+	}
+	firstBytes, ok := firstVal.([]byte)
+	if !ok {
+		return
+	}
+	parsed := parseLengthPrefixedJSON(firstBytes)
+	if len(parsed) != len(chunks) {
+		// Either the blob didn't parse as length-prefixed, or it
+		// parsed to a different count than the number of chunks we
+		// built. In either case, don't risk misattributing data.
+		return
+	}
+	for i, meta := range parsed {
+		chunks[i]["meta_fields"] = meta
+	}
+}
+
+// metaPushdownMaxSize caps how many doc IDs the metadata push-down is
+// willing to return in one shot. Matches the Python reference
+// (DocMetadataService.filter_doc_ids_by_meta_pushdown, default limit=10000)
+// and ES's default index.max_result_window.
+//
+// When the underlying query matches more than this, the push-down
+// returns nil and the caller falls back to the in-memory meta_filter,
+// which is correct (just slower for very large result sets). Returning
+// a truncated slice as a definitive answer would silently drop docs.
+const metaPushdownMaxSize = 10000
+
+// FilterDocIdsByMetaPushdown runs a metadata filter directly against the Infinity table.
+//
+// Return value convention (matching Python's filter_doc_ids_by_meta_pushdown):
+//
+//	nil        -> push-down was not viable / errored / result overflowed the
+//	              push-down cap (caller should fall back to in-memory)
+//	[]string{} -> push-down succeeded but found 0 matching docs (empty result is definitive)
+func (e *infinityEngine) FilterDocIdsByMetaPushdown(ctx context.Context, kbIDs []string, conditions []map[string]interface{}, logic string) []string {
+	if len(conditions) == 0 || len(kbIDs) == 0 {
+		return nil
+	}
+
+	// Check if push-down is supported
+	if !IsPushdownSupported(conditions) {
+		common.Debug("FilterDocIdsByMetaPushdown: push-down not supported for some filters")
+		return nil
+	}
+
+	// Get tenant ID from first KB
+	tenantID, err := dao.GetTenantIDByKBID(kbIDs[0])
+	if err != nil {
+		common.Warn("FilterDocIdsByMetaPushdown: failed to get tenant for KB", zap.String("kbID", kbIDs[0]), zap.Error(err))
+		return nil
+	}
+
+	tableName := buildMetadataTableName(tenantID)
+
+	// Build SQL WHERE clause using the full meta_filter logic
+	whereClause, err := BuildInfinityFilter(conditions, logic)
+	if err != nil {
+		common.Debug("FilterDocIdsByMetaPushdown: build filter failed", zap.String("error", err.Error()))
+		return nil
+	}
+
+	// Add KB filter using IN clause. Escape any single quotes in the IDs
+	// defensively — KB IDs are normally UUIDs, but malformed input must
+	// not be able to break out of the literal and alter the query.
+	quotedKBIDs := make([]string, len(kbIDs))
+	for i, kbID := range kbIDs {
+		quotedKBIDs[i] = "'" + strings.ReplaceAll(kbID, "'", "''") + "'"
+	}
+	kbFilter := "kb_id IN (" + strings.Join(quotedKBIDs, ", ") + ")"
+	// Wrap the translated predicate in parens so the AND with the KB clause
+	// doesn't get re-grouped by an internal OR. Without the parens,
+	// `kbFilter AND a OR b` parses as `(kbFilter AND a) OR b`, which can
+	// match rows in other KBs.
+	whereClause = kbFilter + " AND (" + whereClause + ")"
+
+	// Use Infinity connection to execute query
+	db, err := e.client.conn.GetDatabase(e.client.dbName)
+	if err != nil || db == nil {
+		return nil
+	}
+
+	table, err := db.GetTable(tableName)
+	if err != nil || table == nil {
+		return nil
+	}
+
+	// Execute query using chainable API: Output(...).Filter(...)
+	// .Limit(metaPushdownMaxSize) caps the page size, and
+	// .Option({total_hits_count: true}) makes the exact match count
+	// available in QueryResult.ExtraInfo so we can detect overflow and
+	// fall back to the in-memory meta_filter rather than silently
+	// returning a truncated slice (which the caller treats as definitive).
+	common.Debug("FilterDocIdsByMetaPushdown executing Infinity query", zap.String("whereClause", whereClause))
+	queryTable := table.Output([]string{"id"}).Filter(whereClause)
+	queryTable = queryTable.Limit(metaPushdownMaxSize)
+	queryTable = queryTable.Option(map[string]interface{}{"total_hits_count": true})
+	result, err := queryTable.ToResult()
+	if err != nil {
+		return nil
+	}
+
+	qr, ok := result.(*infinity.QueryResult)
+	if !ok || qr == nil {
+		return nil
+	}
+
+	// Detect overflow via the SDK's ExtraInfo payload (a JSON string set
+	// when total_hits_count is requested). If we can't parse it, log
+	// and fall through — the in-memory path is still correct, just
+	// slower.
+	if total, parsed := totalHitsFromInfinityExtraInfo(qr.ExtraInfo); parsed {
+		if total > int64(metaPushdownMaxSize) {
+			common.Warn("FilterDocIdsByMetaPushdown: result exceeds push-down cap, falling back to in-memory",
+				zap.Int64("total", total),
+				zap.Int("cap", metaPushdownMaxSize),
+				zap.Strings("kbIDs", kbIDs),
+			)
+			return nil
+		}
+	} else if qr.ExtraInfo != "" {
+		// ExtraInfo was non-empty but didn't carry total_hits_count in the
+		// expected shape — unusual, but worth flagging so we don't quietly
+		// lose the overflow signal if Infinity changes its payload.
+		common.Debug("FilterDocIdsByMetaPushdown: Infinity ExtraInfo present but total_hits_count missing",
+			zap.String("extraInfo", qr.ExtraInfo),
+		)
+	}
+
+	// Extract doc IDs from the result.
+	docIDs := make([]string, 0)
+	if idData, exists := qr.Data["id"]; exists {
+		for _, id := range idData {
+			if idStr, ok := id.(string); ok {
+				docIDs = append(docIDs, idStr)
+			}
+		}
+	}
+
+	common.Debug("FilterDocIdsByMetaPushdown returned doc IDs", zap.Int("count", len(docIDs)))
+	return docIDs
+}
+
+// totalHitsFromInfinityExtraInfo parses the JSON blob Infinity returns
+// in QueryResult.ExtraInfo when the total_hits_count option is set. The
+// shape is not part of the public SDK contract today (it's a string
+// field with an undocumented layout), so we accept several common
+// spellings and stay tolerant of future changes.
+//
+// Returns (total, true) when a non-negative integer is found, otherwise
+// (0, false) so the caller can decide how to react.
+func totalHitsFromInfinityExtraInfo(extraInfo string) (int64, bool) {
+	if extraInfo == "" {
+		return 0, false
+	}
+	// Try a permissive decode first — Infinity has historically
+	// returned things like {"total_hits_count": 42} but we don't want
+	// to bind to that exact shape forever.
+	var generic map[string]interface{}
+	if err := json.Unmarshal([]byte(extraInfo), &generic); err != nil {
+		return 0, false
+	}
+	for _, key := range []string{"total_hits_count", "totalHitsCount", "total"} {
+		raw, ok := generic[key]
+		if !ok {
+			continue
+		}
+		switch v := raw.(type) {
+		case float64:
+			if v < 0 {
+				return 0, false
+			}
+			return int64(v), true
+		case int64:
+			if v < 0 {
+				return 0, false
+			}
+			return v, true
+		case int:
+			if v < 0 {
+				return 0, false
+			}
+			return int64(v), true
+		case json.Number:
+			n, err := v.Int64()
+			if err == nil && n >= 0 {
+				return n, true
+			}
+		}
+	}
+	return 0, false
+}