Feat: add skills space to context engine (#13908)

### What problem does this PR solve? issue #13714 ### Type of change - [x] New Feature (non-breaking change which adds functionality)
2026-08-15 05:04:27 +08:00 · 2026-04-30 12:36:03 +08:00
parent bb3b99f0a5
commit 4ee0702aed
101 changed files with 19161 additions and 633 deletions
--- a/internal/engine/elasticsearch/client.go
+++ b/internal/engine/elasticsearch/client.go
@@ -36,10 +36,16 @@ type elasticsearchEngine struct {

 // NewEngine creates an Elasticsearch engine
 func NewEngine(cfg interface{}) (*elasticsearchEngine, error) {
+	if cfg == nil {
+		return nil, fmt.Errorf("elasticsearch config is nil, please check your configuration file for 'doc_engine.es' settings")
+	}
 	esConfig, ok := cfg.(*server.ElasticsearchConfig)
 	if !ok {
 		return nil, fmt.Errorf("invalid Elasticsearch config type, expected *config.ElasticsearchConfig")
 	}
+	if esConfig == nil {
+		return nil, fmt.Errorf("elasticsearch config is nil, please check your configuration file for 'doc_engine.es' settings")
+	}

 	// Create ES client
 	client, err := elasticsearch.NewClient(elasticsearch.Config{
@@ -78,8 +84,8 @@ func NewEngine(cfg interface{}) (*elasticsearchEngine, error) {
 	return engine, nil
 }

-// Type returns the engine type
-func (e *elasticsearchEngine) Type() string {
+// GetType returns the engine type
+func (e *elasticsearchEngine) GetType() string {
 	return "elasticsearch"
 }

@@ -243,3 +249,39 @@ func convertBytes(bytes int64) string {
 	}
 	return fmt.Sprintf("%d b", bytes)
 }
+
+// extractErrorReason extracts the error reason from Elasticsearch error response
+// It tries to find the most specific error message in the response
+func extractErrorReason(bodyBytes []byte) string {
+	var errResp map[string]interface{}
+	if err := json.Unmarshal(bodyBytes, &errResp); err != nil {
+		return ""
+	}
+
+	// Try to get error from root_cause
+	if errorObj, ok := errResp["error"].(map[string]interface{}); ok {
+		if rootCauses, ok := errorObj["root_cause"].([]interface{}); ok && len(rootCauses) > 0 {
+			if rootCause, ok := rootCauses[0].(map[string]interface{}); ok {
+				if reason, ok := rootCause["reason"].(string); ok && reason != "" {
+					return reason
+				}
+			}
+		}
+		// Fallback to main error reason
+		if reason, ok := errorObj["reason"].(string); ok && reason != "" {
+			return reason
+		}
+		// Try failed_shards
+		if failedShards, ok := errorObj["failed_shards"].([]interface{}); ok && len(failedShards) > 0 {
+			if shard, ok := failedShards[0].(map[string]interface{}); ok {
+				if reason, ok := shard["reason"].(map[string]interface{}); ok {
+					if r, ok := reason["reason"].(string); ok && r != "" {
+						return r
+					}
+				}
+			}
+		}
+	}
+
+	return ""
+}
--- a/internal/engine/elasticsearch/document.go
+++ b/internal/engine/elasticsearch/document.go
@@ -0,0 +1,259 @@
+//
+//  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 elasticsearch
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+
+	"github.com/elastic/go-elasticsearch/v8/esapi"
+)
+
+// IndexDocument indexes a single document
+func (e *elasticsearchEngine) IndexDocument(ctx context.Context, indexName, docID string, doc interface{}) error {
+	if indexName == "" {
+		return fmt.Errorf("index name cannot be empty")
+	}
+	if docID == "" {
+		return fmt.Errorf("document id cannot be empty")
+	}
+	if doc == nil {
+		return fmt.Errorf("document cannot be nil")
+	}
+
+	// Serialize document
+	data, err := json.Marshal(doc)
+	if err != nil {
+		return fmt.Errorf("failed to marshal document: %w", err)
+	}
+
+	// Index document
+	req := esapi.IndexRequest{
+		Index:      indexName,
+		DocumentID: docID,
+		Body:       bytes.NewReader(data),
+		Refresh:    "true",
+	}
+
+	res, err := req.Do(ctx, e.client)
+	if err != nil {
+		return fmt.Errorf("failed to index document: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.IsError() {
+		body, _ := io.ReadAll(res.Body)
+		reason := extractErrorReason(body)
+		if reason != "" {
+			return fmt.Errorf("elasticsearch error: %s", reason)
+		}
+		return fmt.Errorf("elasticsearch returned error: %s, body: %s", res.Status(), string(body))
+	}
+
+	return nil
+}
+
+// BulkIndex indexes documents in bulk
+func (e *elasticsearchEngine) BulkIndex(ctx context.Context, indexName string, docs []interface{}) (interface{}, error) {
+	if indexName == "" {
+		return nil, fmt.Errorf("index name cannot be empty")
+	}
+	if len(docs) == 0 {
+		return nil, fmt.Errorf("documents cannot be empty")
+	}
+
+	// Build bulk request
+	var buf bytes.Buffer
+	for _, doc := range docs {
+		docMap, ok := doc.(map[string]interface{})
+		if !ok {
+			return nil, fmt.Errorf("document must be map[string]interface{}")
+		}
+
+		docID, hasID := docMap["_id"]
+		if !hasID {
+			return nil, fmt.Errorf("document missing _id field")
+		}
+
+		// Delete _id field to avoid duplication
+		delete(docMap, "_id")
+
+		// Add index operation
+		meta := map[string]interface{}{
+			"_index": indexName,
+			"_id":    docID,
+		}
+		metaData, _ := json.Marshal(meta)
+		docData, _ := json.Marshal(docMap)
+
+		buf.Write(metaData)
+		buf.WriteByte('\n')
+		buf.Write(docData)
+		buf.WriteByte('\n')
+	}
+
+	// Execute bulk request
+	req := esapi.BulkRequest{
+		Body:    &buf,
+		Refresh: "true",
+	}
+
+	res, err := req.Do(ctx, e.client)
+	if err != nil {
+		return nil, fmt.Errorf("bulk index failed: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.IsError() {
+		body, _ := io.ReadAll(res.Body)
+		reason := extractErrorReason(body)
+		if reason != "" {
+			return nil, fmt.Errorf("elasticsearch error: %s", reason)
+		}
+		return nil, fmt.Errorf("elasticsearch returned error: %s", res.Status())
+	}
+
+	// Parse response
+	var result map[string]interface{}
+	if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
+		return nil, fmt.Errorf("failed to parse response: %w", err)
+	}
+
+	// Check for errors
+	if errors, ok := result["errors"].(bool); ok && errors {
+		// Get error details
+		if items, ok := result["items"].([]interface{}); ok && len(items) > 0 {
+			for _, item := range items {
+				if itemMap, ok := item.(map[string]interface{}); ok {
+					for _, op := range itemMap {
+						if opMap, ok := op.(map[string]interface{}); ok {
+							if errInfo, ok := opMap["error"].(map[string]interface{}); ok {
+								if reason, ok := errInfo["reason"].(string); ok {
+									return nil, fmt.Errorf("bulk index error: %s", reason)
+								}
+							}
+						}
+					}
+				}
+			}
+		}
+		return nil, fmt.Errorf("bulk index has errors")
+	}
+
+	response := &BulkResponse{
+		Took:    int64(result["took"].(float64)),
+		Errors:  result["errors"].(bool),
+		Indexed: len(docs),
+	}
+
+	return response, nil
+}
+
+// BulkResponse bulk operation response
+type BulkResponse struct {
+	Took    int64
+	Errors  bool
+	Indexed int
+}
+
+// GetDocument gets a document
+func (e *elasticsearchEngine) GetDocument(ctx context.Context, indexName, docID string) (interface{}, error) {
+	if indexName == "" {
+		return nil, fmt.Errorf("index name cannot be empty")
+	}
+	if docID == "" {
+		return nil, fmt.Errorf("document id cannot be empty")
+	}
+
+	// Get document
+	req := esapi.GetRequest{
+		Index:      indexName,
+		DocumentID: docID,
+	}
+
+	res, err := req.Do(ctx, e.client)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get document: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.StatusCode == 404 {
+		return nil, fmt.Errorf("document not found")
+	}
+
+	if res.IsError() {
+		body, _ := io.ReadAll(res.Body)
+		reason := extractErrorReason(body)
+		if reason != "" {
+			return nil, fmt.Errorf("elasticsearch error: %s", reason)
+		}
+		return nil, fmt.Errorf("elasticsearch returned error: %s", res.Status())
+	}
+
+	// Parse response
+	var result map[string]interface{}
+	if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
+		return nil, fmt.Errorf("failed to parse response: %w", err)
+	}
+
+	if found, ok := result["found"].(bool); !ok || !found {
+		return nil, fmt.Errorf("document not found")
+	}
+
+	return result["_source"], nil
+}
+
+// DeleteDocument deletes a document
+func (e *elasticsearchEngine) DeleteDocument(ctx context.Context, indexName, docID string) error {
+	if indexName == "" {
+		return fmt.Errorf("index name cannot be empty")
+	}
+	if docID == "" {
+		return fmt.Errorf("document id cannot be empty")
+	}
+
+	// Delete document
+	req := esapi.DeleteRequest{
+		Index:      indexName,
+		DocumentID: docID,
+		Refresh:    "true",
+	}
+
+	res, err := req.Do(ctx, e.client)
+	if err != nil {
+		return fmt.Errorf("failed to delete document: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.StatusCode == 404 {
+		return fmt.Errorf("document not found")
+	}
+
+	if res.IsError() {
+		body, _ := io.ReadAll(res.Body)
+		reason := extractErrorReason(body)
+		if reason != "" {
+			return fmt.Errorf("elasticsearch error: %s", reason)
+		}
+		return fmt.Errorf("elasticsearch returned error: %s, body: %s", res.Status(), string(body))
+	}
+
+	return nil
+}
--- a/internal/engine/elasticsearch/index.go
+++ b/internal/engine/elasticsearch/index.go
@@ -22,19 +22,13 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"os"

 	"github.com/elastic/go-elasticsearch/v8/esapi"
 )

 // CreateDataset creates an index
 func (e *elasticsearchEngine) CreateDataset(ctx context.Context, indexName, datasetID string, vectorSize int, parserID string) error {
-	// Elasticsearch doesn't support vector_size or parser_id in the same way
-	// Build mapping for ES (if needed)
-	// TODO
-	mapping := map[string]interface{}{
-		"dataset_id": datasetID,
-	}
-
 	if indexName == "" {
 		return fmt.Errorf("index name cannot be empty")
 	}
@@ -48,6 +42,25 @@ func (e *elasticsearchEngine) CreateDataset(ctx context.Context, indexName, data
 		return fmt.Errorf("index '%s' already exists", indexName)
 	}

+	// Load mapping based on index type
+	var mapping map[string]interface{}
+	if datasetID == "skill" {
+		// Load skill-specific mapping
+		skillMapping, err := loadSkillMapping()
+		if err != nil {
+			return fmt.Errorf("failed to load skill mapping: %w", err)
+		}
+		mapping = skillMapping
+	} else {
+		// Default mapping for dataset
+		mapping = map[string]interface{}{
+			"settings": map[string]interface{}{
+				"number_of_shards":   1,
+				"number_of_replicas": 0,
+			},
+		}
+	}
+
 	// Prepare request body
 	var body io.Reader
 	if mapping != nil {
@@ -71,7 +84,12 @@ func (e *elasticsearchEngine) CreateDataset(ctx context.Context, indexName, data
 	defer res.Body.Close()

 	if res.IsError() {
-		return fmt.Errorf("elasticsearch returned error: %s", res.Status())
+		bodyBytes, _ := io.ReadAll(res.Body)
+		reason := extractErrorReason(bodyBytes)
+		if reason != "" {
+			return fmt.Errorf("elasticsearch error: %s", reason)
+		}
+		return fmt.Errorf("elasticsearch returned error: %s, body: %s", res.Status(), string(bodyBytes))
 	}

 	// Parse response
@@ -88,6 +106,157 @@ func (e *elasticsearchEngine) CreateDataset(ctx context.Context, indexName, data
 	return nil
 }

+// loadSkillMapping loads the skill index mapping from config file
+func loadSkillMapping() (map[string]interface{}, error) {
+	// Try multiple possible locations for the mapping file
+	possiblePaths := []string{
+		"conf/skill_es_mapping.json",
+		"../conf/skill_es_mapping.json",
+		"/app/conf/skill_es_mapping.json",
+	}
+
+	var data []byte
+	var err error
+	for _, path := range possiblePaths {
+		data, err = os.ReadFile(path)
+		if err == nil {
+			break
+		}
+	}
+
+	if err != nil {
+		// Fallback to default skill mapping if file not found
+		return getDefaultSkillMapping(), nil
+	}
+
+	var mapping map[string]interface{}
+	if err := json.Unmarshal(data, &mapping); err != nil {
+		return nil, fmt.Errorf("failed to parse skill mapping: %w", err)
+	}
+
+	return mapping, nil
+}
+
+// getDefaultSkillMapping returns the default skill index mapping
+func getDefaultSkillMapping() map[string]interface{} {
+	return map[string]interface{}{
+		"settings": map[string]interface{}{
+			"index": map[string]interface{}{
+				"number_of_shards":   1,
+				"number_of_replicas": 0,
+				"refresh_interval":   "1000ms",
+			},
+		},
+		"mappings": map[string]interface{}{
+			"dynamic": false,
+			"properties": map[string]interface{}{
+				"skill_id": map[string]interface{}{
+					"type":  "keyword",
+					"store": true,
+				},
+				"name": map[string]interface{}{
+					"type":  "text",
+					"index": false,
+					"store": true,
+				},
+				"name_tks": map[string]interface{}{
+					"type":     "text",
+					"analyzer": "whitespace",
+					"store":    true,
+				},
+				"tags": map[string]interface{}{
+					"type":  "text",
+					"index": false,
+					"store": true,
+				},
+				"tags_tks": map[string]interface{}{
+					"type":     "text",
+					"analyzer": "whitespace",
+					"store":    true,
+				},
+				"description": map[string]interface{}{
+					"type":  "text",
+					"index": false,
+					"store": true,
+				},
+				"description_tks": map[string]interface{}{
+					"type":     "text",
+					"analyzer": "whitespace",
+					"store":    true,
+				},
+				"content": map[string]interface{}{
+					"type":  "text",
+					"index": false,
+					"store": true,
+				},
+				"content_tks": map[string]interface{}{
+					"type":     "text",
+					"analyzer": "whitespace",
+					"store":    true,
+				},
+				"q_3072_vec": map[string]interface{}{
+					"type":       "dense_vector",
+					"dims":       3072,
+					"index":      true,
+					"similarity": "cosine",
+				},
+				"q_2560_vec": map[string]interface{}{
+					"type":       "dense_vector",
+					"dims":       2560,
+					"index":      true,
+					"similarity": "cosine",
+				},
+				"q_1536_vec": map[string]interface{}{
+					"type":       "dense_vector",
+					"dims":       1536,
+					"index":      true,
+					"similarity": "cosine",
+				},
+				"q_1024_vec": map[string]interface{}{
+					"type":       "dense_vector",
+					"dims":       1024,
+					"index":      true,
+					"similarity": "cosine",
+				},
+				"q_768_vec": map[string]interface{}{
+					"type":       "dense_vector",
+					"dims":       768,
+					"index":      true,
+					"similarity": "cosine",
+				},
+				"q_512_vec": map[string]interface{}{
+					"type":       "dense_vector",
+					"dims":       512,
+					"index":      true,
+					"similarity": "cosine",
+				},
+				"q_256_vec": map[string]interface{}{
+					"type":       "dense_vector",
+					"dims":       256,
+					"index":      true,
+					"similarity": "cosine",
+				},
+				"version": map[string]interface{}{
+					"type":  "keyword",
+					"store": true,
+				},
+				"status": map[string]interface{}{
+					"type":  "keyword",
+					"store": true,
+				},
+				"create_time": map[string]interface{}{
+					"type":  "long",
+					"store": true,
+				},
+				"update_time": map[string]interface{}{
+					"type":  "long",
+					"store": true,
+				},
+			},
+		},
+	}
+}
+
 // DropTable deletes an index
 func (e *elasticsearchEngine) DropTable(ctx context.Context, indexName string) error {
 	if indexName == "" {
@@ -115,6 +284,11 @@ func (e *elasticsearchEngine) DropTable(ctx context.Context, indexName string) e
 	defer res.Body.Close()

 	if res.IsError() {
+		bodyBytes, _ := io.ReadAll(res.Body)
+		reason := extractErrorReason(bodyBytes)
+		if reason != "" {
+			return fmt.Errorf("elasticsearch error: %s", reason)
+		}
 		return fmt.Errorf("elasticsearch returned error: %s", res.Status())
 	}

@@ -143,6 +317,11 @@ func (e *elasticsearchEngine) TableExists(ctx context.Context, indexName string)
 		return false, nil
 	}

+	bodyBytes, _ := io.ReadAll(res.Body)
+	reason := extractErrorReason(bodyBytes)
+	if reason != "" {
+		return false, fmt.Errorf("elasticsearch error: %s", reason)
+	}
 	return false, fmt.Errorf("elasticsearch returned error: %s", res.Status())
 }

--- a/internal/engine/elasticsearch/search.go
+++ b/internal/engine/elasticsearch/search.go
@@ -22,6 +22,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"strings"

 	"github.com/elastic/go-elasticsearch/v8/esapi"
 	"go.uber.org/zap"
@@ -63,17 +64,28 @@ func (e *elasticsearchEngine) searchUnified(ctx context.Context, req *types.Sear
 		limit = 30 // default ES size
 	}

-	// Build filter clauses (default: available=1, meaning available_int >= 1)
-	// Reference: rag/utils/es_conn.py L60-L78
-	filterClauses := buildFilterClauses(req.KbIDs, 1)
+	// Check if this is a skill index
+	isSkillIndex := len(req.IndexNames) > 0 && strings.HasPrefix(req.IndexNames[0], "skill_")
+
+	// Build filter clauses
+	var filterClauses []map[string]interface{}
+	if isSkillIndex {
+		filterClauses = buildSkillFilterClauses()
+	} else {
+		filterClauses = buildFilterClauses(req.KbIDs, 1)
+	}
+
+	// Add filters from req.Filter
+	if req.Filter != nil && len(req.Filter) > 0 {
+		filterClauses = append(filterClauses, buildFilterFromMap(req.Filter)...)
+	}

 	// Build search query body
 	queryBody := make(map[string]interface{})

 	// Determine search type from MatchExprs
 	var matchText string
-	var matchDense interface{}
-	var textWeight float64 = 1.0
+	var matchDense *types.MatchDenseExpr
 	var hasVectorMatch bool

 	for _, expr := range req.MatchExprs {
@@ -83,59 +95,82 @@ func (e *elasticsearchEngine) searchUnified(ctx context.Context, req *types.Sear
 		switch e := expr.(type) {
 		case string:
 			matchText = e
+		case *types.MatchTextExpr:
+			matchText = e.MatchingText
 		case *types.MatchDenseExpr:
 			hasVectorMatch = true
 			matchDense = e
-			textWeight = 0.3 // default, should be passed via SimilarityThreshold
 		}
 	}

 	var vectorFieldName string
-	if !hasVectorMatch {
+	if !hasVectorMatch || matchDense == nil {
 		// Keyword-only search
-		queryBody["query"] = buildESKeywordQuery(matchText, filterClauses, 1.0)
+		if isSkillIndex {
+			queryBody["query"] = buildSkillKeywordQuery(matchText, filterClauses, 1.0)
+		} else {
+			queryBody["query"] = buildESKeywordQuery(matchText, filterClauses, 1.0)
+		}
 	} else {
 		// Hybrid search: keyword + vector
-		// Calculate text weight (use SimilarityThreshold as text weight if provided)
+		textWeight := 0.7 // default: vector weight = 0.3
+		vectorWeight := 0.3
+		if matchDense.ExtraOptions != nil {
+			if vw, ok := matchDense.ExtraOptions["text_weight"].(float64); ok {
+				textWeight = vw
+			}
+			if vw, ok := matchDense.ExtraOptions["vector_weight"].(float64); ok {
+				vectorWeight = vw
+			}
+		}

 		// Build boolean query for text match and filters
-		boolQuery := buildESKeywordQuery(matchText, filterClauses, 1.0)
+		var boolQuery map[string]interface{}
+		if isSkillIndex {
+			boolQuery = buildSkillKeywordQuery(matchText, filterClauses, 1.0)
+		} else {
+			boolQuery = buildESKeywordQuery(matchText, filterClauses, 1.0)
+		}
 		// Add boost to the bool query (as in Python code)
 		if boolMap, ok := boolQuery["bool"].(map[string]interface{}); ok {
 			boolMap["boost"] = textWeight
 		}
+
 		// Build kNN query
-		var vectorData []float64
-		if md, ok := matchDense.(*types.MatchDenseExpr); ok {
-			vectorData = md.EmbeddingData
-			vectorFieldName = md.VectorColumnName
-			k := md.TopN
-			if k <= 0 {
-				k = req.Limit
-			}
-			if k <= 0 {
-				k = 1024
-			}
-			numCandidates := k * 2
-
-			knnQuery := map[string]interface{}{
-				"field":          vectorFieldName,
-				"query_vector":   vectorData,
-				"k":              k,
-				"num_candidates": numCandidates,
-				"filter":         boolQuery,
-				"similarity":     0.0,
-			}
-
-			queryBody["knn"] = knnQuery
-			queryBody["query"] = boolQuery
+		vectorData := matchDense.EmbeddingData
+		vectorFieldName = matchDense.VectorColumnName
+		k := matchDense.TopN
+		if k <= 0 {
+			k = req.Limit
 		}
+		if k <= 0 {
+			k = 1024
+		}
+		numCandidates := k * 2
+
+		similarity := 0.0
+		if matchDense.ExtraOptions != nil {
+			if sim, ok := matchDense.ExtraOptions["similarity"].(float64); ok {
+				similarity = sim
+			}
+		}
+
+		knnQuery := map[string]interface{}{
+			"field":          vectorFieldName,
+			"query_vector":   vectorData,
+			"k":              k,
+			"num_candidates": numCandidates,
+			"similarity":     similarity,
+			"boost":          vectorWeight,
+		}
+
+		queryBody["knn"] = knnQuery
+		queryBody["query"] = boolQuery

 		// Add vector column to Source fields (matching Python ES: src.append(f"q_{len(q_vec)}_vec"))
 		// Only modify Source if it was explicitly set by the caller
 		if vectorFieldName != "" && len(req.SelectFields) > 0 {
 			sourceFields := req.SelectFields
-			// Check if vector column already in source
 			found := false
 			for _, f := range sourceFields {
 				if f == vectorFieldName {
@@ -153,6 +188,14 @@ func (e *elasticsearchEngine) searchUnified(ctx context.Context, req *types.Sear
 	queryBody["size"] = limit
 	queryBody["from"] = offset

+	// Add sorting if specified
+	if req.OrderBy != nil {
+		sort := parseOrderByExpr(req.OrderBy)
+		if len(sort) > 0 {
+			queryBody["sort"] = sort
+		}
+	}
+
 	// Serialize query
 	var buf bytes.Buffer
 	if err := json.NewEncoder(&buf).Encode(queryBody); err != nil {
@@ -228,7 +271,7 @@ func calculatePagination(page, size, topK int) (int, int) {
 	return offset, RERANK_LIMIT
 }

-// buildFilterClauses builds ES filter clauses from kb_ids, doc_ids and available_int
+// buildFilterClauses builds ES filter clauses from kb_ids and available_int
 // Reference: rag/utils/es_conn.py L60-L78
 // When available=0: available_int < 1
 // When available!=0: NOT (available_int < 1)
@@ -272,22 +315,96 @@ func buildFilterClauses(kbIDs []string, available int) []map[string]interface{}
 	return filters
 }

+// buildSkillFilterClauses builds ES filter clauses for skill index
+// Skill index uses 'status' field instead of 'available_int'
+func buildSkillFilterClauses() []map[string]interface{} {
+	// Filter for active skills (status = "1")
+	return []map[string]interface{}{
+		{
+			"term": map[string]interface{}{
+				"status": "1",
+			},
+		},
+	}
+}
+
+// buildFilterFromMap converts a generic filter map to ES filter clauses
+func buildFilterFromMap(filter map[string]interface{}) []map[string]interface{} {
+	var filters []map[string]interface{}
+	for field, value := range filter {
+		switch v := value.(type) {
+		case []string:
+			filters = append(filters, map[string]interface{}{
+				"terms": map[string]interface{}{field: v},
+			})
+		case []interface{}:
+			filters = append(filters, map[string]interface{}{
+				"terms": map[string]interface{}{field: v},
+			})
+		default:
+			filters = append(filters, map[string]interface{}{
+				"term": map[string]interface{}{field: v},
+			})
+		}
+	}
+	return filters
+}
+
 // buildESKeywordQuery builds keyword-only search query for ES
 // Uses query_string if matchText is in query_string format, otherwise uses multi_match
 // boost is applied to the text match clause (query_string or multi_match)
 func buildESKeywordQuery(matchText string, filterClauses []map[string]interface{}, boost float64) map[string]interface{} {
 	var mustClause map[string]interface{}

-	// Use query_string for complex queries
-	queryString := map[string]interface{}{
-		"query":                matchText,
-		"fields":               []string{"title_tks^10", "title_sm_tks^5", "important_kwd^30", "important_tks^20", "question_tks^20", "content_ltks^2", "content_sm_ltks"},
-		"type":                 "best_fields",
-		"minimum_should_match": "30%",
-		"boost":                boost,
+	// Handle wildcard query (match all)
+	if matchText == "*" || matchText == "" {
+		mustClause = map[string]interface{}{
+			"match_all": map[string]interface{}{},
+		}
+	} else {
+		// Use query_string for complex queries
+		queryString := map[string]interface{}{
+			"query":                matchText,
+			"fields":               []string{"title_tks^10", "title_sm_tks^5", "important_kwd^30", "important_tks^20", "question_tks^20", "content_ltks^2", "content_sm_ltks"},
+			"type":                 "best_fields",
+			"minimum_should_match": "30%",
+			"boost":                boost,
+		}
+		mustClause = map[string]interface{}{
+			"query_string": queryString,
+		}
 	}
-	mustClause = map[string]interface{}{
-		"query_string": queryString,
+
+	return map[string]interface{}{
+		"bool": map[string]interface{}{
+			"must":   mustClause,
+			"filter": filterClauses,
+		},
+	}
+}
+
+// buildSkillKeywordQuery builds keyword-only search query for skill index
+// Skill index uses different field names: name_tks, tags_tks, description_tks, content_tks
+func buildSkillKeywordQuery(matchText string, filterClauses []map[string]interface{}, boost float64) map[string]interface{} {
+	var mustClause map[string]interface{}
+
+	// Handle wildcard query (match all)
+	if matchText == "*" || matchText == "" {
+		mustClause = map[string]interface{}{
+			"match_all": map[string]interface{}{},
+		}
+	} else {
+		// Use query_string for complex queries with skill-specific fields
+		queryString := map[string]interface{}{
+			"query":                matchText,
+			"fields":               []string{"name_tks^10", "tags_tks^5", "description_tks^3", "content_tks^1"},
+			"type":                 "best_fields",
+			"minimum_should_match": "30%",
+			"boost":                boost,
+		}
+		mustClause = map[string]interface{}{
+			"query_string": queryString,
+		}
 	}

 	return map[string]interface{}{
@@ -306,18 +423,40 @@ func convertESResponse(esResp *SearchResponse, vectorFieldName string) []map[str

 	chunks := make([]map[string]interface{}, len(esResp.Hits.Hits))
 	for i, hit := range esResp.Hits.Hits {
-
-		//// vectorField is list of float64, which need to be converted to float32
-
 		chunks[i] = hit.Source
 		chunks[i]["_score"] = hit.Score
 		chunks[i]["_id"] = hit.ID
-		//vectorField := hit.Source[vectorFieldName]
-		//chunks[i][vectorFieldName] = utility.Float64ToFloat32(vectorField)
 	}
 	return chunks
 }

+// parseOrderByExpr parses the OrderBy expression into ES sort format
+func parseOrderByExpr(orderBy *types.OrderByExpr) []map[string]interface{} {
+	if orderBy == nil || len(orderBy.Fields) == 0 {
+		return nil
+	}
+
+	var result []map[string]interface{}
+	for _, field := range orderBy.Fields {
+		direction := "asc"
+		if field.Type == types.SortDesc {
+			direction = "desc"
+		}
+
+		if field.Field == "_score" || field.Field == "score" {
+			result = append(result, map[string]interface{}{
+				"_score": direction,
+			})
+		} else {
+			result = append(result, map[string]interface{}{
+				field.Field: direction,
+			})
+		}
+	}
+
+	return result
+}
+
 // Helper query builder functions (legacy)

 // BuildMatchTextQuery builds a text match query
--- a/internal/engine/engine.go
+++ b/internal/engine/engine.go
@@ -53,6 +53,11 @@ type DocEngine interface {
 	DropTable(ctx context.Context, indexName string) error
 	TableExists(ctx context.Context, indexName string) (bool, error)

+	// Document operations (used by skill indexing)
+	IndexDocument(ctx context.Context, indexName, docID string, doc interface{}) error
+	DeleteDocument(ctx context.Context, indexName, docID string) error
+	BulkIndex(ctx context.Context, indexName string, docs []interface{}) (interface{}, error)
+
 	// Utility functions for search result processing
 	GetFields(chunks []map[string]interface{}, fields []string) map[string]map[string]interface{}
 	GetAggregation(chunks []map[string]interface{}, fieldName string) []map[string]interface{}
@@ -62,6 +67,9 @@ type DocEngine interface {
 	// Health check
 	Ping(ctx context.Context) error
 	Close() error
+
+	// GetType returns the engine type
+	GetType() string
 }

 // Type returns the engine type (helper method for runtime type checking)
--- a/internal/engine/infinity/client.go
+++ b/internal/engine/infinity/client.go
@@ -129,10 +129,16 @@ type infinityEngine struct {

 // NewEngine creates an Infinity engine
 func NewEngine(cfg interface{}) (*infinityEngine, error) {
+	if cfg == nil {
+		return nil, fmt.Errorf("infinity config is nil, please check your configuration file for 'doc_engine.infinity' settings")
+	}
 	infConfig, ok := cfg.(*server.InfinityConfig)
 	if !ok {
 		return nil, fmt.Errorf("invalid infinity config type, expected *config.InfinityConfig")
 	}
+	if infConfig == nil {
+		return nil, fmt.Errorf("infinity config is nil, please check your configuration file for 'doc_engine.infinity' settings")
+	}

 	client, err := NewInfinityClient(infConfig)
 	if err != nil {
@@ -168,8 +174,8 @@ func NewEngine(cfg interface{}) (*infinityEngine, error) {
 	return engine, nil
 }

-// Type returns the engine type
-func (e *infinityEngine) Type() string {
+// GetType returns the engine type
+func (e *infinityEngine) GetType() string {
 	return "infinity"
 }

--- a/internal/engine/infinity/common.go
+++ b/internal/engine/infinity/common.go
@@ -312,3 +312,27 @@ func buildFilterFromCondition(condition map[string]interface{}, tableColumns map
 	}
 	return strings.Join(conditions, " AND ")
 }
+
+// columnExists checks if a column exists in the table
+func (e *infinityEngine) columnExists(table *infinity.Table, columnName string) (bool, error) {
+	colsResp, err := table.ShowColumns()
+	if err != nil {
+		return false, err
+	}
+
+	result, ok := colsResp.(*infinity.QueryResult)
+	if !ok {
+		return false, fmt.Errorf("unexpected response type: %T", colsResp)
+	}
+
+	// ShowColumns returns a result set where Data contains arrays of column values
+	if nameArr, ok := result.Data["name"]; ok {
+		for i := 0; i < len(nameArr); i++ {
+			colName, _ := nameArr[i].(string)
+			if colName == columnName {
+				return true, nil
+			}
+		}
+	}
+	return false, nil
+}
--- a/internal/engine/infinity/dataset.go
+++ b/internal/engine/infinity/dataset.go
@@ -36,14 +36,28 @@ import (
 // CreateDataset creates a table in Infinity
 // indexName is the table name prefix (e.g., "ragflow_<tenant_id>")
 // The full table name is built as "{indexName}_{datasetID}"
+// For skill index (datasetID="skill"), tableName is just indexName and uses skill_infinity_mapping.json
 func (e *infinityEngine) CreateDataset(ctx context.Context, indexName, datasetID string, vectorSize int, parserID string) error {
 	vecSize := vectorSize

-	// Build full table name: {indexName}_{datasetID}
-	tableName := fmt.Sprintf("%s_%s", indexName, datasetID)
+	// Determine table name and mapping file based on index type
+	var tableName string
+	var mappingFile string
+
+	if datasetID == "skill" {
+		// Skill index: table name is just indexName (e.g., "skill_abc123_def456")
+		tableName = indexName
+		mappingFile = "skill_infinity_mapping.json"
+		logger.Info("Creating skill index table", zap.String("tableName", tableName), zap.String("mappingFile", mappingFile))
+	} else {
+		// Regular document index: table name is {indexName}_{datasetID}
+		tableName = fmt.Sprintf("%s_%s", indexName, datasetID)
+		mappingFile = e.mappingFileName
+		logger.Info("Creating regular index table", zap.String("tableName", tableName), zap.String("mappingFile", mappingFile))
+	}

 	// Use configured schema
-	fpMapping := filepath.Join(utility.GetProjectRoot(), "conf", e.mappingFileName)
+	fpMapping := filepath.Join(utility.GetProjectRoot(), "conf", mappingFile)

 	schemaData, err := os.ReadFile(fpMapping)
 	if err != nil {
@@ -61,54 +75,90 @@ func (e *infinityEngine) CreateDataset(ctx context.Context, indexName, datasetID
 		return fmt.Errorf("Failed to get database: %w", err)
 	}

+	// Determine vector column name
+	vectorColName := fmt.Sprintf("q_%d_vec", vecSize)
+
 	// Check if table already exists
 	exists, err := e.TableExists(ctx, tableName)
 	if err != nil {
 		return fmt.Errorf("Failed to check if table exists: %w", err)
 	}
+
+	var table *infinity.Table
 	if exists {
-		return fmt.Errorf("table '%s' already exists", tableName)
-	}
-
-	// Build column definitions (preserving JSON order)
-	var columns infinity.TableSchema
-	for _, fieldName := range schema.Keys {
-		fieldInfo := schema.Fields[fieldName]
-		col := infinity.ColumnDefinition{
-			Name:     fieldName,
-			DataType: fieldInfo.Type,
-			Default:  fieldInfo.Default,
-			// Comment:  fieldInfo.Comment,
+		// Table exists, open it and check if vector column needs to be added
+		logger.Info("Table already exists, checking for vector column", zap.String("tableName", tableName))
+		table, err = db.GetTable(tableName)
+		if err != nil {
+			return fmt.Errorf("Failed to open existing table %s: %w", tableName, err)
 		}
-		columns = append(columns, &col)
-	}

-	// Add vector column
-	vectorColName := fmt.Sprintf("q_%d_vec", vecSize)
-	columns = append(columns, &infinity.ColumnDefinition{
-		Name:     vectorColName,
-		DataType: fmt.Sprintf("vector,%d,float", vecSize),
-	})
+		// Check if vector column exists (for embedding model changes)
+		colExists, err := e.columnExists(table, vectorColName)
+		if err != nil {
+			logger.Warn("Failed to check column existence", zap.String("column", vectorColName), zap.Error(err))
+		}

-	// Add chunk_data column for table parser
-	if parserID == "table" {
+		// Add new vector column if it doesn't exist (handles embedding model change)
+		if !colExists {
+			logger.Info("Adding new vector column for embedding model change", zap.String("column", vectorColName), zap.Int("size", vecSize))
+			addColSchema := infinity.TableSchema{
+				&infinity.ColumnDefinition{
+					Name:     vectorColName,
+					DataType: fmt.Sprintf("vector,%d,float", vecSize),
+				},
+			}
+			if _, err := table.AddColumns(addColSchema); err != nil {
+				logger.Error("Failed to add vector column "+vectorColName, err)
+				return fmt.Errorf("Failed to add vector column %s: %w", vectorColName, err)
+			}
+			logger.Info("Successfully added vector column", zap.String("column", vectorColName))
+		}
+	} else {
+		// Table doesn't exist, create it with vector column in the initial schema
+		logger.Info(fmt.Sprintf("Creating table with vector column: %s with dimension %d", vectorColName, vecSize))
+
+		// Build column definitions (preserving JSON order)
+		var columns infinity.TableSchema
+		for _, fieldName := range schema.Keys {
+			fieldInfo := schema.Fields[fieldName]
+			col := infinity.ColumnDefinition{
+				Name:     fieldName,
+				DataType: fieldInfo.Type,
+				Default:  fieldInfo.Default,
+				// Comment:  fieldInfo.Comment,
+			}
+			columns = append(columns, &col)
+		}
+
+		// Add vector column
 		columns = append(columns, &infinity.ColumnDefinition{
-			Name:     "chunk_data",
-			DataType: "json",
-			Default:  "{}",
+			Name:     vectorColName,
+			DataType: fmt.Sprintf("vector,%d,float", vecSize),
 		})
+
+		// Add chunk_data column for table parser
+		if parserID == "table" {
+			columns = append(columns, &infinity.ColumnDefinition{
+				Name:     "chunk_data",
+				DataType: "json",
+				Default:  "{}",
+			})
+		}
+
+		// Create table
+		table, err = db.CreateTable(tableName, columns, infinity.ConflictTypeIgnore)
+		if err != nil {
+			return fmt.Errorf("Failed to create table: %w", err)
+		}
+		logger.Debug("Infinity created table", zap.String("tableName", tableName))
 	}

-	// Create table
-	table, err := db.CreateTable(tableName, columns, infinity.ConflictTypeIgnore)
-	if err != nil {
-		return fmt.Errorf("Failed to create table: %w", err)
-	}
-	logger.Debug("Infinity created table", zap.String("tableName", tableName))
-
-	// Create HNSW index on vector column
+	// Create HNSW index on vector column with unique name based on vector size
+	// Use unique index name to avoid conflict when embedding model changes
+	vectorIndexName := fmt.Sprintf("q_%d_vec_idx", vecSize)
 	_, err = table.CreateIndex(
-		"q_vec_idx",
+		vectorIndexName,
 		infinity.NewIndexInfo(vectorColName, infinity.IndexTypeHnsw, map[string]string{
 			"M":               "16",
 			"ef_construction": "50",
@@ -119,8 +169,9 @@ func (e *infinityEngine) CreateDataset(ctx context.Context, indexName, datasetID
 		"",
 	)
 	if err != nil {
-		return fmt.Errorf("Failed to create HNSW index: %w", err)
+		return fmt.Errorf("Failed to create HNSW index %s: %w", vectorIndexName, err)
 	}
+	logger.Info("Created vector index", zap.String("indexName", vectorIndexName), zap.String("column", vectorColName))

 	// Create full-text indexes for varchar fields with analyzers
 	for _, fieldName := range schema.Keys {
--- a/internal/engine/infinity/document.go
+++ b/internal/engine/infinity/document.go
@@ -0,0 +1,239 @@
+//
+//  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 (
+	"context"
+	"fmt"
+	"strings"
+
+	"go.uber.org/zap"
+	"ragflow/internal/logger"
+)
+
+// IndexDocument indexes a single document
+// For skill index (tableName starts with "skill_"), uses InsertSkill
+// For regular document index, returns not implemented error
+func (e *infinityEngine) IndexDocument(ctx context.Context, tableName, docID string, doc interface{}) error {
+	// Check if this is a skill index
+	if strings.HasPrefix(tableName, "skill_") {
+		return e.InsertSkill(ctx, tableName, docID, doc)
+	}
+	return fmt.Errorf("infinity insert not implemented for regular documents: waiting for official Go SDK")
+}
+
+// InsertSkill inserts a skill document into skill index
+// Auto-creates the table if it doesn't exist
+func (e *infinityEngine) InsertSkill(ctx context.Context, tableName, docID string, doc interface{}) error {
+	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 {
+		// Table doesn't exist, try to create it
+		errMsg := strings.ToLower(err.Error())
+		if !strings.Contains(errMsg, "not found") && !strings.Contains(errMsg, "doesn't exist") {
+			return fmt.Errorf("failed to get table %s: %w", tableName, err)
+		}
+
+		// Cannot auto-create skill table without knowing the vector dimension
+		// The table should be created by SkillIndexerService.EnsureIndex before calling this
+		return fmt.Errorf("skill table %s does not exist, please ensure index is initialized first", tableName)
+	}
+
+	// Transform doc to map
+	docMap, ok := doc.(map[string]interface{})
+	if !ok {
+		return fmt.Errorf("invalid doc type, expected map[string]interface{}")
+	}
+
+	// Prepare insert data
+	insertDoc := make(map[string]interface{})
+	for k, v := range docMap {
+		insertDoc[k] = v
+	}
+	// Ensure skill_id is set (schema uses skill_id, not id)
+	insertDoc["skill_id"] = docID
+
+	// Delete existing document with same skill_id
+	// Escape single quotes to prevent filter injection
+	docIDEscaped := strings.ReplaceAll(docID, "'", "''")
+	filter := fmt.Sprintf("skill_id = '%s'", docIDEscaped)
+	delResp, delErr := table.Delete(filter)
+	if delErr != nil {
+		logger.Warn(fmt.Sprintf("Failed to delete existing skill document: %v", delErr))
+	} else if delResp.DeletedRows > 0 {
+		logger.Debug(fmt.Sprintf("Deleted %d existing skill document(s)", delResp.DeletedRows))
+	}
+
+	// Insert the document
+	_, err = table.Insert([]map[string]interface{}{insertDoc})
+	if err != nil {
+		return fmt.Errorf("failed to insert skill document into %s: %w", tableName, err)
+	}
+	return nil
+}
+
+// BulkIndex indexes documents in bulk
+// For skill index (tableName starts with "skill_"), uses BulkInsertSkill
+// For regular document index, returns not implemented error
+func (e *infinityEngine) BulkIndex(ctx context.Context, tableName string, docs []interface{}) (interface{}, error) {
+	// Check if this is a skill index
+	if strings.HasPrefix(tableName, "skill_") {
+		inserted, err := e.BulkInsertSkill(ctx, tableName, docs)
+		return &BulkResponse{Inserted: inserted}, err
+	}
+	return nil, fmt.Errorf("infinity bulk insert not implemented for regular documents: waiting for official Go SDK")
+}
+
+// BulkInsertSkill inserts multiple skill documents in bulk with upsert semantics.
+// For each document, deletes existing rows with the same skill_id before inserting,
+// matching the behavior of InsertSkill. Creates shallow copies of input maps to
+// avoid mutating caller data.
+func (e *infinityEngine) BulkInsertSkill(ctx context.Context, tableName string, docs []interface{}) (int, error) {
+	db, err := e.client.conn.GetDatabase(e.client.dbName)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get database: %w", err)
+	}
+
+	table, err := db.GetTable(tableName)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get table %s: %w", tableName, err)
+	}
+
+	// Collect skill_ids for upsert and create shallow copies of docs
+	skillIDs := make([]string, 0, len(docs))
+	insertDocs := make([]map[string]interface{}, 0, len(docs))
+
+	for _, doc := range docs {
+		docMap, ok := doc.(map[string]interface{})
+		if !ok {
+			logger.Warn("Invalid doc type in bulk insert, expected map[string]interface{}")
+			continue
+		}
+
+		// Create shallow copy to avoid mutating caller's map
+		insertDoc := make(map[string]interface{})
+		for k, v := range docMap {
+			insertDoc[k] = v
+		}
+
+		// Ensure skill_id is set if id or skill_id exists in doc
+		var skillID string
+		if id, hasID := docMap["id"]; hasID {
+			skillID = fmt.Sprintf("%v", id)
+			insertDoc["skill_id"] = skillID
+		} else if sid, hasSkillID := docMap["skill_id"]; hasSkillID {
+			skillID = fmt.Sprintf("%v", sid)
+		}
+
+		if skillID != "" {
+			skillIDs = append(skillIDs, skillID)
+		}
+		insertDocs = append(insertDocs, insertDoc)
+	}
+
+	if len(insertDocs) == 0 {
+		logger.Warn("No valid documents to bulk insert", zap.String("tableName", tableName))
+		return 0, nil
+	}
+
+	// Upsert: delete existing documents with same skill_ids before inserting
+	for _, skillID := range skillIDs {
+		// Escape single quotes to prevent filter injection
+		docIDEscaped := strings.ReplaceAll(skillID, "'", "''")
+		filter := fmt.Sprintf("skill_id = '%s'", docIDEscaped)
+		delResp, delErr := table.Delete(filter)
+		if delErr != nil {
+			logger.Warn("Failed to delete existing skill document before bulk insert",
+				zap.String("tableName", tableName),
+				zap.String("skill_id", skillID),
+				zap.Error(delErr))
+		} else if delResp.DeletedRows > 0 {
+			logger.Debug("Deleted existing skill document before bulk insert",
+				zap.String("tableName", tableName),
+				zap.String("skill_id", skillID),
+				zap.Int64("deletedRows", delResp.DeletedRows))
+		}
+	}
+
+	// Insert the documents
+	_, err = table.Insert(insertDocs)
+	if err != nil {
+		return 0, fmt.Errorf("failed to bulk insert skill documents: %w", err)
+	}
+
+	logger.Debug("Bulk upserted skill documents",
+		zap.String("tableName", tableName),
+		zap.Int("count", len(insertDocs)),
+		zap.Int("skillIDs", len(skillIDs)))
+	return len(insertDocs), nil
+}
+
+// BulkResponse bulk operation response
+type BulkResponse struct {
+	Inserted int
+}
+
+// GetDocument gets a document
+func (e *infinityEngine) GetDocument(ctx context.Context, tableName, docID string) (interface{}, error) {
+	return nil, fmt.Errorf("infinity get document not implemented: waiting for official Go SDK")
+}
+
+// DeleteDocument deletes a document by ID
+func (e *infinityEngine) DeleteDocument(ctx context.Context, tableName, docID string) error {
+	if tableName == "" {
+		return fmt.Errorf("table name cannot be empty")
+	}
+	if docID == "" {
+		return fmt.Errorf("document id cannot be empty")
+	}
+
+	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 table: %w", err)
+	}
+
+	// Use filter to delete document by ID
+	// Skill index uses 'skill_id', regular indices use 'id'
+	idField := "id"
+	if strings.HasPrefix(tableName, "skill_") {
+		idField = "skill_id"
+	}
+	// Escape single quotes to prevent filter injection
+	docIDEscaped := strings.ReplaceAll(docID, "'", "''")
+	filter := fmt.Sprintf("%s = '%s'", idField, docIDEscaped)
+	resp, err := table.Delete(filter)
+	if err != nil {
+		return fmt.Errorf("failed to delete document: %w", err)
+	}
+
+	logger.Debug("Deleted document from Infinity",
+		zap.String("tableName", tableName),
+		zap.String("docID", docID),
+		zap.String("idField", idField),
+		zap.Int64("deletedRows", resp.DeletedRows))
+
+	return nil
+}
--- a/internal/engine/infinity/search.go
+++ b/internal/engine/infinity/search.go
@@ -40,7 +40,7 @@ import (
 // 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) {
-	logger.Info("Search in Infinity started", zap.Any("indexNames", req.IndexNames))
+	logger.Debug("Search in Infinity started", zap.Any("indexNames", req.IndexNames))
 	if logger.IsDebugEnabled() {
 		// Format match expressions for logging
 		var matchExprsStr string
@@ -88,16 +88,27 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 	}

 	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
+		}
 	}

 	var outputColumns []string
 	if isMetadataTable {
 		outputColumns = []string{"id", "kb_id", "meta_fields"}
+	} else if isSkillIndex {
+		outputColumns = []string{
+			"skill_id", "space_id", "folder_id", "name", "tags", "description", "content",
+			"version", "status", "create_time", "update_time",
+		}
+		outputColumns = convertSelectFields(outputColumns, true)
 	} else {
 		outputColumns = []string{
 			"id", "doc_id", "kb_id", "content_ltks", "content_with_weight",
@@ -119,12 +130,24 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 				continue
 			}
 			switch e := expr.(type) {
+			case string:
+				if e != "" {
+					hasTextMatch = true
+					matchText = &types.MatchTextExpr{
+						MatchingText: e,
+						TopN:         pageSize,
+					}
+				}
 			case *types.MatchTextExpr:
-				hasTextMatch = true
-				matchText = e
+				if e.MatchingText != "" {
+					hasTextMatch = true
+					matchText = e
+				}
 			case *types.MatchDenseExpr:
-				hasVectorMatch = true
-				matchDense = e
+				if len(e.EmbeddingData) > 0 {
+					hasVectorMatch = true
+					matchDense = e
+				}
 			}
 		}
 	}
@@ -132,14 +155,21 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 	if hasTextMatch || hasVectorMatch {
 		if hasTextMatch {
 			outputColumns = append(outputColumns, "score()")
-		} else if hasVectorMatch {
+		}
+		// similarity() is only allowed by Infinity when there is ONLY MATCH VECTOR.
+		// When both text and vector matches exist (hybrid search with Fusion),
+		// only score() is valid — Fusion produces a unified SCORE column.
+		if hasVectorMatch && !hasTextMatch {
 			outputColumns = append(outputColumns, "similarity()")
 		}
-		if !slices.Contains(outputColumns, common.PAGERANK_FLD) {
-			outputColumns = append(outputColumns, common.PAGERANK_FLD)
-		}
-		if !slices.Contains(outputColumns, common.TAG_FLD) {
-			outputColumns = append(outputColumns, common.TAG_FLD)
+		// Skill index does not have pagerank_fea and tag_feas columns
+		if !isSkillIndex {
+			if !slices.Contains(outputColumns, common.PAGERANK_FLD) {
+				outputColumns = append(outputColumns, common.PAGERANK_FLD)
+			}
+			if !slices.Contains(outputColumns, common.TAG_FLD) {
+				outputColumns = append(outputColumns, common.TAG_FLD)
+			}
 		}
 	}

@@ -147,7 +177,7 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 		outputColumns = append(outputColumns, "row_id()")
 	}

-	outputColumns = convertSelectFields(outputColumns)
+	outputColumns = convertSelectFields(outputColumns, isSkillIndex)
 	if hasVectorMatch && matchDense != nil && matchDense.VectorColumnName != "" {
 		outputColumns = append(outputColumns, matchDense.VectorColumnName)
 	}
@@ -167,11 +197,21 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 		if req.Filter != nil {
 			if availInt, ok := req.Filter["available_int"]; ok {
 				filterParts = append(filterParts, fmt.Sprintf("available_int=%v", availInt))
+			} else if status, ok := req.Filter["status"]; ok {
+				filterParts = append(filterParts, fmt.Sprintf("status='%s'", status))
+			} else {
+				if isSkillIndex {
+					filterParts = append(filterParts, "status='1'")
+				} else {
+					filterParts = append(filterParts, "available_int=1")
+				}
+			}
+		} else {
+			if isSkillIndex {
+				filterParts = append(filterParts, "status='1'")
 			} else {
 				filterParts = append(filterParts, "available_int=1")
 			}
-		} else {
-			filterParts = append(filterParts, "available_int=1")
 		}
 	}

@@ -257,6 +297,13 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 			var textFields []string
 			if matchText != nil && len(matchText.Fields) > 0 {
 				textFields = matchText.Fields
+			} else if isSkillIndex {
+			textFields = []string{
+				"name^10",
+				"tags^5",
+				"description^3",
+				"content^1",
+			}
 			} else {
 				textFields = []string{
 					"title_tks^10",
@@ -339,12 +386,16 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 					vectorTopN = int(matchDense.TopN)
 				}

-				denseFilterStr := filterStr
-				if denseFilterStr == "" {
+			denseFilterStr := filterStr
+			if denseFilterStr == "" {
+				if isSkillIndex {
+					denseFilterStr = "status='1'"
+				} else {
 					denseFilterStr = "available_int=1"
 				}
+			}

-				if hasTextMatch {
+				if hasTextMatch && fusionExpr == nil {
 					fieldsStr := strings.Join(convertedFields, ",")
 					filterFulltext := fmt.Sprintf("filter_fulltext('%s', '%s')", fieldsStr, questionText)
 					denseFilterStr = fmt.Sprintf("(%s) AND %s", denseFilterStr, filterFulltext)
@@ -354,13 +405,11 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 					"filter":    denseFilterStr,
 				}

-				logger.Debug(fmt.Sprintf(
-					"MatchDenseExpr:\n"+
-						"    field=%s\n"+
-						"    topn=%d\n"+
-						"    extra_options=%v",
-					fieldName, vectorTopN, extraOptions,
-				))
+				logger.Debug("MatchDense for hybrid search",
+					zap.String("fieldName", fieldName),
+					zap.String("distanceType", distanceType),
+					zap.Int("topN", vectorTopN),
+					zap.Bool("hasFusion", fusionExpr != nil))

 				table = table.MatchDense(fieldName, vectorData, dataType, distanceType, vectorTopN, extraOptions)
 			}
@@ -380,13 +429,11 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 						fusionParams[k] = v
 					}
 				}
-				logger.Debug(fmt.Sprintf(
-					"FusionExpr:\n"+
-						"    method=%s\n"+
-						"    topn=%d\n"+
-						"    fusion_params=%v",
-					fusionMethod, fusionTopK, fusionParams,
-				))
+
+				logger.Debug("Applying Fusion for hybrid search",
+					zap.String("method", fusionMethod),
+					zap.Int("topN", fusionTopK),
+					zap.Any("params", fusionParams))

 				table = table.Fusion(fusionMethod, fusionTopK, fusionParams)
 			}
@@ -422,6 +469,12 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 			// Execute query
 			df, err := table.ToDataFrame()
 			if err != nil {
+				logger.Warn("Infinity query failed",
+					zap.String("tableName", tableName),
+					zap.Bool("hasTextMatch", hasTextMatch),
+					zap.Bool("hasVectorMatch", hasVectorMatch),
+					zap.Bool("hasFusion", fusionExpr != nil),
+					zap.Error(err))
 				continue
 			}

@@ -437,7 +490,19 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 			}

 			// Apply field name mapping and row_id handling
-			GetFields(chunks, nil)
+			// Skill index uses different schema
+			// so we skip the document-specific field mappings
+			if !isSkillIndex {
+				GetFields(chunks, nil)
+			} else {
+				// For skill index, only handle ROW_ID -> row_id() mapping
+				for _, chunk := range chunks {
+					if val, ok := chunk["ROW_ID"]; ok {
+						chunk["row_id()"] = val
+						delete(chunk, "ROW_ID")
+					}
+				}
+			}

 			// Parse total_hits_count from ExtraInfo
 			var tableTotal int64
@@ -462,12 +527,19 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (

 	if hasTextMatch || hasVectorMatch {
 		scoreColumn := ""
-		if hasTextMatch {
+		if hasTextMatch && hasVectorMatch {
+			scoreColumn = "SCORE"
+		} else if hasTextMatch {
 			scoreColumn = "SCORE"
 		} else if hasVectorMatch {
 			scoreColumn = "SIMILARITY"
 		}
-		allResults = calculateScores(allResults, scoreColumn)
+		pagerankField := common.PAGERANK_FLD
+		if isSkillIndex {
+			pagerankField = "" // Skill index has no pagerank field
+		}
+
+		allResults = calculateScores(allResults, scoreColumn, pagerankField)
 		allResults = sortByScore(allResults, len(allResults))
 	}

@@ -475,7 +547,7 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 		allResults = allResults[:pageSize]
 	}

-	logger.Info("Search in Infinity completed", zap.Any("indexNames", req.IndexNames), zap.Int("returnedRows", len(allResults)), zap.Int64("totalHits", totalHits))
+	logger.Debug("Search in Infinity completed", zap.Int("returnedRows", len(allResults)), zap.Int64("totalHits", totalHits))

 	return &types.SearchResult{
 		Chunks: allResults,
@@ -483,9 +555,9 @@ func (e *infinityEngine) Search(ctx context.Context, req *types.SearchRequest) (
 	}, nil
 }

-// convertSelectFields converts RAG field names to Infinity column names for SELECT (output_columns).
-// Example: docnm_kwd → docnm, content_ltks → content
-func convertSelectFields(output []string) []string {
+// convertSelectFields converts field names to Infinity format
+// isSkillIndex indicates if this is a skill index (uses skill_id instead of id)
+func convertSelectFields(output []string, isSkillIndex ...bool) []string {
 	fieldMapping := map[string]string{
 		"docnm_kwd":           "docnm",
 		"title_tks":           "docnm",
@@ -501,6 +573,11 @@ func convertSelectFields(output []string) []string {
 		"authors_sm_tks":      "authors",
 	}

+	skillIndex := false
+	if len(isSkillIndex) > 0 {
+		skillIndex = isSkillIndex[0]
+	}
+
 	needEmptyCount := false
 	for i, field := range output {
 		if field == "important_kwd" {
@@ -522,15 +599,20 @@ func convertSelectFields(output []string) []string {
 	}

 	// Add id and empty count if needed
+	// For skill index, use skill_id instead of id
 	hasID := false
+	idField := "id"
+	if skillIndex {
+		idField = "skill_id"
+	}
 	for _, f := range result {
-		if f == "id" {
+		if f == idField {
 			hasID = true
 			break
 		}
 	}
 	if !hasID {
-		result = append([]string{"id"}, result...)
+		result = append([]string{idField}, result...)
 	}

 	if needEmptyCount {
@@ -540,8 +622,10 @@ func convertSelectFields(output []string) []string {
 	return result
 }

-// convertMatchingField converts RAG field names to Infinity full-text index names for MATCH expressions.
-// Example: docnm_kwd → docnm@ft_docnm_rag_coarse, content_ltks → content@ft_content_rag_coarse
+// convertMatchingField converts field names for matching
+// For regular document indices: maps _tks/_kwd fields to column@index_name format
+// For skill indices: maps raw field names to column@index_name format
+// Infinity requires column@index_name when a column has multiple full-text indexes
 func convertMatchingField(fieldWeightStr string) string {
 	// Split on ^ to get field name
 	parts := strings.Split(fieldWeightStr, "^")
@@ -562,6 +646,11 @@ func convertMatchingField(fieldWeightStr string) string {
 		"authors_tks":         "authors@ft_authors_rag_coarse",
 		"authors_sm_tks":      "authors@ft_authors_rag_fine",
 		"tag_kwd":             "tag_kwd@ft_tag_kwd_whitespace__",
+		// Skill index fields
+		"name":               "name@ft_name_rag_coarse",
+		"tags":               "tags@ft_tags_rag_coarse",
+		"description":        "description@ft_description_rag_coarse",
+		"content":            "content@ft_content_rag_coarse",
 	}

 	if newField, ok := fieldMapping[field]; ok {
@@ -728,8 +817,8 @@ func equivalentConditionToStr(condition map[string]interface{}) string {
 	return strings.Join(cond, " AND ")
 }

-// calculateScores calculates _score = score_column + pagerank_fld
-func calculateScores(chunks []map[string]interface{}, scoreColumn string) []map[string]interface{} {
+// calculateScores calculates _score = score_column + pagerank
+func calculateScores(chunks []map[string]interface{}, scoreColumn, pagerankField string) []map[string]interface{} {
 	for i := range chunks {
 		score := 0.0
 		if scoreVal, ok := chunks[i][scoreColumn]; ok {
@@ -737,9 +826,11 @@ func calculateScores(chunks []map[string]interface{}, scoreColumn string) []map[
 				score += f
 			}
 		}
-		if prVal, ok := chunks[i][common.PAGERANK_FLD]; ok {
-			if f, ok := utility.ToFloat64(prVal); ok {
-				score += f
+		if pagerankField != "" {
+			if prVal, ok := chunks[i][pagerankField]; ok {
+				if f, ok := utility.ToFloat64(prVal); ok {
+					score += f
+				}
 			}
 		}
 		chunks[i]["_score"] = score
@@ -1008,4 +1099,4 @@ func (e *infinityEngine) GetHighlight(chunks []map[string]interface{}, keywords
 	}

 	return result
-}
+}