diff --git a/internal/parser/chunk/postprocess.go b/internal/parser/chunk/postprocess.go index fd79fad9e..585cc4945 100644 --- a/internal/parser/chunk/postprocess.go +++ b/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 diff --git a/internal/parser/chunk/postprocess_test.go b/internal/parser/chunk/postprocess_test.go new file mode 100644 index 000000000..cb237ff95 --- /dev/null +++ b/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) + } +}