fix: support wiki map writes on Infinity (#20203)

internal/engine/infinity/chunk.go

@@ -133,6 +133,24 @@ func (e *Engine) createChunkStoreWithDB(db *infinity.Database, baseName, dataset
 			}
 			common.Info("Successfully added vector column", zap.String("column", vectorColName))
 		}
+
+		// A metadata-only Wiki MAP write can create the dataset table before a
+		// table-parser document arrives. Reconcile the parser-specific column on
+		// the existing-table path so creation order does not make table chunks
+		// unwritable.
+		if parserID == "table" {
+			chunkDataExists, checkErr := e.columnExists(table, "chunk_data")
+			if checkErr != nil {
+				return fmt.Errorf("failed to check chunk_data column: %w", checkErr)
+			}
+			if !chunkDataExists {
+				if _, addErr := table.AddColumns(infinity.TableSchema{
+					&infinity.ColumnDefinition{Name: "chunk_data", DataType: "json", Default: "{}"},
+				}); addErr != nil {
+					return fmt.Errorf("failed to add chunk_data column: %w", addErr)
+				}
+			}
+		}
 	} else {
 		// Table doesn't exist, create it with vector column in the initial schema
 		common.Info(fmt.Sprintf("Creating table with vector column: %s with dimension %d", vectorColName, vecSize))
@@ -856,21 +874,18 @@ func (e *Engine) Search(ctx context.Context, req *types.SearchRequest) (*types.S
 		}
 	}
 
-	// Build filter string from req.Filter
+	// Build the backend-independent portion of the filter. JSON-list predicates
+	// are completed per table below because legacy tables stored those columns
+	// as varchar while current tables use JSON.
+	var filterCopy map[string]interface{}
 	if req.Filter != nil {
-		filterCopy := make(map[string]interface{})
+		filterCopy = make(map[string]interface{})
 		for k, v := range req.Filter {
 			if k != "kb_id" {
 				filterCopy[k] = v
 			}
 		}
-
-		condStr := equivalentConditionToStr(filterCopy)
-		if condStr != "" {
-			filterParts = append(filterParts, condStr)
-		}
 	}
-	filterStr := strings.Join(filterParts, " AND ")
 
 	orderBy := req.OrderBy
 	var rankFeature map[string]float64
