From b59d6e8ba10fff5d47cb0fd2ca8724e974444ac6 Mon Sep 17 00:00:00 2001 From: Jack Date: Tue, 4 Aug 2026 10:05:27 +0800 Subject: [PATCH] refactor(ingestion/task): extract index-doc mapping into task/indexdoc package (#17749) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Extract the pipeline-output → search-engine index document mapping helpers out of the `task` package into a dedicated, dependency-light leaf package `internal/ingestion/task/indexdoc`. These functions are pure transforms (they only depend on `common`/`utility`) and are not task-orchestration concerns: - `NormalizeChunks`, `DeepCopyChunks` (was unexported `deepCopyChunks`), `toChunkMaps` → `indexdoc/normalize.go` - `ProcessChunksForPipeline`, `RenameTextToContentWithWeight`, `GetEmbeddingTokenConsumption`, `cleanupConsumedChunkFields`, `mergeChunkMetadata`, `processChunkPositions`, `AggregateTableDocMetadata`, `resolveTableColumnConfig` → `indexdoc/process.go` - `AddPositions` → `indexdoc/position.go` - `EmbeddingTokenConsumptionKey` constant → `indexdoc/constants.go` (task/constants.go keeps only `GRAPH_RAPTOR_FAKE_DOC_ID`) Call sites in `pipeline_executor.go` and `golden_compare.go` now reference the `indexdoc` package; package-task tests qualify the moved symbols. ## Why The `task` package had grown into a "orchestration + pure mapping + debug" mix. Splitting the pure mapping helpers into a leaf package sharpens package boundaries, removes a misleading top-level `ingestion/chunk` candidate (there are already `parser/chunk` and `service/chunk`), and lets the golden tool / future reuse pull in the mapping logic without dragging in `task`'s `dao`/`engine`/`service` dependency graph (Go subpackage import does not pull in the parent). ## Test plan - `build.sh --test ./internal/ingestion/task/...` — **green** (task 4.7s, indexdoc 0.007s), matching the pre-change baseline. - `gofmt` clean; `build.sh` builds both `ragflow-cli` and `ragflow_server` successfully. - Integration/E2E tiers are delegated to CI (need real MySQL/MinIO/ES services). Note: `pipeline_e2e_test.go` has a **pre-existing** compile error (`server.ElasticsearchConfig` / `server.InfinityConfig` are now defined under `internal/server/config/`, not re-exported by `internal/server`). This is unrelated to this change — the diff to that file is only the added `indexdoc` import and the qualified `EmbeddingTokenConsumptionKey` reference. --- internal/ingestion/task/constants.go | 3 --- internal/ingestion/task/golden_compare.go | 8 ++++--- .../ingestion/task/golden_compare_test.go | 4 +++- internal/ingestion/task/indexdoc/constants.go | 20 ++++++++++++++++ .../{chunk_utils.go => indexdoc/normalize.go} | 24 ++++++++++++------- .../normalize_test.go} | 2 +- .../ingestion/task/{ => indexdoc}/position.go | 2 +- .../task/{ => indexdoc}/position_test.go | 2 +- .../{chunk_process.go => indexdoc/process.go} | 2 +- .../process_test.go} | 2 +- internal/ingestion/task/pipeline_e2e_test.go | 3 ++- internal/ingestion/task/pipeline_executor.go | 13 +++++----- .../ingestion/task/pipeline_executor_test.go | 3 ++- 13 files changed, 59 insertions(+), 29 deletions(-) create mode 100644 internal/ingestion/task/indexdoc/constants.go rename internal/ingestion/task/{chunk_utils.go => indexdoc/normalize.go} (80%) rename internal/ingestion/task/{chunk_utils_test.go => indexdoc/normalize_test.go} (99%) rename internal/ingestion/task/{ => indexdoc}/position.go (99%) rename internal/ingestion/task/{ => indexdoc}/position_test.go (99%) rename internal/ingestion/task/{chunk_process.go => indexdoc/process.go} (99%) rename internal/ingestion/task/{chunk_process_test.go => indexdoc/process_test.go} (99%) diff --git a/internal/ingestion/task/constants.go b/internal/ingestion/task/constants.go index e0450e56f2..d7eebaa8b5 100644 --- a/internal/ingestion/task/constants.go +++ b/internal/ingestion/task/constants.go @@ -20,7 +20,4 @@ package task const ( // GRAPH_RAPTOR_FAKE_DOC_ID is the fake doc_id used for RAPTOR-generated chunks. GRAPH_RAPTOR_FAKE_DOC_ID = "graph_raptor_fake_doc" - - // EmbeddingTokenConsumptionKey is the key in pipeline output for embedding token count. - EmbeddingTokenConsumptionKey = "embedding_token_consumption" ) diff --git a/internal/ingestion/task/golden_compare.go b/internal/ingestion/task/golden_compare.go index dc2bdd4350..90252122cb 100644 --- a/internal/ingestion/task/golden_compare.go +++ b/internal/ingestion/task/golden_compare.go @@ -18,6 +18,8 @@ package task import ( "time" + + indexdoc "ragflow/internal/ingestion/task/indexdoc" ) // GoldenCompareResult is the structured output used by the local golden tools. @@ -35,13 +37,13 @@ func ProcessPipelineOutputForGolden( kbID string, docName string, ) (GoldenCompareResult, error) { - normalized := NormalizeChunks(pipelineOutput) + normalized := indexdoc.NormalizeChunks(pipelineOutput) if normalized == nil { normalized = []map[string]any{} } - processed := deepCopyChunks(normalized) - metadata, err := ProcessChunksForPipeline(processed, docID, kbID, docName, time.Now()) + processed := indexdoc.DeepCopyChunks(normalized) + metadata, err := indexdoc.ProcessChunksForPipeline(processed, docID, kbID, 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 eb2761bdc0..2ebc5a9004 100644 --- a/internal/ingestion/task/golden_compare_test.go +++ b/internal/ingestion/task/golden_compare_test.go @@ -3,6 +3,8 @@ package task import ( "testing" "time" + + indexdoc "ragflow/internal/ingestion/task/indexdoc" ) func TestProcessPipelineOutputForGolden_Markdown(t *testing.T) { @@ -44,7 +46,7 @@ func TestProcessChunksForPipeline_StableFields(t *testing.T) { }, } - meta, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "sample.md", now) + meta, err := indexdoc.ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "sample.md", now) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) } diff --git a/internal/ingestion/task/indexdoc/constants.go b/internal/ingestion/task/indexdoc/constants.go new file mode 100644 index 0000000000..3111f6f6f6 --- /dev/null +++ b/internal/ingestion/task/indexdoc/constants.go @@ -0,0 +1,20 @@ +// +// 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 indexdoc + +// EmbeddingTokenConsumptionKey is the key in pipeline output for embedding token count. +const EmbeddingTokenConsumptionKey = "embedding_token_consumption" diff --git a/internal/ingestion/task/chunk_utils.go b/internal/ingestion/task/indexdoc/normalize.go similarity index 80% rename from internal/ingestion/task/chunk_utils.go rename to internal/ingestion/task/indexdoc/normalize.go index b79cbecc2e..cc678cf679 100644 --- a/internal/ingestion/task/chunk_utils.go +++ b/internal/ingestion/task/indexdoc/normalize.go @@ -14,7 +14,7 @@ // limitations under the License. // -package task +package indexdoc import ( "fmt" @@ -30,16 +30,16 @@ func NormalizeChunks(output map[string]any) []map[string]any { } if chunks, ok := output["chunks"].([]map[string]any); ok { - return deepCopyChunks(chunks) + return DeepCopyChunks(chunks) } if chunks, ok := toChunkMaps(output["chunks"]); ok { - return deepCopyChunks(chunks) + return DeepCopyChunks(chunks) } if json, ok := output["json"].([]map[string]any); ok { - return deepCopyChunks(json) + return DeepCopyChunks(json) } if json, ok := toChunkMaps(output["json"]); ok { - return deepCopyChunks(json) + return DeepCopyChunks(json) } if md, ok := output["markdown"].(string); ok && md != "" { return []map[string]any{{"text": md}} @@ -69,10 +69,16 @@ func toChunkMaps(v any) ([]map[string]any, bool) { return out, true } -// deepCopyChunks returns a deep copy of the chunk slice and each chunk map. -// Slice values (e.g. []float64 vectors) are fully copied, not shared. -// Mirrors Python: copy.deepcopy() -func deepCopyChunks(chunks []map[string]any) []map[string]any { +// DeepCopyChunks returns a copy of the chunk slice and each chunk map. +// The chunk maps themselves are freshly allocated, and the value types that +// the pipeline actually emits — []float64 (vectors), []int, and []string — are +// element-wise copied so callers cannot mutate the originals through them. +// Other value types (nested maps, [][]float64 positions, etc.) are shared by +// reference, not recursively deep-copied; positions are later flattened and +// copied independently by processChunkPositions. It is therefore NOT a full +// recursive deep copy (unlike Python's copy.deepcopy), only the copy needed +// for the post-processing pass over pipeline output. +func DeepCopyChunks(chunks []map[string]any) []map[string]any { if chunks == nil { return nil } diff --git a/internal/ingestion/task/chunk_utils_test.go b/internal/ingestion/task/indexdoc/normalize_test.go similarity index 99% rename from internal/ingestion/task/chunk_utils_test.go rename to internal/ingestion/task/indexdoc/normalize_test.go index 0698f09171..e27e9ab0cd 100644 --- a/internal/ingestion/task/chunk_utils_test.go +++ b/internal/ingestion/task/indexdoc/normalize_test.go @@ -1,4 +1,4 @@ -package task +package indexdoc import ( "testing" diff --git a/internal/ingestion/task/position.go b/internal/ingestion/task/indexdoc/position.go similarity index 99% rename from internal/ingestion/task/position.go rename to internal/ingestion/task/indexdoc/position.go index e9f53bf9ad..26ef0a65df 100644 --- a/internal/ingestion/task/position.go +++ b/internal/ingestion/task/indexdoc/position.go @@ -14,7 +14,7 @@ // limitations under the License. // -package task +package indexdoc // AddPositions adds position fields to a chunk map. // Input positions is a flat []float64 grouped as [pn, left, right, top, bottom] diff --git a/internal/ingestion/task/position_test.go b/internal/ingestion/task/indexdoc/position_test.go similarity index 99% rename from internal/ingestion/task/position_test.go rename to internal/ingestion/task/indexdoc/position_test.go index 2cb79159dc..50464b0d55 100644 --- a/internal/ingestion/task/position_test.go +++ b/internal/ingestion/task/indexdoc/position_test.go @@ -1,4 +1,4 @@ -package task +package indexdoc import ( "testing" diff --git a/internal/ingestion/task/chunk_process.go b/internal/ingestion/task/indexdoc/process.go similarity index 99% rename from internal/ingestion/task/chunk_process.go rename to internal/ingestion/task/indexdoc/process.go index d6c352a166..d5516e8d94 100644 --- a/internal/ingestion/task/chunk_process.go +++ b/internal/ingestion/task/indexdoc/process.go @@ -14,7 +14,7 @@ // limitations under the License. // -package task +package indexdoc import ( "fmt" diff --git a/internal/ingestion/task/chunk_process_test.go b/internal/ingestion/task/indexdoc/process_test.go similarity index 99% rename from internal/ingestion/task/chunk_process_test.go rename to internal/ingestion/task/indexdoc/process_test.go index 1f1e409777..61f1c0b2da 100644 --- a/internal/ingestion/task/chunk_process_test.go +++ b/internal/ingestion/task/indexdoc/process_test.go @@ -1,4 +1,4 @@ -package task +package indexdoc import ( "testing" diff --git a/internal/ingestion/task/pipeline_e2e_test.go b/internal/ingestion/task/pipeline_e2e_test.go index ac9cc690d8..b68970c335 100644 --- a/internal/ingestion/task/pipeline_e2e_test.go +++ b/internal/ingestion/task/pipeline_e2e_test.go @@ -30,6 +30,7 @@ import ( "ragflow/internal/engine" "ragflow/internal/engine/elasticsearch" "ragflow/internal/engine/infinity" + indexdoc "ragflow/internal/ingestion/task/indexdoc" "ragflow/internal/ingestion/testutil" "ragflow/internal/server" "ragflow/internal/service" @@ -247,7 +248,7 @@ func TestPipelineE2E_PipelineExecutor(t *testing.T) { "q_2_vec": []float64{0.3, 0.4}, // Pre-vectorized to skip embedding }, }, - EmbeddingTokenConsumptionKey: 100, + indexdoc.EmbeddingTokenConsumptionKey: 100, }, dsl, nil }) diff --git a/internal/ingestion/task/pipeline_executor.go b/internal/ingestion/task/pipeline_executor.go index f8a3fbd5b5..2d9a3168be 100644 --- a/internal/ingestion/task/pipeline_executor.go +++ b/internal/ingestion/task/pipeline_executor.go @@ -31,6 +31,7 @@ import ( "ragflow/internal/ingestion/component" "ragflow/internal/ingestion/knowledge_compile" pipelinepkg "ragflow/internal/ingestion/pipeline" + indexdoc "ragflow/internal/ingestion/task/indexdoc" "gorm.io/gorm" ) @@ -198,13 +199,13 @@ func (s *PipelineExecutor) Execute(ctx context.Context) (*PipelineResult, error) // performs no DB/index writes — the embedding vectors already computed by the // pipeline run are left on the chunks. This keeps debug runs side-effect free. func (s *PipelineExecutor) collectDebugOutput(ctx context.Context, pipelineOutput map[string]any, start time.Time) (*PipelineResult, error) { - chunks := NormalizeChunks(pipelineOutput) + chunks := indexdoc.NormalizeChunks(pipelineOutput) return &PipelineResult{ DocID: s.taskCtx.Doc.ID, KbID: s.taskCtx.Doc.KbID, Chunks: chunks, ChunkCount: countDistinctChunkIDs(chunks), - TokenConsumption: GetEmbeddingTokenConsumption(pipelineOutput), + TokenConsumption: indexdoc.GetEmbeddingTokenConsumption(pipelineOutput), Duration: time.Since(start).Seconds(), }, nil } @@ -217,13 +218,13 @@ func (s *PipelineExecutor) processOutput(ctx context.Context, pipelineOutput map return nil, err } - chunks := NormalizeChunks(pipelineOutput) + chunks := indexdoc.NormalizeChunks(pipelineOutput) if len(chunks) == 0 { return nil, nil } - embeddingTokenConsumption := GetEmbeddingTokenConsumption(pipelineOutput) - metadata, err := ProcessChunksForPipeline( + embeddingTokenConsumption := indexdoc.GetEmbeddingTokenConsumption(pipelineOutput) + metadata, err := indexdoc.ProcessChunksForPipeline( chunks, s.taskCtx.Doc.ID, s.taskCtx.Doc.KbID, @@ -234,7 +235,7 @@ func (s *PipelineExecutor) processOutput(ctx context.Context, pipelineOutput map return nil, err } - tableMeta := AggregateTableDocMetadata(chunks, map[string]interface{}(s.taskCtx.Doc.ParserConfig)) + tableMeta := indexdoc.AggregateTableDocMetadata(chunks, map[string]interface{}(s.taskCtx.Doc.ParserConfig)) if tableMeta != nil { if metadata == nil { metadata = make(map[string]any) diff --git a/internal/ingestion/task/pipeline_executor_test.go b/internal/ingestion/task/pipeline_executor_test.go index 1fce515841..76609ba8d4 100644 --- a/internal/ingestion/task/pipeline_executor_test.go +++ b/internal/ingestion/task/pipeline_executor_test.go @@ -14,6 +14,7 @@ import ( "ragflow/internal/dao" "ragflow/internal/entity" pipelinepkg "ragflow/internal/ingestion/pipeline" + indexdoc "ragflow/internal/ingestion/task/indexdoc" ) // ============================================================================= @@ -201,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 := 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.KbID, *svc.taskCtx.Doc.Name, time.Now()) if err != nil { t.Fatalf("ProcessChunksForPipeline: %v", err) }