From 07d1c89e5ea8ff7ca8488462f7b2ea48e4019cc9 Mon Sep 17 00:00:00 2001 From: Jack Date: Wed, 5 Aug 2026 09:46:48 +0800 Subject: [PATCH] refactor(ingestion): own kb_id at the engine write boundary (drop producer stamp) (#17818) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem During ingestion, `indexdoc.ProcessChunksForPipeline` stamped `ck["kb_id"]` on every chunk. This was both: - **a dead write** — `elasticsearch.InsertChunks` unconditionally overwrites the value with `datasetID` (`chunk.go:211`), so the producer's value never reached the index; - **the wrong shape** — it was emitted as `[]string`, while both engines actually need a single string. This is the `kb_id` slice of the ingestion -> engine schema leak tracked in #17371: ingestion was carrying index-physical schema knowledge it should not own. ## Fix Make the search engines the single owner of `kb_id` at the write boundary, and stop ingestion from emitting it: - **Elasticsearch** (`chunk.go:211`) already sets `docCopy["kb_id"] = datasetID` — unchanged. - **Infinity** (`chunk.go`) `InsertChunks` now stamps `insertChunks[i]["kb_id"] = datasetID` right after `transformChunkFields` (previously it only *read/normalized* the producer value, which forced ingestion to supply it). Both engines are now consistent. - `ProcessChunksForPipeline` no longer stamps `kb_id` and the now-leaky `kbID` parameter is removed. The same removal is propagated to `ProcessPipelineOutputForGolden` and the `compare_pipeline_golden` dev tool (its `-kb-id` flag is dropped). The stored `kb_id` value is byte-for-byte unchanged: `datasetID` passed to `InsertChunks` is `taskCtx.Doc.KbID`, i.e. the same id that was previously set on the producer chunk. ## Verification - `bash build.sh --test ./internal/ingestion/task/indexdoc/... ./internal/engine/infinity/...` — both green. - `internal/ingestion/task` has **two pre-existing** failures (`TestPipelineExecutor_Run_RealCanvasDSL_UsesGeneralPipeline`, `TestRunPipeline_RealPipelineOutput_ProducesIndexFields`) that assert `inserted chunk count = 1, want 2` — a parser/assertion mismatch (the Go parser merges the 2-paragraph fixture into 1 chunk). They are unrelated to this change, which never touches chunk counting. The `kb_id`-related test failure this change would otherwise introduce is fixed by updating the tests below. - Updated the pinning unit test: `TestProcessChunksForPipeline_SetsDocID` (formerly `...SetsDocIDAndKBID`) now asserts `kb_id` is **not** set by the producer. Removed the `kb_id` assertion and the now-dead `taskChunkFieldEqualsStr` helper from `pipeline_real_integration_test.go`. ## Scope This closes only the `kb_id` portion of #17371. The remaining index-physical fields (`docnm_kwd`, `create_timestamp_flt`, `page_num_int`/`top_int`/ `position_int`, etc.) are intentionally left for a follow-up (P2). --- internal/engine/infinity/chunk.go | 4 +++ internal/ingestion/task/golden_compare.go | 3 +- .../ingestion/task/golden_compare_test.go | 4 +-- internal/ingestion/task/indexdoc/process.go | 7 ++-- .../ingestion/task/indexdoc/process_test.go | 34 ++++++++++--------- internal/ingestion/task/pipeline_executor.go | 1 - .../ingestion/task/pipeline_executor_test.go | 2 +- .../task/pipeline_real_integration_test.go | 18 ---------- .../task/tool/compare_pipeline_golden.go | 7 ++-- 9 files changed, 34 insertions(+), 46 deletions(-) diff --git a/internal/engine/infinity/chunk.go b/internal/engine/infinity/chunk.go index 17963ffb94..5bfa040918 100644 --- a/internal/engine/infinity/chunk.go +++ b/internal/engine/infinity/chunk.go @@ -353,6 +353,10 @@ func (e *Engine) InsertChunks(ctx context.Context, chunks []map[string]interface insertChunks := make([]map[string]interface{}, len(chunks)) for i, chunk := range chunks { insertChunks[i] = transformChunkFields(chunk, embeddingCols) + // kb_id is owned by the engine at the write boundary (mirrors ES + // chunk.go InsertChunks). The ingestion producer no longer stamps it, + // so the producer value (if any) is intentionally overridden here. + insertChunks[i]["kb_id"] = datasetID } // Delete existing rows with matching IDs diff --git a/internal/ingestion/task/golden_compare.go b/internal/ingestion/task/golden_compare.go index 90252122cb..d5f94b89e1 100644 --- a/internal/ingestion/task/golden_compare.go +++ b/internal/ingestion/task/golden_compare.go @@ -34,7 +34,6 @@ type GoldenCompareResult struct { func ProcessPipelineOutputForGolden( pipelineOutput map[string]any, docID string, - kbID string, docName string, ) (GoldenCompareResult, error) { normalized := indexdoc.NormalizeChunks(pipelineOutput) @@ -43,7 +42,7 @@ func ProcessPipelineOutputForGolden( } processed := indexdoc.DeepCopyChunks(normalized) - metadata, err := indexdoc.ProcessChunksForPipeline(processed, docID, kbID, docName, time.Now()) + metadata, err := indexdoc.ProcessChunksForPipeline(processed, docID, docName, time.Now()) if err != nil { return GoldenCompareResult{}, err } diff --git a/internal/ingestion/task/golden_compare_test.go b/internal/ingestion/task/golden_compare_test.go index 2ebc5a9004..96f9e7c36d 100644 --- a/internal/ingestion/task/golden_compare_test.go +++ b/internal/ingestion/task/golden_compare_test.go @@ -12,7 +12,7 @@ func TestProcessPipelineOutputForGolden_Markdown(t *testing.T) { "markdown": "# Title\n\nContent", } - result, err := ProcessPipelineOutputForGolden(input, "doc-1", "kb-1", "sample.md") + result, err := ProcessPipelineOutputForGolden(input, "doc-1", "sample.md") if err != nil { t.Fatalf("ProcessPipelineOutputForGolden: %v", err) } @@ -46,7 +46,7 @@ func TestProcessChunksForPipeline_StableFields(t *testing.T) { }, } - meta, err := indexdoc.ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "sample.md", now) + meta, err := indexdoc.ProcessChunksForPipeline(chunks, "doc-1", "sample.md", now) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } diff --git a/internal/ingestion/task/indexdoc/process.go b/internal/ingestion/task/indexdoc/process.go index 867741fbbd..511d5f8502 100644 --- a/internal/ingestion/task/indexdoc/process.go +++ b/internal/ingestion/task/indexdoc/process.go @@ -61,10 +61,14 @@ func GetEmbeddingTokenConsumption(output map[string]any) int { // collapse every such chunk onto the same empty-text ChunkID, silently // overwriting each other in the index. The caller fails the task so the // violation surfaces instead of corrupting the index. +// +// Note: kb_id is intentionally NOT stamped here. It is an index-physical +// concern owned by the search engine at the write boundary (ES chunk.go +// InsertChunks and Infinity chunk.go InsertChunks both set kb_id = datasetID), +// so ingestion never carries the index schema for it. See issue #17371. func ProcessChunksForPipeline( chunks []map[string]any, docID string, - kbID string, docName string, now time.Time, ) (map[string]any, error) { @@ -77,7 +81,6 @@ func ProcessChunksForPipeline( for _, ck := range chunks { ck["doc_id"] = docID - ck["kb_id"] = kbID ck["docnm_kwd"] = docName ck["create_time"] = timeStr ck["create_timestamp_flt"] = timestamp diff --git a/internal/ingestion/task/indexdoc/process_test.go b/internal/ingestion/task/indexdoc/process_test.go index 26e633c3d6..168f040098 100644 --- a/internal/ingestion/task/indexdoc/process_test.go +++ b/internal/ingestion/task/indexdoc/process_test.go @@ -43,9 +43,9 @@ func TestRenameTextToContentWithWeight_NoTextKey(t *testing.T) { // ProcessChunksForPipeline - Python: processChunks() // ============================================================================= -func TestProcessChunksForPipeline_SetsDocIDAndKBID(t *testing.T) { +func TestProcessChunksForPipeline_SetsDocID(t *testing.T) { chunks := []map[string]any{{"text": "hello world"}} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -53,14 +53,16 @@ func TestProcessChunksForPipeline_SetsDocIDAndKBID(t *testing.T) { if chunks[0]["doc_id"] != "doc-1" { t.Errorf("doc_id = %q, want \"doc-1\"", chunks[0]["doc_id"]) } - if kbID, ok := chunks[0]["kb_id"].(string); !ok || kbID != "kb-1" { - t.Errorf("kb_id = %v, want \"kb-1\" (string)", chunks[0]["kb_id"]) + // kb_id is intentionally NOT set here: it is owned by the search engine at + // the write boundary (ES/Infinity InsertChunks), not by ingestion. See #17371. + if _, exists := chunks[0]["kb_id"]; exists { + t.Errorf("kb_id should not be set by ProcessChunksForPipeline, got %v", chunks[0]["kb_id"]) } } func TestProcessChunksForPipeline_SetsDocNameKwd(t *testing.T) { chunks := []map[string]any{{"text": "hello"}} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -72,7 +74,7 @@ func TestProcessChunksForPipeline_SetsDocNameKwd(t *testing.T) { func TestProcessChunksForPipeline_SetsTimeFields(t *testing.T) { now := time.Now() chunks := []map[string]any{{"text": "hello"}} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", now) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", now) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -96,7 +98,7 @@ func TestProcessChunksForPipeline_SetsTimeFields(t *testing.T) { func TestProcessChunksForPipeline_GeneratesID(t *testing.T) { chunks := []map[string]any{{"text": "hello"}} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -112,7 +114,7 @@ func TestProcessChunksForPipeline_GeneratesID(t *testing.T) { // component.ChunkID computes a valid id from empty text, rather than erroring. func TestProcessChunksForPipeline_GeneratesIDOnNonStringText(t *testing.T) { chunks := []map[string]any{{"text": []any{"bad-shape"}}} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -134,7 +136,7 @@ func TestProcessChunksForPipeline_RemovesInternalPipelineFields(t *testing.T) { "_pdf_positions": []any{[]any{0, 1, 2, 3, 4}}, }} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -145,7 +147,7 @@ func TestProcessChunksForPipeline_RemovesInternalPipelineFields(t *testing.T) { func TestProcessChunksForPipeline_PreservesExistingID(t *testing.T) { chunks := []map[string]any{{"text": "hello", "id": "existing-id"}} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -156,7 +158,7 @@ func TestProcessChunksForPipeline_PreservesExistingID(t *testing.T) { func TestProcessChunksForPipeline_QuestionsProcessing(t *testing.T) { chunks := []map[string]any{{"text": "hello", "questions": "Q1\nQ2\nQ3"}} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -178,7 +180,7 @@ func TestProcessChunksForPipeline_QuestionsProcessing(t *testing.T) { func TestProcessChunksForPipeline_KeywordsProcessing(t *testing.T) { chunks := []map[string]any{{"text": "hello", "keywords": "kw1,kw2;kw3"}} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -197,7 +199,7 @@ func TestProcessChunksForPipeline_KeywordsProcessing(t *testing.T) { func TestProcessChunksForPipeline_SummaryProcessing(t *testing.T) { chunks := []map[string]any{{"text": "hello", "summary": "This is a summary."}} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -231,7 +233,7 @@ func TestProcessChunksForPipeline_PreservesTokenizerProducedFields(t *testing.T) "content_ltks": "tokenizer-output-ltks", "content_sm_ltks": "tokenizer-output-smltks", }} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -266,7 +268,7 @@ func TestProcessChunksForPipeline_PreservesTokenizerProducedFields(t *testing.T) func TestProcessChunksForPipeline_TextRenamed(t *testing.T) { chunks := []map[string]any{{"text": "hello world"}} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } @@ -281,7 +283,7 @@ func TestProcessChunksForPipeline_TextRenamed(t *testing.T) { func TestProcessChunksForPipeline_PreservesContentWithWeight(t *testing.T) { chunks := []map[string]any{{"content_with_weight": "already set", "text": "hello"}} - _, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now()) + _, err := ProcessChunksForPipeline(chunks, "doc-1", "test-doc.pdf", time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } diff --git a/internal/ingestion/task/pipeline_executor.go b/internal/ingestion/task/pipeline_executor.go index e976efaa22..8316db14f4 100644 --- a/internal/ingestion/task/pipeline_executor.go +++ b/internal/ingestion/task/pipeline_executor.go @@ -227,7 +227,6 @@ func (s *PipelineExecutor) processOutput(ctx context.Context, pipelineOutput map metadata, err := indexdoc.ProcessChunksForPipeline( chunks, s.taskCtx.Doc.ID, - s.taskCtx.Doc.KbID, *s.taskCtx.Doc.Name, time.Now(), ) diff --git a/internal/ingestion/task/pipeline_executor_test.go b/internal/ingestion/task/pipeline_executor_test.go index 76609ba8d4..b366e90466 100644 --- a/internal/ingestion/task/pipeline_executor_test.go +++ b/internal/ingestion/task/pipeline_executor_test.go @@ -202,7 +202,7 @@ func TestKB_Doc_Tenant_Accessors(t *testing.T) { func TestPipelineExecutor_ProcessChunks_WrapsProcessChunksForPipeline(t *testing.T) { svc := mustNewPipelineExecutor(t, makeTaskCtx(), "flow-1", 0) chunks := []map[string]any{{"text": "hello world"}} - meta, err := indexdoc.ProcessChunksForPipeline(chunks, svc.taskCtx.Doc.ID, svc.taskCtx.Doc.KbID, *svc.taskCtx.Doc.Name, time.Now()) + meta, err := indexdoc.ProcessChunksForPipeline(chunks, svc.taskCtx.Doc.ID, *svc.taskCtx.Doc.Name, time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } diff --git a/internal/ingestion/task/pipeline_real_integration_test.go b/internal/ingestion/task/pipeline_real_integration_test.go index 5ade785ac6..cc6ca70719 100644 --- a/internal/ingestion/task/pipeline_real_integration_test.go +++ b/internal/ingestion/task/pipeline_real_integration_test.go @@ -251,9 +251,6 @@ func TestPipelineExecutor_Run_RealPDF_ProducesIndexedChunks(t *testing.T) { if got := chunk["doc_id"]; got != docID { t.Fatalf("chunk[%d].doc_id = %v, want %q", i, got, docID) } - if !taskChunkFieldEqualsStr(chunk["kb_id"], kbID) { - t.Fatalf("chunk[%d].kb_id = %v, want %q", i, chunk["kb_id"], kbID) - } if got := chunk["docnm_kwd"]; got != docName { t.Fatalf("chunk[%d].docnm_kwd = %v, want %q", i, got, docName) } @@ -718,18 +715,3 @@ func taskS3SafeBucketName(s string) string { s = strings.ReplaceAll(s, "_", "-") return s } - -// taskChunkFieldEqualsStr compares a chunk field to a plain string, tolerating -// either form a producer may emit: a plain string (e.g. kb_id is "kb-1" after -// T1) or a single-element slice that survives a JSON round-trip as []any{"kb-1"}. -func taskChunkFieldEqualsStr(v any, want string) bool { - switch val := v.(type) { - case string: - return val == want - case []string: - return len(val) == 1 && val[0] == want - case []any: - return len(val) == 1 && fmt.Sprint(val[0]) == want - } - return false -} diff --git a/internal/ingestion/task/tool/compare_pipeline_golden.go b/internal/ingestion/task/tool/compare_pipeline_golden.go index f97542d180..1dc85f949a 100644 --- a/internal/ingestion/task/tool/compare_pipeline_golden.go +++ b/internal/ingestion/task/tool/compare_pipeline_golden.go @@ -17,7 +17,6 @@ import ( func main() { caseDir := flag.String("case-dir", "", "case directory under internal/ingestion/task/testdata/") docID := flag.String("doc-id", "doc-1", "document id") - kbID := flag.String("kb-id", "kb-1", "knowledge base id") docName := flag.String("doc-name", "sample.md", "document name") flag.Parse() @@ -35,7 +34,7 @@ func main() { expectedMetadata := mustReadJSONMap(filepath.Join(outputDir, "merged_metadata.json")) expectedProcessError, hasExpectedProcessError := readOptionalJSONMap(filepath.Join(outputDir, "process_error.json")) - actual, actualErr := runActual(input, *docID, *kbID, *docName) + actual, actualErr := runActual(input, *docID, *docName) if hasExpectedProcessError || actualErr != nil { reportProcessErrorOutcome(expectedProcessError, hasExpectedProcessError, actualErr) return @@ -58,13 +57,13 @@ func main() { os.Exit(1) } -func runActual(input map[string]any, docID string, kbID string, docName string) (result task.GoldenCompareResult, err error) { +func runActual(input map[string]any, docID string, docName string) (result task.GoldenCompareResult, err error) { defer func() { if r := recover(); r != nil { err = fmt.Errorf("panic: %v", r) } }() - return task.ProcessPipelineOutputForGolden(input, docID, kbID, docName) + return task.ProcessPipelineOutputForGolden(input, docID, docName) } func reportProcessErrorOutcome(expected map[string]any, hasExpected bool, actualErr error) {