@@ -948,6 +963,25 @@ func (e *Engine) Search(ctx context.Context, req *types.SearchRequest) (*types.S
 			if err != nil {
 				continue
 			}
+			tableFilterParts := append([]string(nil), filterParts...)
+			if len(filterCopy) > 0 {
+				var tableColumns map[string]struct {
+					Type    string
+					Default interface{}
+				}
+				if hasJSONListFilter(filterCopy) {
+					tableColumns, err = loadTableColumns(tbl)
+					if err != nil {
+						common.Warn("Failed to load Infinity columns for JSON-list filter",
+							zap.String("tableName", tableName), zap.Error(err))
+						continue
+					}
+				}
+				if condition := equivalentConditionToStr(filterCopy, tableColumns); condition != "" {
+					tableFilterParts = append(tableFilterParts, condition)
+				}
+			}
+			filterStr := strings.Join(tableFilterParts, " AND ")
 			table := tbl.Output(outputColumns)
 
 			var textFields []string
@@ -1323,6 +1357,7 @@ func (e *Engine) GetChunk(ctx context.Context, tableName, chunkID string, datase
 	}
 
 	common.Debug("infinity get chunk", zap.String("chunkID", chunkID), zap.Any("tables", tableNames))
+	decodeJSONFields(chunk)
 
 	// Apply field mappings (same as in GetFields)
 	// docnm -> docnm_kwd, title_tks, title_sm_tks
@@ -1380,6 +1415,7 @@ func (e *Engine) GetChunk(ctx context.Context, tableName, chunkID string, datase
 // Used by Search() to mutate chunks with derived fields before returning.
 func applyFieldMappings(chunks []map[string]interface{}) {
 	for _, chunk := range chunks {
+		decodeJSONFields(chunk)
 		// docnm -> docnm_kwd, title_tks, title_sm_tks
 		if val, ok := chunk["docnm"].(string); ok {
 			chunk["docnm_kwd"] = val
@@ -1536,6 +1572,7 @@ func (e *Engine) GetFields(chunks []map[string]interface{}, fields []string) map
 	needImportantKwdEmptyCount := fieldsAll["important_kwd"]
 
 	for _, chunk := range chunks {
+		decodeJSONFields(chunk)
 		// Build column map for case-insensitive lookup (Python line 747)
 		columnMap := make(map[string]string)
 		for k := range chunk {
@@ -2231,7 +2268,10 @@ func floatsEqual(a, b float64) bool {
 }
 
 // equivalentConditionToStr converts a condition map to an Infinity filter string
-func equivalentConditionToStr(condition map[string]interface{}) string {
+func equivalentConditionToStr(condition map[string]interface{}, tableColumns map[string]struct {
+	Type    string
+	Default interface{}
+}) string {
 	if len(condition) == 0 {
 		return ""
 	}
@@ -2262,6 +2302,16 @@ func equivalentConditionToStr(condition map[string]interface{}) string {
 			continue
 		}
 
+		// JSON-list fields use member containment. Legacy varchar columns keep
+		// the old full-text behavior.
+		if fieldJSONList(k) && tableColumns != nil {
+			jsonConditions := jsonListFilterConditions(k, v, tableColumns)
+			if len(jsonConditions) > 0 {
+				cond = append(cond, "("+strings.Join(jsonConditions, " OR ")+")")
+			}
+			continue
+		}
+
 		// Handle keyword fields (using full-text filter)
 		if fieldKeyword(k) {
 			// For keyword fields, values are always treated as strings for filter_fulltext
@@ -2388,6 +2438,43 @@ func equivalentConditionToStr(condition map[string]interface{}) string {
 	return strings.Join(cond, " AND ")
 }
 
+// decodeJSONFields converts Infinity JSON-column strings back into their Go
+// array/object representation. Legacy list rows used ###-joined strings, so
+// preserve read compatibility for those columns.
+func decodeJSONFields(chunk map[string]interface{}) {
+	for fieldName, value := range chunk {
+		fieldNameLower := strings.ToLower(fieldName)
+		if !fieldJSON(fieldNameLower) {
+			continue
+		}
+		text, ok := value.(string)
+		if !ok {
+			continue
+		}
+		if text == "" {
+			if fieldJSONList(fieldNameLower) {
+				chunk[fieldName] = []interface{}{}
+			}
+			continue
+		}
+		var decoded interface{}
+		if err := json.Unmarshal([]byte(text), &decoded); err == nil {
+			chunk[fieldName] = decoded
+			continue
+		}
+		if fieldJSONList(fieldNameLower) {
+			parts := strings.Split(text, "###")
+			decodedList := make([]interface{}, 0, len(parts))
+			for _, part := range parts {
+				if part != "" {
+					decodedList = append(decodedList, part)
+				}
+			}
+			chunk[fieldName] = decodedList
+		}
+	}
+}
+
 // calculateScores calculates _score = score_column + pagerank
 func calculateScores(chunks []map[string]interface{}, scoreColumn, pagerankField string) []map[string]interface{} {
 	for i := range chunks {
@@ -2628,6 +2715,10 @@ func transformChunkFields(chunk map[string]interface{}, embeddingCols [][2]inter
 			if _, exists := chunk["question_kwd"]; !exists {
 				d["questions"] = utility.ConvertToString(v)
 			}
+		case "tenant_id":
+			// Infinity isolates tenants by table name (ragflow_<tenant>), so
+			// tenant_id is an internal routing field, not a table column.
+			continue
 		case "kb_id":
 			// 1. First check if it's a string
 			if str, ok := v.(string); ok {
@@ -2654,8 +2745,9 @@ func transformChunkFields(chunk map[string]interface{}, embeddingCols [][2]inter
 		case "chunk_data":
 			d["chunk_data"] = utility.ConvertMapToJSONString(v)
 		default:
-			// Check for *_feas fields
-			if strings.HasSuffix(k, "_feas") {
+			if fieldJSON(k) {
+				d[k] = encodeJSONField(v)
+			} else if strings.HasSuffix(k, "_feas") {
 				jsonBytes, _ := json.Marshal(v)
 				d[k] = string(jsonBytes)
 			} else if fieldKeyword(k) {
@@ -2697,6 +2789,17 @@ func transformChunkFields(chunk map[string]interface{}, embeddingCols [][2]inter
 	return d
 }
 
+func encodeJSONField(value interface{}) interface{} {
+	if text, ok := value.(string); ok && json.Valid([]byte(text)) {
+		return text
+	}
+	encoded, err := json.Marshal(value)
+	if err != nil {
+		return value
+	}
+	return string(encoded)
+}
+
 // DropChunkStore drops a chunk table from Infinity
 func (e *Engine) DropChunkStore(ctx context.Context, baseName, datasetID string) error {
 	return e.dropTable(ctx, buildChunkTableName(baseName, datasetID))

internal/engine/infinity/chunk_transform_test.go

@@ -143,3 +143,67 @@ func TestTransformChunkFieldsJoinsNativeKeywordSlice(t *testing.T) {
 		t.Errorf("important_kwd_empty_count = %v, want 1", got["important_kwd_empty_count"])
 	}
 }
+
+// TestTransformChunkFieldsJSONSlices serializes Go-native slices for Infinity
+// JSON columns. Passing []string directly to the SDK is rejected as an
+// unsupported slice element type.
+func TestTransformChunkFieldsJSONSlices(t *testing.T) {
+	got := transformChunkFields(map[string]interface{}{
+		"source_chunk_ids": []string{"chunk-1", "chunk-2"},
+		"source_doc_ids":   []string{"doc-1"},
+	}, nil)
+
+	if want := `["chunk-1","chunk-2"]`; got["source_chunk_ids"] != want {
+		t.Errorf("source_chunk_ids = %#v, want %q", got["source_chunk_ids"], want)
+	}
+	if want := `["doc-1"]`; got["source_doc_ids"] != want {
+		t.Errorf("source_doc_ids = %#v, want %q", got["source_doc_ids"], want)
+	}
+}
+
+func TestTransformChunkFieldsKeepsEncodedJSON(t *testing.T) {
+	got := transformChunkFields(map[string]interface{}{
+		"source_doc_ids": `["doc-1"]`,
+	}, nil)
+
+	if want := `["doc-1"]`; got["source_doc_ids"] != want {
+		t.Errorf("source_doc_ids = %#v, want %q", got["source_doc_ids"], want)
+	}
+}
+
+func TestDecodeJSONFields(t *testing.T) {
+	row := map[string]interface{}{
+		"source_doc_ids":      `["doc-1","doc-2"]`,
+		"aliases":             "alpha###beta",
+		"source_chunk_hashes": `{"chunk-1":"hash-1"}`,
+	}
+
+	decodeJSONFields(row)
+
+	if want := []interface{}{"doc-1", "doc-2"}; !reflect.DeepEqual(row["source_doc_ids"], want) {
+		t.Errorf("source_doc_ids = %#v, want %#v", row["source_doc_ids"], want)
+	}
+	if want := []interface{}{"alpha", "beta"}; !reflect.DeepEqual(row["aliases"], want) {
+		t.Errorf("aliases = %#v, want %#v", row["aliases"], want)
+	}
+	if want := map[string]interface{}{"chunk-1": "hash-1"}; !reflect.DeepEqual(row["source_chunk_hashes"], want) {
+		t.Errorf("source_chunk_hashes = %#v, want %#v", row["source_chunk_hashes"], want)
+	}
+}
+
+// TestTransformChunkFieldsDropsTenantRoutingField ensures the tenant routing
+// field is not sent to Infinity, whose per-tenant table schema does not have a
+// tenant_id column.
+func TestTransformChunkFieldsDropsTenantRoutingField(t *testing.T) {
+	got := transformChunkFields(map[string]interface{}{
+		"tenant_id": "tenant-1",
+		"kb_id":     "kb-1",
+	}, nil)
+
+	if _, ok := got["tenant_id"]; ok {
+		t.Fatal("tenant_id must not be sent to Infinity")
+	}
+	if got["kb_id"] != "kb-1" {
+		t.Fatalf("kb_id = %#v, want %q", got["kb_id"], "kb-1")
+	}
+}

internal/engine/infinity/common.go

@@ -158,6 +158,126 @@ func fieldKeyword(fieldName string) bool {
 	return false
 }
 
+// fieldJSON reports fields stored as Infinity JSON columns. The Infinity Go
+// SDK expects JSON columns as encoded strings, not Go slices or maps.
+func fieldJSON(fieldName string) bool {
+	switch fieldName {
+	case "source_chunk_ids", "source_doc_ids", "compilation_template_ids",
+		"doc_ids_kwd", "entity_names_kwd", "entity_names", "outlinks_kwd",
+		"related_kb_pages_kwd", "claims", "page_ids", "source_chunk_hashes",
+		"rechunked_from_chunk_ids", "aliases":
+		return true
+	default:
+		return false
+	}
+}
+
+// fieldJSONList reports JSON columns whose value is an array and whose filter
+// semantics are membership rather than whole-document equality.
+func fieldJSONList(fieldName string) bool {
+	switch fieldName {
+	case "source_chunk_ids", "source_doc_ids", "compilation_template_ids",
+		"doc_ids_kwd", "entity_names_kwd", "entity_names", "outlinks_kwd",
+		"related_kb_pages_kwd", "claims", "page_ids",
+		"rechunked_from_chunk_ids", "aliases":
+		return true
+	default:
+		return false
+	}
+}
+
+func jsonListFilterConditions(fieldName string, value interface{}, tableColumns map[string]struct {
+	Type    string
+	Default interface{}
+}) []string {
+	values := []interface{}{value}
+	switch typed := value.(type) {
+	case []string:
+		values = make([]interface{}, 0, len(typed))
+		for _, item := range typed {
+			values = append(values, item)
+		}
+	case []interface{}:
+		values = typed
+	}
+
+	column, found := tableColumns[fieldName]
+	if !found {
+		return []string{"1=0"}
+	}
+	columnType := ""
+	columnType = strings.ToLower(column.Type)
+	conditions := make([]string, 0, len(values))
+	for _, item := range values {
+		switch {
+		case strings.Contains(columnType, "json"):
+			literal, err := json.Marshal(item)
+			if err != nil {
+				continue
+			}
+			conditions = append(conditions, fmt.Sprintf("json_contains(%s, '%s')",
+				fieldName, strings.ReplaceAll(string(literal), "'", "''")))
+		case strings.Contains(columnType, "char"):
+			text, ok := item.(string)
+			if !ok || text == "" {
+				continue
+			}
+			conditions = append(conditions, fmt.Sprintf("filter_fulltext('%s', '%s')",
+				convertMatchingField(fieldName), strings.ReplaceAll(text, "'", "''")))
+		}
+	}
+	if len(conditions) == 0 {
+		return []string{"1=0"}
+	}
+	return conditions
+}
+
+func hasJSONListFilter(condition map[string]interface{}) bool {
+	for fieldName := range condition {
+		if fieldJSONList(fieldName) {
+			return true
+		}
+	}
+	return false
+}
+
+func loadTableColumns(table *infinity.Table) (map[string]struct {
+	Type    string
+	Default interface{}
+}, error) {
+	columns := make(map[string]struct {
+		Type    string
+		Default interface{}
+	})
+	response, err := table.ShowColumns()
+	if err != nil {
+		return nil, err
+	}
+	result, ok := response.(*infinity.QueryResult)
+	if !ok {
+		return nil, fmt.Errorf("unexpected response type: %T", response)
+	}
+	names := result.Data["name"]
+	types := result.Data["type"]
+	defaults := result.Data["default"]
+	for i, rawName := range names {
+		name, _ := rawName.(string)
+		columnType := ""
+		if i < len(types) {
+			columnType, _ = types[i].(string)
+		}
+		var defaultValue interface{}
+		if i < len(defaults) {
+			defaultValue = defaults[i]
+		}
+		columns[name] = struct {
+			Type    string
+			Default interface{}
+		}{Type: columnType, Default: defaultValue}
+	}
+	return columns, nil
+}
+
 // existsCondition builds a NOT EXISTS or field!=" condition
 func existsCondition(field string, tableColumns map[string]struct {
 	Type    string
@@ -208,6 +328,16 @@ func buildFilterFromCondition(condition map[string]interface{}, tableColumns map
 			continue
 		}
 
+		// JSON-list fields use member containment. Legacy tables stored these
+		// columns as ###-joined varchar values, so retain their full-text fallback.
+		if fieldJSONList(k) && tableColumns != nil {
+			jsonConditions := jsonListFilterConditions(k, v, tableColumns)
+			if len(jsonConditions) > 0 {
+				conditions = append(conditions, "("+strings.Join(jsonConditions, " OR ")+")")
+			}
+			continue
+		}
+
 		// Handle keyword fields -> filter_fulltext with converted field name
 		if fieldKeyword(k) {
 			var orConds []string

internal/engine/infinity/metadata.go

@@ -586,7 +586,7 @@ func (e *Engine) SearchMetadata(ctx context.Context, req *types.SearchMetadataRe
 	// Build filter from req.Filter
 	var filterStr string
 	if req.Filter != nil {
-		filterStr = equivalentConditionToStr(req.Filter)
+		filterStr = equivalentConditionToStr(req.Filter, nil)
 	}
 
 	// Get database and table

internal/engine/infinity/sql_test.go

@@ -424,3 +424,48 @@ func TestBuildFilterFromCondition_StringSliceIDPreservesScope(t *testing.T) {
 		t.Errorf("empty []string id: got %q, want '1=1'", got)
 	}
 }
+
+func TestBuildFilterFromConditionJSONListMembership(t *testing.T) {
+	jsonColumns := map[string]struct {
+		Type    string
+		Default interface{}
+	}{
+		"source_doc_ids": {Type: "Json", Default: "[]"},
+	}
+	condition := map[string]interface{}{"source_doc_ids": []string{"doc-1", "doc-2"}}
+	want := `(json_contains(source_doc_ids, '"doc-1"') OR json_contains(source_doc_ids, '"doc-2"'))`
+	if got := buildFilterFromCondition(condition, jsonColumns); got != want {
+		t.Errorf("JSON-list condition = %q, want %q", got, want)
+	}
+	if got := equivalentConditionToStr(condition, jsonColumns); got != want {
+		t.Errorf("JSON-list search condition = %q, want %q", got, want)
+	}
+}
+
+func TestBuildFilterFromConditionLegacyJSONListMembership(t *testing.T) {
+	legacyColumns := map[string]struct {
+		Type    string
+		Default interface{}
+	}{
+		"source_doc_ids": {Type: "Varchar", Default: ""},
+	}
+	condition := map[string]interface{}{"source_doc_ids": []string{"doc-1"}}
+	want := `(filter_fulltext('source_doc_ids', 'doc-1'))`
+	if got := buildFilterFromCondition(condition, legacyColumns); got != want {
+		t.Errorf("legacy JSON-list condition = %q, want %q", got, want)
+	}
+	if got := equivalentConditionToStr(condition, legacyColumns); got != want {
+		t.Errorf("legacy JSON-list search condition = %q, want %q", got, want)
+	}
+}
+
+func TestBuildFilterFromConditionMissingJSONListColumnNeverMatches(t *testing.T) {
+	condition := map[string]interface{}{"source_doc_ids": []string{"doc-1"}}
+	want := `(1=0)`
+	if got := buildFilterFromCondition(condition, map[string]struct {
+		Type    string
+		Default interface{}
+	}{}); got != want {
+		t.Errorf("missing JSON-list column condition = %q, want %q", got, want)
+	}
+}

internal/ingestion/knowledge_compile/wiki_map_store.go

@@ -34,7 +34,8 @@ const (
 )
 
 type wikiMapVersionStore struct {
-	engine engine.DocEngine
+	engine            engine.DocEngine
+	resolveVectorSize func(context.Context) (int, error)
 }
 
 func (s *wikiMapVersionStore) GetWikiMapActiveState(ctx context.Context, tenantID, datasetID, key string) ([]byte, error) {
@@ -69,6 +70,9 @@ func (s *wikiMapVersionStore) PutWikiMapActiveState(ctx context.Context, state k
 	if state.Key == "" || state.TenantID == "" || state.DatasetID == "" || state.DocumentID == "" {
 		return fmt.Errorf("save Wiki MAP active state: key and scope are required")
 	}
+	if err := s.ensureInfinityChunkStore(ctx, state.TenantID, state.DatasetID); err != nil {
+		return err
+	}
 	row := map[string]interface{}{
 		"id":                  state.Key,
 		"doc_id":              "wiki_map_active:" + state.DocumentID,
@@ -91,6 +95,48 @@ func NewWikiMapVersionStore(docEngine engine.DocEngine) kccommon.WikiMapVersionS
 	return &wikiMapVersionStore{engine: docEngine}
 }
 
+// NewWikiMapVersionStoreWithVectorSizeResolver creates a Wiki MAP store that
+// can initialize an Infinity chunk table before inserting metadata-only MAP
+// rows. Elasticsearch does not need this resolver because it can create its
+// mapping dynamically.
+func NewWikiMapVersionStoreWithVectorSizeResolver(
+	docEngine engine.DocEngine,
+	resolveVectorSize func(context.Context) (int, error),
+) kccommon.WikiMapVersionStore {
+	return &wikiMapVersionStore{
+		engine:            docEngine,
+		resolveVectorSize: resolveVectorSize,
+	}
+}
+
+func (s *wikiMapVersionStore) ensureInfinityChunkStore(ctx context.Context, tenantID, datasetID string) error {
+	if engine.Type(s.engine) != engine.EngineInfinity {
+		return nil
+	}
+	baseName := fmt.Sprintf("ragflow_%s", tenantID)
+	exists, err := s.engine.ChunkStoreExists(ctx, baseName, datasetID)
+	if err != nil {
+		return fmt.Errorf("check Infinity chunk store for Wiki MAP: %w", err)
+	}
+	if exists {
+		return nil
+	}
+	if s.resolveVectorSize == nil {
+		return fmt.Errorf("initialize Infinity chunk store for Wiki MAP: vector-size resolver is not configured")
+	}
+	vectorSize, err := s.resolveVectorSize(ctx)
+	if err != nil {
+		return fmt.Errorf("resolve Wiki MAP vector size: %w", err)
+	}
+	if vectorSize <= 0 {
+		return fmt.Errorf("resolve Wiki MAP vector size: got %d", vectorSize)
+	}
+	if err := s.engine.CreateChunkStore(ctx, baseName, datasetID, vectorSize, ""); err != nil {
+		return fmt.Errorf("initialize Infinity chunk store for Wiki MAP: %w", err)
+	}
+	return nil
+}
+
 func (s *wikiMapVersionStore) GetWikiMapVersions(ctx context.Context, tenantID, datasetID string, keys []string) (map[string][]byte, error) {
 	out := make(map[string][]byte, len(keys))
 	if len(keys) == 0 {
@@ -153,6 +199,9 @@ func (s *wikiMapVersionStore) PutWikiMapVersions(ctx context.Context, versions [
 		byScope[scope] = append(byScope[scope], version)
 	}
 	for _, scopedVersions := range byScope {
+		if err := s.ensureInfinityChunkStore(ctx, scopedVersions[0].TenantID, scopedVersions[0].DatasetID); err != nil {
+			return err
+		}
 		for start := 0; start < len(scopedVersions); start += wikiMapStoreBatchSize {
 			end := min(start+wikiMapStoreBatchSize, len(scopedVersions))
 			batch := scopedVersions[start:end]

internal/ingestion/knowledge_compile/wiki_map_store_test.go

@@ -18,6 +18,7 @@ package knowledge_compile
 import (
 	"context"
 	"reflect"
+	"strings"
 	"testing"
 
 	"ragflow/internal/engine/types"
@@ -26,10 +27,34 @@ import (
 
 type wikiMapStoreEngine struct {
 	fakeEngine
-	rows          map[string]map[string]interface{}
-	inserted      []map[string]interface{}
-	insertBase    string
-	insertDataset string
+	rows              map[string]map[string]interface{}
+	inserted          []map[string]interface{}
+	insertBase        string
+	insertDataset     string
+	engineType        string
+	chunkStoreExists  bool
+	chunkStoreChecks  int
+	createdStores     int
+	createdVectorSize int
+}
+
+func (e *wikiMapStoreEngine) GetType() string {
+	if e.engineType != "" {
+		return e.engineType
+	}
+	return e.fakeEngine.GetType()
+}
+
+func (e *wikiMapStoreEngine) ChunkStoreExists(context.Context, string, string) (bool, error) {
+	e.chunkStoreChecks++
+	return e.chunkStoreExists, nil
+}
+
+func (e *wikiMapStoreEngine) CreateChunkStore(_ context.Context, _, _ string, vectorSize int, _ string) error {
+	e.createdStores++
+	e.createdVectorSize = vectorSize
+	e.chunkStoreExists = true
+	return nil
 }
 
 func (e *wikiMapStoreEngine) Search(_ context.Context, req *types.SearchRequest) (*types.SearchResult, error) {
@@ -158,3 +183,59 @@ func TestWikiMapVersionStoreUsesNonSearchableDocStoreRows(t *testing.T) {
 		t.Fatalf("search KbIDs = %#v", engine.lastSearchReq)
 	}
 }
+
+func TestWikiMapVersionStoreSkipsExistingInfinityStoreBootstrap(t *testing.T) {
+	engine := &wikiMapStoreEngine{
+		rows:             map[string]map[string]interface{}{},
+		engineType:       "infinity",
+		chunkStoreExists: true,
+	}
+	resolverCalls := 0
+	store := NewWikiMapVersionStoreWithVectorSizeResolver(engine, func(context.Context) (int, error) {
+		resolverCalls++
+		return 0, nil
+	})
+	version := kccommon.WikiMapVersion{
+		Key: "version-a", TenantID: "tenant-1", DatasetID: "kb-1",
+		DocumentID: "doc-1", ChunkID: "chunk-1", Payload: []byte(`{}`),
+	}
+
+	if err := store.PutWikiMapVersions(t.Context(), []kccommon.WikiMapVersion{version}); err != nil {
+		t.Fatalf("PutWikiMapVersions() error = %v", err)
+	}
+	if resolverCalls != 0 || engine.createdStores != 0 || engine.chunkStoreChecks != 1 {
+		t.Fatalf("bootstrap calls: resolver=%d create=%d checks=%d", resolverCalls, engine.createdStores, engine.chunkStoreChecks)
+	}
+}
+
+func TestWikiMapVersionStoreCreatesMissingInfinityStore(t *testing.T) {
+	engine := &wikiMapStoreEngine{rows: map[string]map[string]interface{}{}, engineType: "infinity"}
+	store := NewWikiMapVersionStoreWithVectorSizeResolver(engine, func(context.Context) (int, error) {
+		return 768, nil
+	})
+	version := kccommon.WikiMapVersion{
+		Key: "version-a", TenantID: "tenant-1", DatasetID: "kb-1",
+		DocumentID: "doc-1", ChunkID: "chunk-1", Payload: []byte(`{}`),
+	}
+
+	if err := store.PutWikiMapVersions(t.Context(), []kccommon.WikiMapVersion{version}); err != nil {
+		t.Fatalf("PutWikiMapVersions() error = %v", err)
+	}
+	if engine.createdStores != 1 || engine.createdVectorSize != 768 {
+		t.Fatalf("created store: count=%d vector_size=%d", engine.createdStores, engine.createdVectorSize)
+	}
+}
+
+func TestWikiMapActiveStateRequiresResolverOnlyWhenInfinityStoreMissing(t *testing.T) {
+	engine := &wikiMapStoreEngine{rows: map[string]map[string]interface{}{}, engineType: "infinity"}
+	store := NewWikiMapVersionStore(engine).(kccommon.WikiMapActiveStateStore)
+	state := kccommon.WikiMapActiveState{
+		Key: "active-a", TenantID: "tenant-1", DatasetID: "kb-1",
+		DocumentID: "doc-1", Payload: []byte(`{}`),
+	}
+
+	err := store.PutWikiMapActiveState(t.Context(), state)
+	if err == nil || !strings.Contains(err.Error(), "vector-size resolver is not configured") {
+		t.Fatalf("PutWikiMapActiveState() error = %v", err)
+	}
+}

internal/ingestion/task/knowledge_compiler_wiring.go

@@ -539,11 +539,26 @@ func newKnowledgeCompilerDepsResolver() kc.DepsResolver {
 			llmMaxOutput = modelTarget.MaxTokens
 		}
 
+		embedder := &kcEmbedder{svc: svc, solver: modelSolver, tenantID: tenantID, embdID: embeddingModel}
+		mapStore := knowledge_compile.NewWikiMapVersionStoreWithVectorSizeResolver(engine.Get(), func(ctx context.Context) (int, error) {
+			if dimension := embedder.Dimensions(); dimension > 0 {
+				return dimension, nil
+			}
+			vectors, err := embedder.Encode(ctx, []string{"ok"})
+			if err != nil {
+				return 0, err
+			}
+			if len(vectors) == 0 || len(vectors[0]) == 0 {
+				return 0, fmt.Errorf("embedding returned an empty vector")
+			}
+			return len(vectors[0]), nil
+		})
+
 		return kc.Deps{
 			Chat:            &kcChatInvoker{svc: svc, tenantID: tenantID, llmID: llmID},
-			Embed:           &kcEmbedder{svc: svc, solver: modelSolver, tenantID: tenantID, embdID: embeddingModel},
+			Embed:           embedder,
 			WikiPages:       &kcWikiPageStore{docEngine: engine.Get()},
-			WikiMapVersions: knowledge_compile.NewWikiMapVersionStore(engine.Get()),
+			WikiMapVersions: mapStore,
 			// HistoricalKNN / Redis are optional (wiki historical dedup,
 			// datasetnav lock). They are wired separately when the
 			// surrounding pipeline supplies the backing services.