From 3733da2773a6472e93a08d8991257a67871b2f39 Mon Sep 17 00:00:00 2001
From: euvre <93761161+euvre@users.noreply.github.com>
Date: Thu, 23 Jul 2026 09:58:43 +0800
Subject: [PATCH] fix: count distinct chunk IDs so doc.chunk_num matches actual
 indexed chunks (#17182)

---
 internal/ingestion/task/pipeline_executor.go  | 22 ++++++++++-
 .../ingestion/task/pipeline_executor_test.go  | 39 +++++++++++++++++++
 2 files changed, 60 insertions(+), 1 deletion(-)

diff --git a/internal/ingestion/task/pipeline_executor.go b/internal/ingestion/task/pipeline_executor.go
index 76c941812e..1f4f1595b8 100644
--- a/internal/ingestion/task/pipeline_executor.go
+++ b/internal/ingestion/task/pipeline_executor.go
@@ -224,16 +224,36 @@ func (s *PipelineExecutor) processOutput(ctx context.Context, pipelineOutput map
 		return nil, err
 	}
 
+	chunkCount := countDistinctChunkIDs(chunks)
+
 	return &PipelineResult{
 		DocID:            s.taskCtx.Doc.ID,
 		KbID:             s.taskCtx.Doc.KbID,
 		Metadata:         metadata,
-		ChunkCount:       len(chunks),
+		ChunkCount:       chunkCount,
 		TokenConsumption: embeddingTokenConsumption,
 		Duration:         time.Since(start).Seconds(),
 	}, nil
 }
 
+// countDistinctChunkIDs returns the number of distinct chunk IDs in the slice.
+// After ProcessChunksForPipeline, every chunk carries an "id" field computed
+// from xxhash(text+docID). Chunks with identical text share the same id and
+// the search engine treats them as upserts, so the effective stored chunk count
+// is the number of unique ids — not len(chunks). Mirrors the index-side
+// deduplication that happens at write time.
+func countDistinctChunkIDs(chunks []map[string]any) int {
+	seen := make(map[string]struct{}, len(chunks))
+	for _, ck := range chunks {
+		id, _ := ck["id"].(string)
+		if id == "" {
+			continue
+		}
+		seen[id] = struct{}{}
+	}
+	return len(seen)
+}
+
 func (s *PipelineExecutor) recordPipelineLog(docID, dsl, status string) {
 	var dslMap entity.JSONMap
 	if err := json.Unmarshal([]byte(dsl), &dslMap); err != nil {
diff --git a/internal/ingestion/task/pipeline_executor_test.go b/internal/ingestion/task/pipeline_executor_test.go
index 55c6108911..8e504ac6fe 100644
--- a/internal/ingestion/task/pipeline_executor_test.go
+++ b/internal/ingestion/task/pipeline_executor_test.go
@@ -472,3 +472,42 @@ func TestPipelineExecutorRunPipelineWithDSLForwardsSink(t *testing.T) {
 		}
 	}
 }
+
+func TestCountDistinctChunkIDs(t *testing.T) {
+	// Empty list
+	if n := countDistinctChunkIDs(nil); n != 0 {
+		t.Fatalf("nil chunks: got %d, want 0", n)
+	}
+
+	// All unique
+	chunks := []map[string]any{
+		{"id": "a"},
+		{"id": "b"},
+		{"id": "c"},
+	}
+	if n := countDistinctChunkIDs(chunks); n != 3 {
+		t.Fatalf("all unique: got %d, want 3", n)
+	}
+
+	// Duplicates present — this is the key case
+	chunks = []map[string]any{
+		{"id": "x"},
+		{"id": "y"},
+		{"id": "x"}, // duplicate of [0]
+		{"id": "z"},
+		{"id": "y"}, // duplicate of [1]
+	}
+	if n := countDistinctChunkIDs(chunks); n != 3 {
+		t.Fatalf("with duplicates: got %d, want 3", n)
+	}
+
+	// Missing id fields are skipped
+	chunks = []map[string]any{
+		{"id": "one"},
+		{"text": "no id"},
+		{"id": "two"},
+	}
+	if n := countDistinctChunkIDs(chunks); n != 2 {
+		t.Fatalf("mixed present/absent ids: got %d, want 2", n)
+	}
+}