diff --git a/internal/engine/infinity/chunk.go b/internal/engine/infinity/chunk.go index da6098b88d..502c565046 100644 --- a/internal/engine/infinity/chunk.go +++ b/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_), 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)) diff --git a/internal/engine/infinity/chunk_transform_test.go b/internal/engine/infinity/chunk_transform_test.go index 9c92b573b5..a0193cee78 100644 --- a/internal/engine/infinity/chunk_transform_test.go +++ b/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") + } +} diff --git a/internal/engine/infinity/common.go b/internal/engine/infinity/common.go index 56238b78ea..71c4561356 100644 --- a/internal/engine/infinity/common.go +++ b/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 diff --git a/internal/engine/infinity/metadata.go b/internal/engine/infinity/metadata.go index 696c836ee0..8456707153 100644 --- a/internal/engine/infinity/metadata.go +++ b/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 diff --git a/internal/engine/infinity/sql_test.go b/internal/engine/infinity/sql_test.go index feaa07c98d..9c3be5faa8 100644 --- a/internal/engine/infinity/sql_test.go +++ b/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) + } +} diff --git a/internal/ingestion/knowledge_compile/wiki_map_store.go b/internal/ingestion/knowledge_compile/wiki_map_store.go index 85b32358f1..8ac6bc9bd3 100644 --- a/internal/ingestion/knowledge_compile/wiki_map_store.go +++ b/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] diff --git a/internal/ingestion/knowledge_compile/wiki_map_store_test.go b/internal/ingestion/knowledge_compile/wiki_map_store_test.go index 4d686817e8..6b8d521f9e 100644 --- a/internal/ingestion/knowledge_compile/wiki_map_store_test.go +++ b/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) + } +} diff --git a/internal/ingestion/task/knowledge_compiler_wiring.go b/internal/ingestion/task/knowledge_compiler_wiring.go index 9b9cf428c8..770e37ae36 100644 --- a/internal/ingestion/task/knowledge_compiler_wiring.go +++ b/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.