feat(go): drop empty and duplicate chunks in postprocess filter (#16049)

### What problem does this PR solve?

The Go chunk pipeline's `PostprocessOperator` `filter` stage
(`internal/ingestion/chunk/postprocess.go`) only filtered by length
(`min_length`/`max_length`). It could not drop empty/whitespace-only
chunks or duplicate chunks — both standard RAG post-processing steps
(blank chunks shouldn't be indexed; identical chunks waste embedding
compute and add redundant retrieval results).

This adds two optional, default-off booleans to the `filter` config:

- `drop_empty` — drop chunks whose content is empty or whitespace-only.
- `drop_duplicates` — drop chunks whose exact content already appeared
(order-preserving; the first occurrence is kept).

They compose with the existing length bounds and are reflected in
`String()` for plan explainability. Also adds the first unit tests for
the postprocess filter (length bounds, drop_empty, drop_duplicates,
combined, exact-content matching, and config parsing).

Validation: `gofmt` clean, `go vet ./internal/ingestion/chunk/` clean,
`go build` ok, `go test ./internal/ingestion/chunk/` — all tests pass.

Closes #16048

### Type of change

- [x] New Feature (non-breaking change which adds functionality)

Co-authored-by: Ling Qin <qinling0210@163.com>

internal/parser/chunk/postprocess.go

@@ -26,7 +26,9 @@ type mergeConfig struct {
 }
 
 type filterConfig struct {
-	MinLength int `json:"min_length"`
+	MinLength      int  `json:"min_length"`
+	DropEmpty      bool `json:"drop_empty"`      // drop chunks that are empty or whitespace-only
+	DropDuplicates bool `json:"drop_duplicates"` // drop chunks whose content already appeared
 }
 
 // ---------------------------------------------------------------------------
@@ -57,6 +59,12 @@ func NewPostprocessOperator(config map[string]interface{}) (*PostprocessOperator
 		if v, ok := f["min_length"].(float64); ok {
 			op.filter.MinLength = int(v)
 		}
+		if v, ok := f["drop_empty"].(bool); ok {
+			op.filter.DropEmpty = v
+		}
+		if v, ok := f["drop_duplicates"].(bool); ok {
+			op.filter.DropDuplicates = v
+		}
 	}
 
 	return op, nil
@@ -109,6 +117,8 @@ func (o *PostprocessOperator) String() string {
 	if o.filter != nil {
 		fmt.Fprintf(&buf, "  filter:\n")
 		fmt.Fprintf(&buf, "    min_length: %d\n", o.filter.MinLength)
+		fmt.Fprintf(&buf, "    drop_empty: %t\n", o.filter.DropEmpty)
+		fmt.Fprintf(&buf, "    drop_duplicates: %t\n", o.filter.DropDuplicates)
 	}
 
 	return buf.String()
@@ -186,14 +196,31 @@ func (o *PostprocessOperator) mergeChunks(chunks []ChunkData) []ChunkData {
 	return merged
 }
 
-// filterChunks removes chunks outside the length bounds.
+// filterChunks removes chunks outside the length bounds and, when configured,
+// drops empty/whitespace-only chunks and exact-duplicate chunks. Duplicate
+// detection is order-preserving: the first occurrence is kept and later chunks
+// with identical content are dropped.
 func (o *PostprocessOperator) filterChunks(chunks []ChunkData) []ChunkData {
+	var seen map[string]struct{}
+	if o.filter.DropDuplicates {
+		seen = make(map[string]struct{}, len(chunks))
+	}
+
 	filtered := make([]ChunkData, 0, len(chunks))
 	for _, c := range chunks {
+		if o.filter.DropEmpty && strings.TrimSpace(c.Content) == "" {
+			continue
+		}
 		l := len([]rune(c.Content))
 		if o.filter.MinLength > 0 && l < o.filter.MinLength {
 			continue
 		}
+		if o.filter.DropDuplicates {
+			if _, dup := seen[c.Content]; dup {
+				continue
+			}
+			seen[c.Content] = struct{}{}
+		}
 		filtered = append(filtered, c)
 	}
 	return filtered

internal/parser/chunk/postprocess_test.go

@@ -0,0 +1,132 @@
+//
+// 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 chunk
+
+import (
+	"strings"
+	"testing"
+)
+
+// runFilter builds a postprocess operator whose only stage is the filter, runs
+// it over the given chunk contents, and returns the resulting contents in order.
+func runFilter(t *testing.T, filter map[string]interface{}, contents ...string) []string {
+	t.Helper()
+	op, err := NewPostprocessOperator(map[string]interface{}{"filter": filter})
+	if err != nil {
+		t.Fatalf("NewPostprocessOperator returned error: %v", err)
+	}
+	ctx := &ChunkContext{SplitChunks: make([]ChunkData, len(contents))}
+	for i, c := range contents {
+		ctx.SplitChunks[i] = ChunkData{Content: c, Index: i}
+	}
+	if err := op.Execute(ctx); err != nil {
+		t.Fatalf("Execute returned error: %v", err)
+	}
+	out := make([]string, len(ctx.ResultChunks))
+	for i, c := range ctx.ResultChunks {
+		out[i] = c.Content
+	}
+	return out
+}
+
+func eq(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func TestFilterLengthBounds(t *testing.T) {
+	got := runFilter(t, map[string]interface{}{
+		"min_length": float64(2),
+	}, "a", "bb", "cccc", "ddddd")
+	want := []string{"bb", "cccc", "ddddd"}
+	if !eq(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestFilterDropEmpty(t *testing.T) {
+	got := runFilter(t, map[string]interface{}{
+		"drop_empty": true,
+	}, "keep", "   ", "", "\n\t", "also")
+	want := []string{"keep", "also"}
+	if !eq(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestFilterDropDuplicatesPreservesFirst(t *testing.T) {
+	got := runFilter(t, map[string]interface{}{
+		"drop_duplicates": true,
+	}, "a", "b", "a", "c", "b", "a")
+	want := []string{"a", "b", "c"}
+	if !eq(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestFilterDropEmptyAndDuplicatesCombined(t *testing.T) {
+	got := runFilter(t, map[string]interface{}{
+		"drop_empty":      true,
+		"drop_duplicates": true,
+	}, "x", "  ", "x", "y", "")
+	want := []string{"x", "y"}
+	if !eq(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestFilterDuplicatesAreContentExact(t *testing.T) {
+	// Differing whitespace => distinct content => both kept.
+	got := runFilter(t, map[string]interface{}{
+		"drop_duplicates": true,
+	}, "a", "a ", "a")
+	want := []string{"a", "a "}
+	if !eq(got, want) {
+		t.Fatalf("got %v, want %v", got, want)
+	}
+}
+
+func TestNewPostprocessOperatorParsesFilterFlags(t *testing.T) {
+	op, err := NewPostprocessOperator(map[string]interface{}{
+		"filter": map[string]interface{}{
+			"min_length":      float64(3),
+			"drop_empty":      true,
+			"drop_duplicates": true,
+		},
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if op.filter == nil {
+		t.Fatal("filter config not parsed")
+	}
+	if op.filter.MinLength != 3 || !op.filter.DropEmpty || !op.filter.DropDuplicates {
+		t.Errorf("parsed filter = %+v, want min_length=3 drop_empty=true drop_duplicates=true", *op.filter)
+	}
+	// The new flags must surface in String() for plan explainability.
+	s := op.String()
+	if !strings.Contains(s, "drop_empty: true") || !strings.Contains(s, "drop_duplicates: true") {
+		t.Errorf("String() missing filter flags:\n%s", s)
+	}
+}