From 2c25698ce8dabbab00a4aa5ac8e73da0a702a2f4 Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Sat, 18 Jul 2026 01:47:29 +0800 Subject: [PATCH 1/2] perf(readutil): preserve early filters across fallback paths --- .../disttae/local_disttae_datasource.go | 13 +- .../engine/disttae/logtailreplay/rows_iter.go | 146 ++++++++++ .../disttae/logtailreplay/rows_iter_test.go | 88 ++++++ pkg/vm/engine/disttae/pk_filter_observe.go | 11 +- .../snapshot_materialized_datasource.go | 5 +- pkg/vm/engine/readutil/expr_filter.go | 14 +- pkg/vm/engine/readutil/filter_test.go | 67 +++++ pkg/vm/engine/readutil/pk_filter.go | 5 +- pkg/vm/engine/readutil/pk_filter_mem.go | 271 ++++++++++-------- pkg/vm/engine/readutil/pk_filter_mem_test.go | 234 +++++++++++++++ 10 files changed, 719 insertions(+), 135 deletions(-) diff --git a/pkg/vm/engine/disttae/local_disttae_datasource.go b/pkg/vm/engine/disttae/local_disttae_datasource.go index b907ca1d6e8ea..88da457d98399 100644 --- a/pkg/vm/engine/disttae/local_disttae_datasource.go +++ b/pkg/vm/engine/disttae/local_disttae_datasource.go @@ -775,10 +775,9 @@ func (ls *LocalDisttaeDataSource) filterInMemCommittedInserts( if !ls.memPKFilter.Valid() { ls.pStateRows.insIter = ls.pState.NewRowsIter(ls.snapshotTS, nil, false) } else { - ls.pStateRows.insIter = ls.pState.NewPrimaryKeyIter( + ls.pStateRows.insIter = ls.pState.NewPrimaryKeyIterWithFilters( ls.memPKFilter.TS, - ls.memPKFilter.Op(), - ls.memPKFilter.Keys()) + ls.memPKFilter.Specs()) } if summaryBuf != nil { summaryBuf.WriteString(fmt.Sprintf("[PScan] insIter created %v\n", ls.memPKFilter.String())) @@ -1287,6 +1286,14 @@ func (ls *LocalDisttaeDataSource) getInMemDelIter( ls.memPKFilter == nil || !ls.memPKFilter.Valid() { return ls.pState.NewRowsIter(ls.snapshotTS, bid, true), false } + filterSpecs := ls.memPKFilter.Specs() + if len(filterSpecs) > 1 { + return ls.pState.NewPrimaryKeyDelIterWithFilters( + &ls.memPKFilter.TS, + bid, + filterSpecs, + ), false + } inValCnt, ok := ls.memPKFilter.InKind() if !ok { diff --git a/pkg/vm/engine/disttae/logtailreplay/rows_iter.go b/pkg/vm/engine/disttae/logtailreplay/rows_iter.go index 837e136533b3f..c0736584dd127 100644 --- a/pkg/vm/engine/disttae/logtailreplay/rows_iter.go +++ b/pkg/vm/engine/disttae/logtailreplay/rows_iter.go @@ -16,6 +16,7 @@ package logtailreplay import ( "bytes" + "container/heap" "fmt" "math" @@ -814,6 +815,151 @@ func (p *PartitionState) NewPrimaryKeyIter( } } +type primaryKeyIterHeap struct { + children []primaryKeyOrderedIter + indexes []int +} + +func (h primaryKeyIterHeap) Len() int { + return len(h.indexes) +} + +func (h primaryKeyIterHeap) Less(i, j int) bool { + left := h.children[h.indexes[i]].currentIndexEntry() + right := h.children[h.indexes[j]].currentIndexEntry() + return left.Less(right) +} + +func (h primaryKeyIterHeap) Swap(i, j int) { + h.indexes[i], h.indexes[j] = h.indexes[j], h.indexes[i] +} + +func (h *primaryKeyIterHeap) Push(value any) { + h.indexes = append(h.indexes, value.(int)) +} + +func (h *primaryKeyIterHeap) Pop() any { + last := len(h.indexes) - 1 + value := h.indexes[last] + h.indexes = h.indexes[:last] + return value +} + +type primaryKeyUnionIter struct { + heap primaryKeyIterHeap + initialized bool + closed bool + curRow *RowEntry + last PrimaryIndexEntry + hasLast bool +} + +var _ RowsIter = new(primaryKeyUnionIter) + +type primaryKeyOrderedIter interface { + RowsIter + currentIndexEntry() *PrimaryIndexEntry +} + +func (p *primaryKeyIter) currentIndexEntry() *PrimaryIndexEntry { + return p.iter.Item() +} + +func samePrimaryIndexEntry(left, right *PrimaryIndexEntry) bool { + return left.RowEntryID == right.RowEntryID && + left.Time.Compare(&right.Time) == 0 && + bytes.Equal(left.Bytes, right.Bytes) +} + +func (p *primaryKeyUnionIter) Next() bool { + if p.closed { + return false + } + if !p.initialized { + p.initialized = true + for idx := range p.heap.children { + if p.heap.children[idx].Next() { + heap.Push(&p.heap, idx) + } + } + } + + for p.heap.Len() > 0 { + idx := heap.Pop(&p.heap).(int) + child := p.heap.children[idx] + entry := *child.currentIndexEntry() + row := child.Entry() + if child.Next() { + heap.Push(&p.heap, idx) + } + + if p.hasLast && samePrimaryIndexEntry(&entry, &p.last) { + continue + } + p.last = entry + p.hasLast = true + p.curRow = row + return true + } + + return false +} + +func (p *primaryKeyUnionIter) Entry() *RowEntry { + return p.curRow +} + +func (p *primaryKeyUnionIter) Close() error { + if p.closed { + return nil + } + p.closed = true + var firstErr error + for idx := range p.heap.children { + if err := p.heap.children[idx].Close(); err != nil && firstErr == nil { + firstErr = err + } + } + p.heap.children = nil + p.heap.indexes = nil + return firstErr +} + +func (p *PartitionState) NewPrimaryKeyIterWithFilters( + ts types.TS, + filters []readutil.MemPKFilterSpec, +) RowsIter { + if len(filters) == 1 { + return p.NewPrimaryKeyIter(ts, filters[0].Op, filters[0].Keys) + } + + children := make([]primaryKeyOrderedIter, 0, len(filters)) + for idx := range filters { + children = append(children, p.NewPrimaryKeyIter(ts, filters[idx].Op, filters[idx].Keys)) + } + return &primaryKeyUnionIter{ + heap: primaryKeyIterHeap{children: children}, + } +} + +func (p *PartitionState) NewPrimaryKeyDelIterWithFilters( + ts *types.TS, + bid *types.Blockid, + filters []readutil.MemPKFilterSpec, +) RowsIter { + if len(filters) == 1 { + return p.NewPrimaryKeyDelIter(ts, bid, filters[0].Op, filters[0].Keys) + } + + children := make([]primaryKeyOrderedIter, 0, len(filters)) + for idx := range filters { + children = append(children, p.NewPrimaryKeyDelIter(ts, bid, filters[idx].Op, filters[idx].Keys)) + } + return &primaryKeyUnionIter{ + heap: primaryKeyIterHeap{children: children}, + } +} + func (p *PartitionState) NewPrimaryKeyDelIter( ts *types.TS, bid *types.Blockid, diff --git a/pkg/vm/engine/disttae/logtailreplay/rows_iter_test.go b/pkg/vm/engine/disttae/logtailreplay/rows_iter_test.go index a50b4fb5e3bf2..7ba073ad41bb7 100644 --- a/pkg/vm/engine/disttae/logtailreplay/rows_iter_test.go +++ b/pkg/vm/engine/disttae/logtailreplay/rows_iter_test.go @@ -569,6 +569,94 @@ func TestPrefixIn(t *testing.T) { require.Equal(t, []byte{4}, pkIter.iter.Item().Bytes) } +func TestPrimaryKeyUnionIter(t *testing.T) { + state := NewPartitionState("", false, 42, false) + pool := mpool.MustNewZero() + packer := types.NewPacker() + defer packer.Close() + + rowIDVec := vector.NewVec(types.T_Rowid.ToType()) + tsVec := vector.NewVec(types.T_TS.ToType()) + pkVec := vector.NewVec(types.T_int64.ToType()) + segmentID := objectio.NewSegmentid() + blockIDs := make([]*types.Blockid, 0, 6) + for idx := 0; idx < 6; idx++ { + blockID := objectio.NewBlockid(segmentID, uint16(idx), 0) + blockIDs = append(blockIDs, blockID) + vector.AppendFixed(rowIDVec, objectio.NewRowid(blockID, 0), false, pool) + vector.AppendFixed(tsVec, types.BuildTS(1, 0), false, pool) + vector.AppendFixed(pkVec, int64(idx), false, pool) + } + state.HandleRowsInsert(context.Background(), &api.Batch{ + Attrs: []string{"rowid", "time", "pk"}, + Vecs: []api.Vector{ + mustVectorToProto(rowIDVec), + mustVectorToProto(tsVec), + mustVectorToProto(pkVec), + }, + }, 0, packer, pool) + + iter := state.NewPrimaryKeyIterWithFilters(types.BuildTS(2, 0), []readutil.MemPKFilterSpec{ + {Op: function.EQUAL, Keys: [][]byte{readutil.EncodePrimaryKey(int64(1), packer)}}, + {Op: function.BETWEEN, Keys: [][]byte{ + readutil.EncodePrimaryKey(int64(1), packer), + readutil.EncodePrimaryKey(int64(3), packer), + }}, + }) + union, ok := iter.(*primaryKeyUnionIter) + require.True(t, ok) + require.Len(t, union.heap.children, 2) + + var got []int64 + for iter.Next() { + values, err := types.Unpack(iter.Entry().PrimaryIndexBytes) + require.NoError(t, err) + got = append(got, values[0].(int64)) + } + require.Equal(t, []int64{1, 2, 3}, got) + require.NoError(t, iter.Close()) + require.NoError(t, iter.Close()) + require.False(t, iter.Next()) + + empty := state.NewPrimaryKeyIterWithFilters(types.BuildTS(2, 0), nil) + require.False(t, empty.Next()) + require.NoError(t, empty.Close()) + + deletedRowIDs := vector.NewVec(types.T_Rowid.ToType()) + deleteTS := vector.NewVec(types.T_TS.ToType()) + deletedPKs := vector.NewVec(types.T_int64.ToType()) + tombstoneRowIDs := vector.NewVec(types.T_Rowid.ToType()) + vector.AppendFixed(deletedRowIDs, objectio.NewRowid(blockIDs[1], 0), false, pool) + vector.AppendFixed(deleteTS, types.BuildTS(3, 0), false, pool) + vector.AppendFixed(deletedPKs, int64(1), false, pool) + vector.AppendFixed(tombstoneRowIDs, types.RandomRowid(), false, pool) + state.HandleRowsDelete(context.Background(), &api.Batch{ + Attrs: []string{"rowid", "time"}, + Vecs: []api.Vector{ + mustVectorToProto(deletedRowIDs), + mustVectorToProto(deleteTS), + mustVectorToProto(deletedPKs), + mustVectorToProto(tombstoneRowIDs), + }, + }, packer, pool) + + deleteSnapshot := types.BuildTS(4, 0) + deleteIter := state.NewPrimaryKeyDelIterWithFilters( + &deleteSnapshot, + blockIDs[1], + []readutil.MemPKFilterSpec{ + {Op: function.EQUAL, Keys: [][]byte{readutil.EncodePrimaryKey(int64(1), packer)}}, + {Op: function.BETWEEN, Keys: [][]byte{ + readutil.EncodePrimaryKey(int64(1), packer), + readutil.EncodePrimaryKey(int64(3), packer), + }}, + }, + ) + require.True(t, deleteIter.Next()) + require.False(t, deleteIter.Next()) + require.NoError(t, deleteIter.Close()) +} + func TestPrefixBetweenOpenBounds(t *testing.T) { pkTree := btree.NewBTreeGOptions((*PrimaryIndexEntry).Less, btree.Options{Degree: 64}) for _, key := range []string{"aa1", "aa2", "ab1", "ab2", "ac1"} { diff --git a/pkg/vm/engine/disttae/pk_filter_observe.go b/pkg/vm/engine/disttae/pk_filter_observe.go index 2d17760f427c1..d45085db4db26 100644 --- a/pkg/vm/engine/disttae/pk_filter_observe.go +++ b/pkg/vm/engine/disttae/pk_filter_observe.go @@ -43,13 +43,21 @@ func logPKFilterInMemSummary( filterValid bool filterOp int keyCnt int + disjunctCnt int exact bool exactHit bool ) if filter != nil { filterValid = filter.Valid() filterOp = filter.Op() - keyCnt = len(filter.Keys()) + specs := filter.Specs() + disjunctCnt = len(specs) + for idx := range specs { + keyCnt += len(specs[idx].Keys) + } + if disjunctCnt > 1 { + filterOp = -1 + } exact, exactHit = filter.Exact() } @@ -61,6 +69,7 @@ func logPKFilterInMemSummary( zap.Bool("has-mem-pk-filter", filter != nil), zap.Bool("mem-pk-filter-valid", filterValid), zap.Int("filter-op", filterOp), + zap.Int("disjunct-count", disjunctCnt), zap.Int("key-count", keyCnt), zap.Bool("exact-filter", exact), zap.Bool("exact-hit", exactHit), diff --git a/pkg/vm/engine/disttae/snapshot_materialized_datasource.go b/pkg/vm/engine/disttae/snapshot_materialized_datasource.go index 571fb03a1bb09..d93358d34a6a8 100644 --- a/pkg/vm/engine/disttae/snapshot_materialized_datasource.go +++ b/pkg/vm/engine/disttae/snapshot_materialized_datasource.go @@ -214,10 +214,9 @@ func (ds *materializedSnapshotDataSource) buildCommittedInMemBatches( iterKind := "rows" if ds.memPKFilter != nil && ds.memPKFilter.Valid() { iterKind = "primary-key" - iter = ds.pState.NewPrimaryKeyIter( + iter = ds.pState.NewPrimaryKeyIterWithFilters( ds.snapshotTS, - ds.memPKFilter.Op(), - ds.memPKFilter.Keys(), + ds.memPKFilter.Specs(), ) } else { iter = ds.pState.NewRowsIter(ds.snapshotTS, nil, false) diff --git a/pkg/vm/engine/readutil/expr_filter.go b/pkg/vm/engine/readutil/expr_filter.go index f1229885aa3d1..85fa28986b9ac 100644 --- a/pkg/vm/engine/readutil/expr_filter.go +++ b/pkg/vm/engine/readutil/expr_filter.go @@ -125,12 +125,14 @@ func CompileFilterExprs( ops3 := make([]ObjectFilterOp, 0, len(exprs)) ops4 := make([]BlockFilterOp, 0, len(exprs)) ops5 := make([]SeekFirstBlockOp, 0, len(exprs)) + compiled := 0 for _, expr := range exprs { expr_op1, expr_op2, expr_op3, expr_op4, expr_op5, can, hsh := CompileFilterExpr(expr, tableDef, fs) if !can { - return nil, nil, nil, nil, nil, false, false + continue } + compiled++ if expr_op1 != nil { ops1 = append(ops1, expr_op1) } @@ -148,6 +150,9 @@ func CompileFilterExprs( } highSelectivityHint = highSelectivityHint || hsh } + if compiled == 0 { + return nil, nil, nil, nil, nil, false, false + } fastFilterOp = func(obj *objectio.ObjectStats) (bool, error) { for _, op := range ops1 { ok, err := op(obj) @@ -335,12 +340,14 @@ func CompileFilterExpr( objectOps := make([]ObjectFilterOp, 0, len(exprImpl.F.Args)) blockOps := make([]BlockFilterOp, 0, len(exprImpl.F.Args)) seekOps := make([]SeekFirstBlockOp, 0, len(exprImpl.F.Args)) + compiled := 0 for idx := range exprImpl.F.Args { op1, op2, op3, op4, op5, can, hsh := CompileFilterExpr(exprImpl.F.Args[idx], tableDef, fs) if !can { - return nil, nil, nil, nil, nil, false, false + continue } + compiled++ fastOps = append(fastOps, op1) loadOps = append(loadOps, op2) @@ -350,6 +357,9 @@ func CompileFilterExpr( highSelectivityHint = highSelectivityHint || hsh } + if compiled == 0 { + return nil, nil, nil, nil, nil, false, false + } fastFilterOp = func(stats *objectio.ObjectStats) (bool, error) { for idx := range fastOps { diff --git a/pkg/vm/engine/readutil/filter_test.go b/pkg/vm/engine/readutil/filter_test.go index c8ae31ec65a60..9e1aad9bf67b8 100644 --- a/pkg/vm/engine/readutil/filter_test.go +++ b/pkg/vm/engine/readutil/filter_test.go @@ -3529,6 +3529,73 @@ func TestCompileFilterExpr_PrefixInRangeAllFlags(t *testing.T) { } } +func TestCompileFilterExprsPreservesSupportedConjuncts(t *testing.T) { + tableDef := &plan.TableDef{ + Name: "test_tbl", + Name2ColIndex: map[string]int32{"id": 0}, + Pkey: &plan.PrimaryKeyDef{ + Names: []string{"id"}, + PkeyColName: "id", + }, + Cols: []*plan.ColDef{{ + Name: "id", + ColId: 0, + Seqnum: 0, + Primary: true, + Typ: plan.Type{Id: int32(types.T_int64)}, + }}, + } + supported := MakeFunctionExprForTest(">=", []*plan.Expr{ + MakeColExprForTest(0, types.T_int64, "id"), + plan2.MakePlan2Int64ConstExprWithType(10), + }) + unsupported := MakeFunctionExprForTest("abs", []*plan.Expr{ + MakeColExprForTest(0, types.T_int64, "id"), + }) + mp := mpool.MustNew(t.Name()) + proc := testutil.NewProcessWithMPool(t, "", mp) + var executors []colexec.ExpressionExecutor + plan2.ReplaceFoldExpr(proc, supported, &executors) + plan2.EvalFoldExpr(proc, supported, &executors) + for _, executor := range executors { + defer executor.Free() + } + + assertCompiled := func(t *testing.T, exprs []*plan.Expr) { + t.Helper() + fast, load, object, block, seek, canCompile, _ := CompileFilterExprs(exprs, tableDef, nil) + require.True(t, canCompile) + require.NotNil(t, fast) + require.NotNil(t, load) + require.NotNil(t, object) + require.NotNil(t, block) + require.NotNil(t, seek) + } + + t.Run("top-level implicit and", func(t *testing.T) { + assertCompiled(t, []*plan.Expr{supported, unsupported}) + }) + t.Run("nested and", func(t *testing.T) { + assertCompiled(t, []*plan.Expr{MakeFunctionExprForTest("and", []*plan.Expr{supported, unsupported})}) + }) + t.Run("nested or remains all or nothing", func(t *testing.T) { + _, _, _, _, _, canCompile, _ := CompileFilterExprs( + []*plan.Expr{MakeFunctionExprForTest("or", []*plan.Expr{supported, unsupported})}, + tableDef, + nil, + ) + require.False(t, canCompile) + }) + t.Run("all unsupported", func(t *testing.T) { + _, _, _, _, _, canCompile, _ := CompileFilterExprs( + []*plan.Expr{unsupported, unsupported}, + tableDef, + nil, + ) + require.False(t, canCompile) + }) +} + func TestCompileFilterExpr_Between(t *testing.T) { tableDef := &plan.TableDef{ Name: "test_tbl", diff --git a/pkg/vm/engine/readutil/pk_filter.go b/pkg/vm/engine/readutil/pk_filter.go index 55607d9f38153..871d6ee7a0588 100644 --- a/pkg/vm/engine/readutil/pk_filter.go +++ b/pkg/vm/engine/readutil/pk_filter.go @@ -1361,8 +1361,9 @@ func mergeFilters( finalFilter.Oid = left.Oid if !finalFilter.Valid && connector == function.AND && !unsafeInput { - // nonempty disjuncts makes the memory pk filter invalid, - // so we choose the one whose disjuncts is empty as default here. + // Keep one atomic conjunct when representing the full intersection + // would require distributing AND over OR. It remains a safe early + // filter because the residual expression evaluates the full predicate. if len(left.Disjuncts) > 0 { finalFilter = *right if right.Vec != nil { diff --git a/pkg/vm/engine/readutil/pk_filter_mem.go b/pkg/vm/engine/readutil/pk_filter_mem.go index d4f3a921bef15..95acc9e8ed9e1 100644 --- a/pkg/vm/engine/readutil/pk_filter_mem.go +++ b/pkg/vm/engine/readutil/pk_filter_mem.go @@ -17,6 +17,7 @@ package readutil import ( "bytes" "fmt" + "sort" "github.com/matrixorigin/matrixone/pkg/catalog" "github.com/matrixorigin/matrixone/pkg/container/types" @@ -31,13 +32,22 @@ import ( "go.uber.org/zap" ) +// MemPKFilterSpec describes one atomic primary-key predicate. Multiple specs +// are OR-ed by partition-state iterators. +type MemPKFilterSpec struct { + Op int + Keys [][]byte +} + type MemPKFilter struct { - op int - packed [][]byte - inSet map[string]struct{} // pre-built hashmap for IN filter - isVec bool - isValid bool - TS types.TS + op int + packed [][]byte + inSet map[string]struct{} + prefixes [][]byte + disjuncts []MemPKFilter + isVec bool + isValid bool + TS types.TS exact struct { hit bool @@ -60,6 +70,21 @@ func (f *MemPKFilter) Keys() [][]byte { return f.packed } +func (f *MemPKFilter) Specs() []MemPKFilterSpec { + if !f.isValid { + return nil + } + if len(f.disjuncts) == 0 { + return []MemPKFilterSpec{{Op: f.op, Keys: f.packed}} + } + + specs := make([]MemPKFilterSpec, 0, len(f.disjuncts)) + for idx := range f.disjuncts { + specs = append(specs, f.disjuncts[idx].Specs()...) + } + return specs +} + func NewMemPKFilter( tableDef *plan.TableDef, ts timestamp.Timestamp, @@ -74,8 +99,27 @@ func NewMemPKFilter( return } - // Currently only support single atomic filter in memory path. if len(basePKFilter.Disjuncts) > 0 { + filter.disjuncts = make([]MemPKFilter, 0, len(basePKFilter.Disjuncts)) + for idx := range basePKFilter.Disjuncts { + disjunct, err := NewMemPKFilter( + tableDef, + ts, + packerPool, + basePKFilter.Disjuncts[idx], + engine.FilterHint{}, + ) + if err != nil { + return MemPKFilter{}, err + } + if !disjunct.Valid() { + return MemPKFilter{TS: filter.TS}, nil + } + filter.disjuncts = append(filter.disjuncts, disjunct) + } + filter.isValid = len(filter.disjuncts) > 0 + filter.isVec = true + filter.setFilterHint(tableDef, filterHint) return } if !validBlockPKSearchFilter(basePKFilter) { @@ -262,29 +306,33 @@ func NewMemPKFilter( return } - filter.FilterHint = filterHint - if filter.FilterHint.BF != nil && filter.FilterHint.BF.Valid() { - filter.HasBF = true - filter.BFSeqNum = -1 + filter.setFilterHint(tableDef, filterHint) + + return +} + +func (f *MemPKFilter) setFilterHint(tableDef *plan.TableDef, filterHint engine.FilterHint) { + f.FilterHint = filterHint + if f.FilterHint.BF != nil && f.FilterHint.BF.Valid() { + f.HasBF = true + f.BFSeqNum = -1 // For IVF entries table, use __mo_index_pri_col for BF filtering. for _, col := range tableDef.Cols { if col.Name == catalog.IndexTablePrimaryColName { - filter.BFSeqNum = int16(col.Seqnum) + f.BFSeqNum = int16(col.Seqnum) break } } // For fulltext index table, use doc_id column for BF filtering. - if filter.BFSeqNum == -1 && catalog.IsFullTextIndexTableType(tableDef.TableType, tableDef.Name) { + if f.BFSeqNum == -1 && catalog.IsFullTextIndexTableType(tableDef.TableType, tableDef.Name) { for _, col := range tableDef.Cols { if col.Name == catalog.FullTextIndex_TabCol_Id { - filter.BFSeqNum = int16(col.Seqnum) + f.BFSeqNum = int16(col.Seqnum) break } } } } - - return } func (f *MemPKFilter) InKind() (int, bool) { @@ -311,6 +359,10 @@ func (f *MemPKFilter) Must() bool { func (f *MemPKFilter) String() string { var buf bytes.Buffer + if len(f.disjuncts) > 0 { + buf.WriteString(fmt.Sprintf("InMemPKFilter{disjuncts: %d, isValid: %v}", len(f.disjuncts), f.isValid)) + return buf.String() + } buf.WriteString(fmt.Sprintf("InMemPKFilter{op: %d, isVec: %v, isValid: %v vals: [", f.op, f.isVec, f.isValid)) for x := range f.packed { buf.WriteString(fmt.Sprintf("%x, ", f.packed[x])) @@ -324,126 +376,97 @@ func (f *MemPKFilter) SetNull() { } func (f *MemPKFilter) SetFullData(op int, isVec bool, val ...[]byte) { - f.packed = append(f.packed, val...) + f.packed = append(f.packed[:0], val...) f.op = op f.isVec = isVec f.isValid = true -} - -func (f *MemPKFilter) FilterVector( - vec *vector.Vector, - packer *types.Packer, - skipMask *objectio.Bitmap, -) { - keys := EncodePrimaryKeyVector(vec, packer) - - // For IN filter, use hashmap for O(1) lookup - if f.op == function.IN { - // Lazy build hashmap on first use - if f.inSet == nil && len(f.packed) > 0 { - f.inSet = make(map[string]struct{}, len(f.packed)) - for _, k := range f.packed { - f.inSet[string(k)] = struct{}{} + f.inSet = nil + f.prefixes = nil + f.disjuncts = nil + + if op == function.IN || op == function.PREFIX_IN { + sort.Slice(f.packed, func(i, j int) bool { + return bytes.Compare(f.packed[i], f.packed[j]) < 0 + }) + } + if op == function.IN { + f.inSet = make(map[string]struct{}, len(f.packed)) + for _, key := range f.packed { + f.inSet[string(key)] = struct{}{} + } + } + if op == function.PREFIX_IN { + f.prefixes = make([][]byte, 0, len(f.packed)) + for _, prefix := range f.packed { + if len(f.prefixes) > 0 && bytes.HasPrefix(prefix, f.prefixes[len(f.prefixes)-1]) { + continue } + f.prefixes = append(f.prefixes, prefix) } - for i := 0; i < len(keys); i++ { - if _, ok := f.inSet[string(keys[i])]; !ok { - skipMask.Add(uint64(i)) + } +} + +func (f *MemPKFilter) matches(key []byte) bool { + if len(f.disjuncts) > 0 { + for idx := range f.disjuncts { + if f.disjuncts[idx].matches(key) { + return true } } - return + return false } - // For PREFIX_IN with small list, use linear search - if f.op == function.PREFIX_IN && len(f.packed) > 4 { - return + switch f.op { + case function.EQUAL: + return len(f.packed) == 1 && bytes.Equal(key, f.packed[0]) + case function.PREFIX_EQ: + return len(f.packed) == 1 && bytes.HasPrefix(key, f.packed[0]) + case function.IN: + _, ok := f.inSet[string(key)] + return ok + case function.PREFIX_IN: + idx := sort.Search(len(f.prefixes), func(idx int) bool { + return bytes.Compare(f.prefixes[idx], key) > 0 + }) - 1 + return idx >= 0 && bytes.HasPrefix(key, f.prefixes[idx]) + case function.BETWEEN: + return len(f.packed) == 2 && bytes.Compare(key, f.packed[0]) >= 0 && bytes.Compare(key, f.packed[1]) <= 0 + case RangeRightOpen: + return len(f.packed) == 2 && bytes.Compare(key, f.packed[0]) >= 0 && bytes.Compare(key, f.packed[1]) < 0 + case RangeLeftOpen: + return len(f.packed) == 2 && bytes.Compare(key, f.packed[0]) > 0 && bytes.Compare(key, f.packed[1]) <= 0 + case RangeBothOpen: + return len(f.packed) == 2 && bytes.Compare(key, f.packed[0]) > 0 && bytes.Compare(key, f.packed[1]) < 0 + case function.PREFIX_BETWEEN: + return len(f.packed) == 2 && types.PrefixCompare(key, f.packed[0]) >= 0 && types.PrefixCompare(key, f.packed[1]) <= 0 + case PrefixRangeLeftOpen: + return len(f.packed) == 2 && types.PrefixCompare(key, f.packed[0]) > 0 && types.PrefixCompare(key, f.packed[1]) <= 0 + case PrefixRangeRightOpen: + return len(f.packed) == 2 && types.PrefixCompare(key, f.packed[0]) >= 0 && types.PrefixCompare(key, f.packed[1]) < 0 + case PrefixRangeBothOpen: + return len(f.packed) == 2 && types.PrefixCompare(key, f.packed[0]) > 0 && types.PrefixCompare(key, f.packed[1]) < 0 + case function.GREAT_EQUAL: + return len(f.packed) == 1 && bytes.Compare(key, f.packed[0]) >= 0 + case function.GREAT_THAN: + return len(f.packed) == 1 && bytes.Compare(key, f.packed[0]) > 0 + case function.LESS_EQUAL: + return len(f.packed) == 1 && bytes.Compare(key, f.packed[0]) <= 0 + case function.LESS_THAN: + return len(f.packed) == 1 && bytes.Compare(key, f.packed[0]) < 0 + default: + return true } +} +func (f *MemPKFilter) FilterVector( + vec *vector.Vector, + packer *types.Packer, + skipMask *objectio.Bitmap, +) { + keys := EncodePrimaryKeyVector(vec, packer) for i := 0; i < len(keys); i++ { - switch f.op { - case function.EQUAL: - if !bytes.Equal(keys[i], f.packed[0]) { - skipMask.Add(uint64(i)) - } - - case function.PREFIX_EQ: - if !bytes.HasPrefix(keys[i], f.packed[0]) { - skipMask.Add(uint64(i)) - } - - case function.PREFIX_IN: - in := false - for _, k := range f.packed { - if bytes.HasPrefix(keys[i], k) { - in = true - break - } - } - if !in { - skipMask.Add(uint64(i)) - } - case function.BETWEEN: - if !(bytes.Compare(keys[i], f.packed[0]) >= 0 && bytes.Compare(keys[i], f.packed[1]) <= 0) { - skipMask.Add(uint64(i)) - } - - case RangeRightOpen: - if !(bytes.Compare(keys[i], f.packed[0]) >= 0 && bytes.Compare(keys[i], f.packed[1]) < 0) { - skipMask.Add(uint64(i)) - } - - case RangeLeftOpen: - if !(bytes.Compare(keys[i], f.packed[0]) > 0 && bytes.Compare(keys[i], f.packed[1]) <= 0) { - skipMask.Add(uint64(i)) - } - - case RangeBothOpen: - if !(bytes.Compare(keys[i], f.packed[0]) > 0 && bytes.Compare(keys[i], f.packed[1]) < 0) { - skipMask.Add(uint64(i)) - } - - case function.PREFIX_BETWEEN: - if !(types.PrefixCompare(keys[i], f.packed[0]) >= 0 && types.PrefixCompare(keys[i], f.packed[1]) <= 0) { - skipMask.Add(uint64(i)) - } - - case PrefixRangeLeftOpen: - if !(types.PrefixCompare(keys[i], f.packed[0]) > 0 && types.PrefixCompare(keys[i], f.packed[1]) <= 0) { - skipMask.Add(uint64(i)) - } - - case PrefixRangeRightOpen: - if !(types.PrefixCompare(keys[i], f.packed[0]) >= 0 && types.PrefixCompare(keys[i], f.packed[1]) < 0) { - skipMask.Add(uint64(i)) - } - - case PrefixRangeBothOpen: - if !(types.PrefixCompare(keys[i], f.packed[0]) > 0 && types.PrefixCompare(keys[i], f.packed[1]) < 0) { - skipMask.Add(uint64(i)) - } - - case function.GREAT_EQUAL: - if !(bytes.Compare(keys[i], f.packed[0]) >= 0) { - skipMask.Add(uint64(i)) - } - - case function.GREAT_THAN: - if !(bytes.Compare(keys[i], f.packed[0]) > 0) { - skipMask.Add(uint64(i)) - } - - case function.LESS_EQUAL: - if !(bytes.Compare(keys[i], f.packed[0]) <= 0) { - skipMask.Add(uint64(i)) - } - - case function.LESS_THAN: - if !(bytes.Compare(keys[i], f.packed[0]) < 0) { - skipMask.Add(uint64(i)) - } - - default: - // skip nothing + if !f.matches(keys[i]) { + skipMask.Add(uint64(i)) } } } diff --git a/pkg/vm/engine/readutil/pk_filter_mem_test.go b/pkg/vm/engine/readutil/pk_filter_mem_test.go index 30eb82d153372..26daf089c9687 100644 --- a/pkg/vm/engine/readutil/pk_filter_mem_test.go +++ b/pkg/vm/engine/readutil/pk_filter_mem_test.go @@ -15,7 +15,10 @@ package readutil import ( + "fmt" "math" + "strconv" + "sync" "testing" "github.com/matrixorigin/matrixone/pkg/common/mpool" @@ -289,6 +292,190 @@ func TestMemPKFilter_FilterVector(t *testing.T) { } } +func TestMemPKFilterPrefixInScales(t *testing.T) { + mp := mpool.MustNew(t.Name()) + defer func() { require.Zero(t, mp.CurrNB()) }() + + tableDef := &plan.TableDef{ + Name: "test", + Pkey: &plan.PrimaryKeyDef{ + Names: []string{"pk"}, + PkeyColName: "pk", + }, + Cols: []*plan.ColDef{{ + Name: "pk", + Typ: plan.Type{Id: int32(types.T_varchar)}, + }}, + } + packerPool := fileservice.NewPool( + 8, + func() *types.Packer { return types.NewPacker() }, + func(packer *types.Packer) { packer.Reset() }, + func(packer *types.Packer) { packer.Close() }, + ) + + for _, count := range []int{0, 1, 4, 5, 64, 1024} { + t.Run(strconv.Itoa(count), func(t *testing.T) { + prefixes := vector.NewVec(types.T_varchar.ToType()) + defer prefixes.Free(mp) + for idx := 0; idx < count; idx++ { + require.NoError(t, vector.AppendBytes(prefixes, []byte(fmt.Sprintf("p%04d", idx)), false, mp)) + } + + filter, err := NewMemPKFilter( + tableDef, + types.MaxTs().ToTimestamp(), + packerPool, + BasePKFilter{Valid: true, Op: function.PREFIX_IN, Oid: types.T_varchar, Vec: prefixes}, + engine.FilterHint{}, + ) + require.NoError(t, err) + require.True(t, filter.Valid()) + + input := vector.NewVec(types.T_varchar.ToType()) + defer input.Free(mp) + require.NoError(t, vector.AppendBytes(input, []byte("p0000-tail"), false, mp)) + last := max(count-1, 0) + require.NoError(t, vector.AppendBytes(input, []byte(fmt.Sprintf("p%04d-tail", last)), false, mp)) + require.NoError(t, vector.AppendBytes(input, []byte("miss"), false, mp)) + + packer := types.NewPacker() + defer packer.Close() + skipMask := objectio.GetReusableBitmap() + defer skipMask.Release() + filter.FilterVector(input, packer, &skipMask) + + wantSkipped := 1 + if count == 0 { + wantSkipped = 3 + } + require.Equal(t, wantSkipped, skipMask.Count()) + }) + } + + t.Run("overlap duplicate and empty", func(t *testing.T) { + for _, test := range []struct { + name string + prefixes []string + wantSkipped int + }{ + {name: "overlap", prefixes: []string{"ab", "a", "b", "a"}, wantSkipped: 1}, + {name: "empty prefix", prefixes: []string{"z", ""}, wantSkipped: 0}, + } { + t.Run(test.name, func(t *testing.T) { + prefixes := vector.NewVec(types.T_varchar.ToType()) + defer prefixes.Free(mp) + for _, prefix := range test.prefixes { + require.NoError(t, vector.AppendBytes(prefixes, []byte(prefix), false, mp)) + } + filter, err := NewMemPKFilter( + tableDef, + types.MaxTs().ToTimestamp(), + packerPool, + BasePKFilter{Valid: true, Op: function.PREFIX_IN, Oid: types.T_varchar, Vec: prefixes}, + engine.FilterHint{}, + ) + require.NoError(t, err) + + input := vector.NewVec(types.T_varchar.ToType()) + defer input.Free(mp) + for _, value := range []string{"ax", "abx", "bx", "cx"} { + require.NoError(t, vector.AppendBytes(input, []byte(value), false, mp)) + } + packer := types.NewPacker() + defer packer.Close() + skipMask := objectio.GetReusableBitmap() + defer skipMask.Release() + filter.FilterVector(input, packer, &skipMask) + require.Equal(t, test.wantSkipped, skipMask.Count()) + }) + } + }) +} + +func TestMemPKFilterDisjunctsAndConcurrentRead(t *testing.T) { + mp := mpool.MustNew(t.Name()) + defer func() { require.Zero(t, mp.CurrNB()) }() + tableDef := &plan.TableDef{ + Name: "test", + Pkey: &plan.PrimaryKeyDef{ + Names: []string{"pk"}, + PkeyColName: "pk", + }, + Cols: []*plan.ColDef{{Name: "pk", Typ: plan.Type{Id: int32(types.T_int64)}}}, + } + packerPool := fileservice.NewPool( + 8, + func() *types.Packer { return types.NewPacker() }, + func(packer *types.Packer) { packer.Reset() }, + func(packer *types.Packer) { packer.Close() }, + ) + filter, err := NewMemPKFilter( + tableDef, + types.MaxTs().ToTimestamp(), + packerPool, + BasePKFilter{ + Valid: true, + Disjuncts: []BasePKFilter{ + {Valid: true, Op: function.EQUAL, Oid: types.T_int64, LB: types.EncodeFixed(int64(1))}, + {Valid: true, Op: function.BETWEEN, Oid: types.T_int64, LB: types.EncodeFixed(int64(4)), UB: types.EncodeFixed(int64(6))}, + }, + }, + engine.FilterHint{}, + ) + require.NoError(t, err) + require.True(t, filter.Valid()) + require.Len(t, filter.Specs(), 2) + + input := vector.NewVec(types.T_int64.ToType()) + defer input.Free(mp) + for value := int64(0); value < 10; value++ { + require.NoError(t, vector.AppendFixed(input, value, false, mp)) + } + + const readers = 16 + errs := make(chan int, readers) + var wg sync.WaitGroup + for range readers { + wg.Add(1) + go func() { + defer wg.Done() + packer := types.NewPacker() + defer packer.Close() + skipMask := objectio.GetReusableBitmap() + defer skipMask.Release() + filter.FilterVector(input, packer, &skipMask) + if skipMask.Count() != 6 { + errs <- skipMask.Count() + } + }() + } + wg.Wait() + close(errs) + for count := range errs { + require.Equal(t, 6, count) + } + + t.Run("invalid branch fails open", func(t *testing.T) { + invalid, err := NewMemPKFilter( + tableDef, + types.MaxTs().ToTimestamp(), + packerPool, + BasePKFilter{ + Valid: true, + Disjuncts: []BasePKFilter{ + {Valid: true, Op: function.EQUAL, Oid: types.T_int64, LB: types.EncodeFixed(int64(1))}, + {Valid: true, Op: function.EQUAL, Oid: types.T_int64, LB: []byte{1}}, + }, + }, + engine.FilterHint{}, + ) + require.NoError(t, err) + require.False(t, invalid.Valid()) + require.Empty(t, invalid.Specs()) + }) +} + func TestPKFiltersUseSameFailOpenValidation(t *testing.T) { mp := mpool.MustNew(t.Name()) packerPool := fileservice.NewPool( @@ -346,6 +533,18 @@ func TestPKFiltersUseSameFailOpenValidation(t *testing.T) { }) }) + t.Run("PREFIX_IN null fails open", func(t *testing.T) { + values := vector.NewVec(types.T_varchar.ToType()) + require.NoError(t, vector.AppendBytes(values, nil, true, mp)) + defer values.Free(mp) + assertFailOpen(t, types.T_varchar, BasePKFilter{ + Valid: true, + Op: function.PREFIX_IN, + Oid: types.T_varchar, + Vec: values, + }) + }) + for _, test := range []struct { name string oid types.T @@ -408,3 +607,38 @@ func TestPKFiltersUseSameFailOpenValidation(t *testing.T) { require.Zero(t, mp.CurrNB()) } + +var benchmarkMemPKFilterMatches int + +func BenchmarkMemPKFilterPrefixIn(b *testing.B) { + packer := types.NewPacker() + defer packer.Close() + + for _, prefixCount := range []int{4, 64, 1024} { + b.Run(strconv.Itoa(prefixCount), func(b *testing.B) { + prefixes := make([][]byte, 0, prefixCount) + for idx := 0; idx < prefixCount; idx++ { + encoded := EncodePrimaryKey(fmt.Sprintf("p%04d", idx), packer) + prefixes = append(prefixes, encoded[:len(encoded)-1]) + } + var filter MemPKFilter + filter.SetFullData(function.PREFIX_IN, true, prefixes...) + + keys := make([][]byte, 8192) + for idx := range keys { + keys[idx] = EncodePrimaryKey(fmt.Sprintf("p%04d-tail", idx%(prefixCount*2)), packer) + } + b.ReportAllocs() + b.ResetTimer() + matched := 0 + for range b.N { + for idx := range keys { + if filter.matches(keys[idx]) { + matched++ + } + } + } + benchmarkMemPKFilterMatches = matched + }) + } +} From b7fa79d0bc16fece6b3ff5efb61e5e3566683461 Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Sat, 18 Jul 2026 02:04:41 +0800 Subject: [PATCH 2/2] test(disttae): cover early filter fallback paths --- .../early_filter_fallbacks.result | 114 ++++++++++++++++++ .../early_filter_fallbacks.sql | 58 +++++++++ 2 files changed, 172 insertions(+) create mode 100644 test/distributed/cases/disttae/disttae_filters/reader_filters/partition_reader/early_filter_fallbacks.result create mode 100644 test/distributed/cases/disttae/disttae_filters/reader_filters/partition_reader/early_filter_fallbacks.sql diff --git a/test/distributed/cases/disttae/disttae_filters/reader_filters/partition_reader/early_filter_fallbacks.result b/test/distributed/cases/disttae/disttae_filters/reader_filters/partition_reader/early_filter_fallbacks.result new file mode 100644 index 0000000000000..b08b17f8e4063 --- /dev/null +++ b/test/distributed/cases/disttae/disttae_filters/reader_filters/partition_reader/early_filter_fallbacks.result @@ -0,0 +1,114 @@ +drop database if exists early_filter_fallbacks; +create database early_filter_fallbacks; +use early_filter_fallbacks; +create table committed_spk(pk int primary key, v int); +insert into committed_spk values +(1, 1), (2, 2), (3, 3), (4, 4), (5, 5), +(6, 6), (7, 7), (8, 8), (9, 9), (10, 10); +select pk from committed_spk +where pk = 1 or pk between 3 and 5 or pk in (8, 9) +order by pk; +➤ pk[4,32,0] 𝄀 +1 𝄀 +3 𝄀 +4 𝄀 +5 𝄀 +8 𝄀 +9 +select pk from committed_spk +where pk between 2 and 6 or pk between 4 and 8 +order by pk; +➤ pk[4,32,0] 𝄀 +2 𝄀 +3 𝄀 +4 𝄀 +5 𝄀 +6 𝄀 +7 𝄀 +8 +select mo_ctl('dn', 'flush', 'early_filter_fallbacks.committed_spk'); +➤ mo_ctl(dn, flush, early_filter_fallbacks.committed_spk)[12,-1,0] 𝄀 +{ + "method": "Flush", + "result": [ + { + "returnStr": "OK" + } + ] +} + +select sleep(1); +➤ sleep(1)[-6,8,0] 𝄀 +0 +select pk from committed_spk +where pk between 2 and 8 and abs(v - 5) >= 2 +order by pk; +➤ pk[4,32,0] 𝄀 +2 𝄀 +3 𝄀 +7 𝄀 +8 +create table committed_cpk(a varchar(20), b int, v int, primary key(a, b)); +insert into committed_cpk values +('p1', 1, 1), ('p2', 2, 2), ('p3', 3, 3), ('p4', 4, 4), +('p5', 5, 5), ('p6', 6, 6), ('p7', 7, 7), ('p8', 8, 8); +select a, b from committed_cpk +where a in ('p1', 'p2', 'p3', 'p4', 'p5', 'p6') +order by a, b; +➤ a[12,-1,0] ¦ b[4,32,0] 𝄀 +p1 ¦ 1 𝄀 +p2 ¦ 2 𝄀 +p3 ¦ 3 𝄀 +p4 ¦ 4 𝄀 +p5 ¦ 5 𝄀 +p6 ¦ 6 +select a, b from committed_cpk +where a in ('p1', 'p2', 'p3', 'p4', 'p5', 'p6', 'p6', null) +order by a, b; +➤ a[12,-1,0] ¦ b[4,32,0] 𝄀 +p1 ¦ 1 𝄀 +p2 ¦ 2 𝄀 +p3 ¦ 3 𝄀 +p4 ¦ 4 𝄀 +p5 ¦ 5 𝄀 +p6 ¦ 6 +select a, b from committed_cpk +where a = 'p1' or a between 'p3' and 'p5' or a in ('p7', 'p8') +order by a, b; +➤ a[12,-1,0] ¦ b[4,32,0] 𝄀 +p1 ¦ 1 𝄀 +p3 ¦ 3 𝄀 +p4 ¦ 4 𝄀 +p5 ¦ 5 𝄀 +p7 ¦ 7 𝄀 +p8 ¦ 8 +create table workspace_cpk(a varchar(20), b int, v int, primary key(a, b)); +begin; +insert into workspace_cpk values +('p1', 1, 1), ('p2', 2, 2), ('p3', 3, 3), ('p4', 4, 4), +('p5', 5, 5), ('p6', 6, 6), ('p7', 7, 7), ('p8', 8, 8); +select a, b from workspace_cpk +where a in ('p1', 'p2', 'p3', 'p4', 'p5', 'p6') +order by a, b; +➤ a[12,-1,0] ¦ b[4,32,0] 𝄀 +p1 ¦ 1 𝄀 +p2 ¦ 2 𝄀 +p3 ¦ 3 𝄀 +p4 ¦ 4 𝄀 +p5 ¦ 5 𝄀 +p6 ¦ 6 +select a, b from workspace_cpk +where a = 'p1' or a between 'p3' and 'p5' or a in ('p7', 'p8') +order by a, b; +➤ a[12,-1,0] ¦ b[4,32,0] 𝄀 +p1 ¦ 1 𝄀 +p3 ¦ 3 𝄀 +p4 ¦ 4 𝄀 +p5 ¦ 5 𝄀 +p7 ¦ 7 𝄀 +p8 ¦ 8 +rollback; +select count(*) from workspace_cpk; +➤ count(*)[-5,64,0] 𝄀 +0 +drop database early_filter_fallbacks; diff --git a/test/distributed/cases/disttae/disttae_filters/reader_filters/partition_reader/early_filter_fallbacks.sql b/test/distributed/cases/disttae/disttae_filters/reader_filters/partition_reader/early_filter_fallbacks.sql new file mode 100644 index 0000000000000..cce3147301884 --- /dev/null +++ b/test/distributed/cases/disttae/disttae_filters/reader_filters/partition_reader/early_filter_fallbacks.sql @@ -0,0 +1,58 @@ +drop database if exists early_filter_fallbacks; +create database early_filter_fallbacks; +use early_filter_fallbacks; + +-- Committed small batches remain in partition state. Exercise OR union and +-- overlapping branches without relying on persisted block filtering. +create table committed_spk(pk int primary key, v int); +insert into committed_spk values + (1, 1), (2, 2), (3, 3), (4, 4), (5, 5), + (6, 6), (7, 7), (8, 8), (9, 9), (10, 10); +select pk from committed_spk +where pk = 1 or pk between 3 and 5 or pk in (8, 9) +order by pk; +select pk from committed_spk +where pk between 2 and 6 or pk between 4 and 8 +order by pk; + +-- Once persisted, an unsupported conjunct must not discard the supported PK +-- range from early object/block filtering. The residual evaluates both terms. +-- @ignore:0 +select mo_ctl('dn', 'flush', 'early_filter_fallbacks.committed_spk'); +select sleep(1); +select pk from committed_spk +where pk between 2 and 8 and abs(v - 5) >= 2 +order by pk; + +-- More than four compound-PK prefixes used to make MemPKFilter fail open. +create table committed_cpk(a varchar(20), b int, v int, primary key(a, b)); +insert into committed_cpk values + ('p1', 1, 1), ('p2', 2, 2), ('p3', 3, 3), ('p4', 4, 4), + ('p5', 5, 5), ('p6', 6, 6), ('p7', 7, 7), ('p8', 8, 8); +select a, b from committed_cpk +where a in ('p1', 'p2', 'p3', 'p4', 'p5', 'p6') +order by a, b; +select a, b from committed_cpk +where a in ('p1', 'p2', 'p3', 'p4', 'p5', 'p6', 'p6', null) +order by a, b; +select a, b from committed_cpk +where a = 'p1' or a between 'p3' and 'p5' or a in ('p7', 'p8') +order by a, b; + +-- The same predicates must filter uncommitted workspace batches without a +-- mutable lazy cache or a threshold-based full pass-through. +create table workspace_cpk(a varchar(20), b int, v int, primary key(a, b)); +begin; +insert into workspace_cpk values + ('p1', 1, 1), ('p2', 2, 2), ('p3', 3, 3), ('p4', 4, 4), + ('p5', 5, 5), ('p6', 6, 6), ('p7', 7, 7), ('p8', 8, 8); +select a, b from workspace_cpk +where a in ('p1', 'p2', 'p3', 'p4', 'p5', 'p6') +order by a, b; +select a, b from workspace_cpk +where a = 'p1' or a between 'p3' and 'p5' or a in ('p7', 'p8') +order by a, b; +rollback; +select count(*) from workspace_cpk; + +drop database early_filter_fallbacks;