From ab40afc5640b14230bd8f5564c785c5c6d59b001 Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Wed, 30 Sep 2026 01:40:44 +0800 Subject: [PATCH 1/3] Optimize leading ranges on composite sort keys --- pkg/sql/plan/expr_opt.go | 181 +++++++++++++++++++++----- pkg/sql/plan/expr_opt_test.go | 157 ++++++++++++++++++++++ pkg/vm/engine/readutil/filter_test.go | 127 ++++++++++++++++++ 3 files changed, 436 insertions(+), 29 deletions(-) diff --git a/pkg/sql/plan/expr_opt.go b/pkg/sql/plan/expr_opt.go index 30346f1258e8b..0fc6359b07825 100644 --- a/pkg/sql/plan/expr_opt.go +++ b/pkg/sql/plan/expr_opt.go @@ -169,18 +169,21 @@ func (builder *QueryBuilder) appendCompositePartBlockFilters(filters map[int32][ for nodeID, candidates := range filters { node := builder.qry.Nodes[nodeID] for _, candidate := range candidates { - duplicate := false - for _, existing := range node.BlockFilterList { - if blockFilterEquivalent(existing, candidate.copy) { - duplicate = true - break - } - } - if !duplicate { - node.BlockFilterList = append(node.BlockFilterList, candidate.copy) - } + appendUniqueBlockFilter(node, candidate.copy, false) + } + } +} + +func appendUniqueBlockFilter(node *plan.Node, filter *plan.Expr, copyFilter bool) { + for _, existing := range node.BlockFilterList { + if blockFilterEquivalent(existing, filter) { + return } } + if copyFilter { + filter = DeepCopyExpr(filter) + } + node.BlockFilterList = append(node.BlockFilterList, filter) } // retainConsumedCompositePartBlockFilters keeps only predicates removed by the @@ -282,26 +285,145 @@ func (builder *QueryBuilder) appendCompoundKeyBlockFilters(nodeID int32) { return } allowed := map[int32]struct{}{compoundPos: {}} - for _, filter := range node.FilterList { - if !ExprIsZonemappable(builder.GetContext(), filter) || - !exprOnlyReferencesColumns(filter, node.BindingTags[0], allowed) { - continue + var leadingPos int32 = -1 + if node.TableDef.ClusterBy != nil && util.JudgeIsCompositeClusterByColumn(node.TableDef.ClusterBy.Name) { + parts := util.SplitCompositeClusterByColumnName(node.TableDef.ClusterBy.Name) + if len(parts) > 0 { + if pos, found := node.TableDef.Name2ColIndex[parts[0]]; found { + leadingPos = pos + } } - duplicate := false - for _, existing := range node.BlockFilterList { - if blockFilterEquivalent(existing, filter) { - duplicate = true - break + } else if node.TableDef.Pkey != nil && len(node.TableDef.Pkey.Names) > 1 { + if pos, found := node.TableDef.Name2ColIndex[node.TableDef.Pkey.Names[0]]; found { + leadingPos = pos + } + } + for _, filter := range node.FilterList { + zonemappable := ExprIsZonemappable(builder.GetContext(), filter) + if leadingPos >= 0 && zonemappable { + if prefix := builder.leadingCompositeRangeBlockFilter(filter, node.TableDef, node.BindingTags[0], leadingPos, compoundPos); prefix != nil { + appendUniqueBlockFilter(node, prefix, false) } } - if !duplicate { - node.BlockFilterList = append(node.BlockFilterList, DeepCopyExpr(filter)) + if !zonemappable || !exprOnlyReferencesColumns(filter, node.BindingTags[0], allowed) { + continue } + appendUniqueBlockFilter(node, filter, true) } } visit(nodeID) } +// leadingCompositeRangeBlockFilter adds an object-pruning predicate without +// replacing the SQL row predicate. In particular, NULL and unsupported bound +// types must continue through the ordinary row comparison. +func (builder *QueryBuilder) leadingCompositeRangeBlockFilter(filter *plan.Expr, tableDef *plan.TableDef, tag, leadingPos, compoundPos int32) *plan.Expr { + fn := filter.GetF() + if fn == nil || fn.Func == nil || len(fn.Args) < 2 || fn.Args[0] == nil { + return nil + } + op := fn.Func.ObjName + col := fn.Args[0] + if op == "<" || op == "<=" || op == ">" || op == ">=" { + op = canonicalRangeOp(fn) + if fn.Args[0].GetCol() == nil && len(fn.Args) == 2 { + col = fn.Args[1] + } + } + if col.GetCol() == nil || col.GetCol().RelPos != tag || col.GetCol().ColPos != leadingPos || + !compositeRangeOrderPreserving(types.T(col.Typ.Id)) { + return nil + } + boundCompatible := func(bound *plan.Expr) bool { + if bound == nil { + return false + } + if lit := stripConstLiteralCasts(bound).GetLit(); lit != nil && lit.Isnull { + return false + } + return isRuntimeConstExpr(bound) && bound.Typ.Id == col.Typ.Id && + bound.Typ.Scale == col.Typ.Scale + } + key := &plan.Expr{Typ: tableDef.Cols[compoundPos].Typ, Expr: &plan.Expr_Col{Col: &plan.ColRef{ + RelPos: tag, ColPos: compoundPos, Name: tableDef.Cols[compoundPos].Name, + }}} + serial := func(bound *plan.Expr) *plan.Expr { + ret, ok := builder.bindCompositeKeySerial([]*plan.Expr{bound}) + if !ok { + return nil + } + return ret + } + var name string + var args []*plan.Expr + switch op { + case "between", "in_range": + if (op == "between" && len(fn.Args) != 3) || + (op == "in_range" && len(fn.Args) != 4) { + return nil + } + if !boundCompatible(fn.Args[1]) || !boundCompatible(fn.Args[2]) { + return nil + } + lower, upper := serial(fn.Args[1]), serial(fn.Args[2]) + if lower == nil || upper == nil { + return nil + } + name = "prefix_between" + args = []*plan.Expr{key, lower, upper} + if op == "in_range" { + if fn.Args[3] == nil || !isRuntimeConstExpr(fn.Args[3]) { + return nil + } + name = "prefix_in_range" + args = append(args, fn.Args[3]) + } + case "<", "<=", ">", ">=": + boundExpr := rangeFilterConstValue(fn) + if len(fn.Args) != 2 || !boundCompatible(boundExpr) { + return nil + } + bound := serial(boundExpr) + if bound == nil { + return nil + } + empty := makePlan2StringConstExprWithType("") + empty.Typ.Id = int32(types.T_varchar) + var flag byte + if op == "<" || op == "<=" { + args = []*plan.Expr{key, empty, bound} + if op == "<" { + flag = 2 + } + } else { + args = []*plan.Expr{key, bound, empty} + if op == ">" { + flag = 1 + } + } + name = "prefix_in_range" + args = append(args, makePlan2Uint8ConstExprWithType(flag)) + default: + return nil + } + ret, ok := builder.bindCompositeKeyPredicate(name, args...) + if !ok { + return nil + } + return ret +} + +func compositeRangeOrderPreserving(oid types.T) bool { + switch oid { + case types.T_int8, types.T_int16, types.T_int32, types.T_int64, + types.T_uint8, types.T_uint16, types.T_uint32, types.T_uint64, + types.T_date, types.T_time, types.T_datetime, types.T_timestamp, + types.T_decimal64, types.T_decimal128, types.T_decimal256: + return true + } + return false +} + func existingCompositeBlockFilters(node *plan.Node) []*plan.Expr { if node.TableDef == nil || len(node.BindingTags) == 0 { return nil @@ -2239,6 +2361,11 @@ func (builder *QueryBuilder) doMergeFiltersOnCompositeKey(tableDef *plan.TableDe if _, ok := sortKeyPartCols[col.ColPos]; !ok { continue } + // Keep first-component SQL bounds intact. Their optional compound-key + // object filter is added separately without casting or consuming them. + if col.ColPos == tableDef.Name2ColIndex[Parts[0]] { + continue + } if isLower { colLowerBounds[col.ColPos] = i } else { @@ -2360,6 +2487,10 @@ func (builder *QueryBuilder) doMergeFiltersOnCompositeKey(tableDef *plan.TableDe return filters } lastFuncName := lastFn.Func.ObjName + if len(filterIdx) == 1 && (lastFuncName == "between" || lastFuncName == "in_range" || + lastFuncName == "<" || lastFuncName == "<=" || lastFuncName == ">" || lastFuncName == ">=") { + return filters + } if lastFuncName == "in" { if !hasNonNilFunctionArgs(lastFn, 2) { return filters @@ -2448,10 +2579,6 @@ func (builder *QueryBuilder) doMergeFiltersOnCompositeKey(tableDef *plan.TableDe serialArgs[i] = filters[filterIdx[i]].GetF().Args[1] } - if len(filterIdx) < numParts && len(serialArgs) == 0 { - return filters - } - tmpSerialArgs := DeepCopyExprList(serialArgs) tmpSerialArgs = append(tmpSerialArgs, lastFn.Args[1]) leftArg, ok := builder.bindCompositeKeySerial(tmpSerialArgs) @@ -2488,10 +2615,6 @@ func (builder *QueryBuilder) doMergeFiltersOnCompositeKey(tableDef *plan.TableDe serialArgs[i] = filters[filterIdx[i]].GetF().Args[1] } - if len(filterIdx) < numParts && len(serialArgs) == 0 { - return filters - } - tmpSerialArgs := append(DeepCopyExprList(serialArgs), lastFn.Args[1]) boundArg, ok := builder.bindCompositeKeySerial(tmpSerialArgs) if !ok { diff --git a/pkg/sql/plan/expr_opt_test.go b/pkg/sql/plan/expr_opt_test.go index e808d05a4a45a..29acf25a49d2f 100644 --- a/pkg/sql/plan/expr_opt_test.go +++ b/pkg/sql/plan/expr_opt_test.go @@ -15,6 +15,7 @@ package plan import ( + "math" "testing" "github.com/matrixorigin/matrixone/pkg/common/mpool" @@ -22,6 +23,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/container/types" "github.com/matrixorigin/matrixone/pkg/container/vector" planpb "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect/mysql" "github.com/matrixorigin/matrixone/pkg/sql/util" "github.com/stretchr/testify/require" ) @@ -58,6 +60,161 @@ func TestDoMergeFiltersOnCompositeKeyMergesSortKeyRanges(t *testing.T) { requireFuncNames(t, ret, "in_range") } +func TestLeadingCompositeRangeKeepsRowFilterAndAddsObjectFilter(t *testing.T) { + for _, key := range []struct { + name string + table *planpb.TableDef + }{ + {"primary", makeExprOptCompositeSortKeyTableDef()}, + {"cluster", makeExprOptCompositeClusterKeyTableDef()}, + } { + for _, op := range []string{"<", "<=", ">", ">=", "between", "in_range"} { + t.Run(key.name+"/"+op, func(t *testing.T) { + ctx := NewMockCompilerContext(true) + builder := NewQueryBuilder(planpb.Query_SELECT, ctx, false, false) + tag := builder.genNewBindTag() + col := makeExprOptInt64Col(tag, 0, "a") + var source *planpb.Expr + switch op { + case "between": + source = makeBetweenExpr(col, MakePlan2Int64ConstExprWithType(2), MakePlan2Int64ConstExprWithType(4)) + case "in_range": + source = makeInRangeExpr(col, MakePlan2Int64ConstExprWithType(2), MakePlan2Int64ConstExprWithType(4), 3) + default: + source = makeExprOptBinaryInt64Expr(t, ctx, op, col, 2) + } + builder.qry.Nodes = []*planpb.Node{{ + NodeType: planpb.Node_TABLE_SCAN, BindingTags: []int32{tag}, + TableDef: key.table, FilterList: []*planpb.Expr{source}, + }} + builder.qry.Nodes[0].FilterList = builder.doMergeFiltersOnCompositeKey(key.table, tag, source) + require.Same(t, source, builder.qry.Nodes[0].FilterList[0]) + builder.appendCompoundKeyBlockFilters(0) + builder.appendCompoundKeyBlockFilters(0) + require.Len(t, builder.qry.Nodes[0].BlockFilterList, 1) + want := "prefix_in_range" + if op == "between" { + want = "prefix_between" + } + require.Equal(t, want, builder.qry.Nodes[0].BlockFilterList[0].GetF().Func.ObjName) + prefixArgs := builder.qry.Nodes[0].BlockFilterList[0].GetF().Args + require.Equal(t, key.table.Cols[prefixArgs[0].GetCol().ColPos].Name, prefixArgs[0].GetCol().Name) + encode := func(v int64) string { + packer := types.NewPacker() + defer packer.Close() + packer.EncodeInt64(v) + return string(packer.GetBuf()) + } + lower, upper := "", "" + switch op { + case "between", "in_range": + lower, upper = encode(2), encode(4) + case "<", "<=": + upper = encode(2) + case ">", ">=": + lower = encode(2) + } + require.Equal(t, lower, prefixArgs[1].GetLit().GetSval()) + require.Equal(t, upper, prefixArgs[2].GetLit().GetSval()) + require.Len(t, existingCompositeBlockFilters(builder.qry.Nodes[0]), 1) + }) + } + } +} + +func TestLeadingCompositeRangeRejectsUnsafeEncoding(t *testing.T) { + for _, tc := range []struct { + name string + typ types.T + scale int32 + lower *planpb.Expr + upper *planpb.Expr + }{ + {"float signed-zero risk", types.T_float64, 0, makePlan2Float64ConstExprWithType(math.Copysign(0, -1)), makePlan2Float64ConstExprWithType(1)}, + {"string prefix risk", types.T_varchar, 0, makePlan2StringConstExprWithType("a"), makePlan2StringConstExprWithType("z")}, + {"decimal scale mismatch", types.T_decimal64, 2, makeDecimal64Const(1.001, 3), makeDecimal64Const(2.002, 3)}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := NewMockCompilerContext(true) + builder := NewQueryBuilder(planpb.Query_SELECT, ctx, false, false) + tag := builder.genNewBindTag() + table := makeExprOptCompositeClusterKeyTableDef() + table.Cols[0].Typ.Id = int32(tc.typ) + table.Cols[0].Typ.Scale = tc.scale + col := makeExprOptInt64Col(tag, 0, "a") + col.Typ = table.Cols[0].Typ + filter := makeInRangeExpr(col, tc.lower, tc.upper, 2) + builder.qry.Nodes = []*planpb.Node{{NodeType: planpb.Node_TABLE_SCAN, BindingTags: []int32{tag}, TableDef: table, FilterList: []*planpb.Expr{filter}}} + builder.appendCompoundKeyBlockFilters(0) + require.Empty(t, builder.qry.Nodes[0].BlockFilterList) + }) + } +} + +func TestLeadingCompositePairedBoundsRemainIndependent(t *testing.T) { + ctx := NewMockCompilerContext(true) + builder := NewQueryBuilder(planpb.Query_SELECT, ctx, false, false) + tag := builder.genNewBindTag() + table := makeExprOptCompositeClusterKeyTableDef() + lower := makeExprOptBinaryInt64Expr(t, ctx, ">", makeExprOptInt64Col(tag, 0, "a"), 2) + upper := makeExprOptBinaryInt64Expr(t, ctx, "<", makeExprOptInt64Col(tag, 0, "a"), 4) + builder.qry.Nodes = []*planpb.Node{{NodeType: planpb.Node_TABLE_SCAN, BindingTags: []int32{tag}, TableDef: table, FilterList: []*planpb.Expr{lower, upper}}} + ret := builder.doMergeFiltersOnCompositeKey(table, tag, lower, upper) + require.Same(t, lower, ret[0]) + require.Same(t, upper, ret[1]) + builder.qry.Nodes[0].FilterList = ret + builder.appendCompoundKeyBlockFilters(0) + require.Len(t, builder.qry.Nodes[0].BlockFilterList, 2) + requireFuncNames(t, builder.qry.Nodes[0].BlockFilterList, "prefix_in_range") +} + +func TestLeadingCompositeRangeWithReversedOperands(t *testing.T) { + ctx := NewMockCompilerContext(true) + builder := NewQueryBuilder(planpb.Query_SELECT, ctx, false, false) + tag := builder.genNewBindTag() + table := makeExprOptCompositeClusterKeyTableDef() + filter, err := BindFuncExprImplByPlanExpr(ctx.GetContext(), "<", []*planpb.Expr{ + MakePlan2Int64ConstExprWithType(2), makeExprOptInt64Col(tag, 0, "a"), + }) + require.NoError(t, err) + builder.qry.Nodes = []*planpb.Node{{NodeType: planpb.Node_TABLE_SCAN, BindingTags: []int32{tag}, TableDef: table, FilterList: []*planpb.Expr{filter}}} + builder.appendCompoundKeyBlockFilters(0) + require.Len(t, builder.qry.Nodes[0].BlockFilterList, 1) + prefix := builder.qry.Nodes[0].BlockFilterList[0].GetF() + require.Equal(t, "prefix_in_range", prefix.Func.ObjName) + require.Equal(t, uint32(1), prefix.Args[3].GetLit().GetU8Val()) +} + +func TestLeadingCompositeRangeReachableFromSQL(t *testing.T) { + ctx := NewMockCompilerContext(true) + table := makeExprOptCompositeClusterKeyTableDef() + table.Name = "range_probe" + table.TblId = 29507 + ctx.tables[table.Name] = table + ctx.objects[table.Name] = &planpb.ObjectRef{Obj: int64(table.TblId), ObjName: table.Name, SchemaName: "tpch"} + for _, sql := range []string{ + "select a from range_probe where a >= 10 and a < 20", + "select a from range_probe where a > 10", + } { + t.Run(sql, func(t *testing.T) { + stmt, err := mysql.ParseOne(ctx.GetContext(), sql, 1) + require.NoError(t, err) + built, err := BuildPlan(ctx, stmt, false) + require.NoError(t, err) + found := false + for _, node := range built.GetQuery().Nodes { + if node.NodeType != planpb.Node_TABLE_SCAN || node.TableDef.Name != table.Name { + continue + } + found = true + require.NotEmpty(t, node.FilterList) + requireFuncNames(t, node.BlockFilterList, "prefix_in_range") + } + require.True(t, found) + }) + } +} + func TestDoMergeFiltersOnCompositeKeySupportsFoldedInVector(t *testing.T) { testCases := []struct { name string diff --git a/pkg/vm/engine/readutil/filter_test.go b/pkg/vm/engine/readutil/filter_test.go index 4a8ee8bce50d7..0c1f4982bf9a0 100644 --- a/pkg/vm/engine/readutil/filter_test.go +++ b/pkg/vm/engine/readutil/filter_test.go @@ -38,6 +38,7 @@ import ( plan2 "github.com/matrixorigin/matrixone/pkg/sql/plan" "github.com/matrixorigin/matrixone/pkg/sql/plan/function" "github.com/matrixorigin/matrixone/pkg/testutil" + "github.com/matrixorigin/matrixone/pkg/vm/engine" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/containers" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/index" @@ -3991,6 +3992,132 @@ func TestCompileFilterExpr_PrefixInRangeAllFlags(t *testing.T) { } } +func TestCompositeLeadingRangeExcludesObjectsBeforeMetadataLoad(t *testing.T) { + encode := func(first, second int64) []byte { + packer := types.NewPacker() + defer packer.Close() + packer.EncodeInt64(first) + packer.EncodeInt64(second) + return append([]byte(nil), packer.GetBuf()...) + } + bound := func(value int64) *plan.Expr { + packer := types.NewPacker() + defer packer.Close() + packer.EncodeInt64(value) + return plan2.MakePlan2StringConstExprWithType(string(packer.GetBuf()), true) + } + table := &plan.TableDef{ + Name2ColIndex: map[string]int32{"a": 0, "__mo_cpkey": 1}, + Pkey: &plan.PrimaryKeyDef{PkeyColName: "__mo_cpkey", Names: []string{"a", "b"}}, + Cols: []*plan.ColDef{ + {Name: "a", Seqnum: 0, Typ: plan.Type{Id: int32(types.T_int64)}}, + {Name: "__mo_cpkey", Seqnum: 1, Primary: true, Typ: plan.Type{Id: int32(types.T_varchar)}}, + }, + } + expr := MakeFunctionExprForTest("prefix_in_range", []*plan.Expr{ + MakeColExprForTest(1, types.T_varchar, "__mo_cpkey"), + bound(10), bound(20), plan2.MakePlan2Uint8ConstExprWithType(2), + }) + proc := testutil.NewProcess(t) + defer proc.Free() + var exes []colexec.ExpressionExecutor + plan2.ReplaceFoldExpr(proc, expr, &exes) + plan2.EvalFoldExpr(proc, expr, &exes) + for _, exe := range exes { + defer exe.Free() + } + fast, load, _, _, _, canCompile, _ := CompileFilterExpr(expr, table, nil) + require.True(t, canCompile) + require.NotNil(t, fast) + require.NotNil(t, load) + var objects []objectio.ObjectStats + for i, tc := range []struct { + name string + first int64 + expected bool + }{ + {"below", 9, false}, {"lower", 10, true}, {"inside", 15, true}, + {"upper-open", 20, false}, {"above", 21, false}, + } { + t.Run(tc.name, func(t *testing.T) { + stats := filterTestObject(t, byte(i+1)) + zm := index.NewZM(types.T_varchar, 0) + index.UpdateZM(zm, encode(tc.first, 1)) + index.UpdateZM(zm, encode(tc.first, 2)) + require.NoError(t, objectio.SetObjectStatsSortKeyZoneMap(&stats, zm)) + selected, err := fast(&stats) + require.NoError(t, err) + require.Equal(t, tc.expected, selected) + objects = append(objects, stats) + }) + } + nullFirst := filterTestObject(t, 6) + packer := types.NewPacker() + packer.EncodeNull() + packer.EncodeInt64(1) + nullZM := index.NewZM(types.T_varchar, 0) + index.UpdateZM(nullZM, packer.GetBuf()) + packer.Close() + require.NoError(t, objectio.SetObjectStatsSortKeyZoneMap(&nullFirst, nullZM)) + objects = append(objects, nullFirst) + loads := 0 + countLoad := func(_ context.Context, _ *objectio.ObjectStats, meta objectio.ObjectMeta, bf objectio.BloomFilter) (objectio.ObjectMeta, objectio.BloomFilter, error) { + loads++ + return meta, bf, nil + } + var blocks objectio.BlockInfoSlice + _, loadHit, _, _, _, _, fastTotal, fastHit, err := FilterObjects( + context.Background(), engine.RangesParam{}, fast, countLoad, nil, nil, nil, + nil, objects, nil, &blocks, false, nil, nil, + ) + require.NoError(t, err) + require.Equal(t, len(objects), fastTotal) + require.Equal(t, 4, fastHit) + require.Equal(t, 2, loadHit) + require.Equal(t, loadHit, loads) + for _, tc := range []struct { + name string + lower *plan.Expr + upper *plan.Expr + flag uint8 + selected []bool + }{ + {"less", nil, bound(10), 2, []bool{true, false, false, false, false, true}}, + {"less-or-equal", nil, bound(10), 0, []bool{true, true, false, false, false, true}}, + {"greater", bound(20), nil, 1, []bool{false, false, false, false, true, false}}, + {"greater-or-equal", bound(20), nil, 0, []bool{false, false, false, true, true, false}}, + } { + t.Run(tc.name, func(t *testing.T) { + empty := plan2.MakePlan2StringConstExprWithType("", true) + empty.Typ.Id = int32(types.T_varchar) + if tc.lower == nil { + tc.lower = empty + } + if tc.upper == nil { + tc.upper = empty + } + expr := MakeFunctionExprForTest("prefix_in_range", []*plan.Expr{ + MakeColExprForTest(1, types.T_varchar, "__mo_cpkey"), + tc.lower, tc.upper, plan2.MakePlan2Uint8ConstExprWithType(tc.flag), + }) + var folded []colexec.ExpressionExecutor + _, err := plan2.ReplaceFoldExpr(proc, expr, &folded) + require.NoError(t, err) + require.NoError(t, plan2.EvalFoldExpr(proc, expr, &folded)) + for _, exe := range folded { + exe.Free() + } + oneSidedFast, _, _, _, _, compiled, _ := CompileFilterExpr(expr, table, nil) + require.True(t, compiled) + for i := range objects { + selected, err := oneSidedFast(&objects[i]) + require.NoError(t, err) + require.Equal(t, tc.selected[i], selected) + } + }) + } +} + func TestCompileFilterExprsPreservesSupportedConjuncts(t *testing.T) { tableDef := &plan.TableDef{ Name: "test_tbl", From d1220a51c547219fdf8a8059a14890f40981c4ba Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Wed, 30 Sep 2026 02:10:02 +0800 Subject: [PATCH 2/3] Avoid reading block-filter-only scan columns --- pkg/partitionprune/filter.go | 56 ++++++++++++++++++++-- pkg/partitionprune/filter_test.go | 41 ++++++++++++++++ pkg/sql/plan/expr_opt_test.go | 74 +++++++++++++++++++++++++++-- pkg/sql/plan/query_builder.go | 42 ++++++++++++++-- pkg/vm/engine/readutil/expr_util.go | 9 ++-- 5 files changed, 207 insertions(+), 15 deletions(-) diff --git a/pkg/partitionprune/filter.go b/pkg/partitionprune/filter.go index bd152680c9ac8..208c3effcf331 100644 --- a/pkg/partitionprune/filter.go +++ b/pkg/partitionprune/filter.go @@ -17,6 +17,7 @@ package partitionprune import ( "context" "sort" + "strings" "github.com/matrixorigin/matrixone/pkg/common/moerr" "github.com/matrixorigin/matrixone/pkg/container/batch" @@ -147,7 +148,7 @@ func hashFilterExpr( if !ok { return nil, false, nil } - if left.Col.ColPos != colPosition { + if !matchesPartitionColumn(left.Col, colPosition, metadata.Partitions[0].Expr) { return nil, false, nil } value, canPrune := normalizePartitionValue(exprImpl.F.Args[1]) @@ -262,7 +263,7 @@ func rangeFilterExpr( if !ok { return nil, false, nil } - if left.Col.ColPos != colPosition { + if !matchesPartitionColumn(left.Col, colPosition, metadata.Partitions[0].Expr) { return nil, false, nil } value, canPrune := normalizePartitionValue(exprImpl.F.Args[1]) @@ -460,6 +461,55 @@ func mustGetColPosition(expr *plan.Expr) int32 { return -1 } +// Scan column positions are compacted independently of the stored partition +// expression. Use column identity when the scan supplies a name; retain the +// positional fallback for older nameless expressions. +func matchesPartitionColumn(col *plan.ColRef, position int32, partitionExpr *plan.Expr) bool { + if col.Name != "" { + partitionName, consistent := partitionExpressionColumnName(partitionExpr) + scanName, unambiguous := scanColumnName(col.Name) + return consistent && unambiguous && partitionName != "" && + strings.EqualFold(scanName, partitionName) + } + return col.ColPos == position +} + +// A predicate on one column cannot prune a partition expression that depends +// on a different column or on several columns. +func partitionExpressionColumnName(expr *plan.Expr) (string, bool) { + if expr == nil { + return "", true + } + if col := expr.GetCol(); col != nil { + return col.Name, col.Name != "" + } + if fn := expr.GetF(); fn != nil { + var name string + for _, arg := range fn.Args { + other, ok := partitionExpressionColumnName(arg) + if !ok || (name != "" && other != "" && !strings.EqualFold(name, other)) { + return "", false + } + if other != "" { + name = other + } + } + return name, true + } + return "", true +} + +// The planner emits alias.column. More than one dot is ambiguous because the +// quoted alias or the physical column name may itself contain a dot. In that +// case partition pruning must leave the filter to the row reader. +func scanColumnName(name string) (string, bool) { + if idx := strings.IndexByte(name, '.'); idx >= 0 { + column := name[idx+1:] + return column, !strings.ContainsRune(column, '.') + } + return name, true +} + // listFilter handles partition pruning for list-based partitioning. // It evaluates the filters against list partition expressions and returns matching partition positions. func listFilter( @@ -622,7 +672,7 @@ func listFilterExprNormalized( if !ok { return nil, false, nil } - if left.Col.ColPos != colPosition { + if !matchesPartitionColumn(left.Col, colPosition, metadata.Partitions[0].Expr) { return nil, false, nil } left.Col.ColPos = 0 diff --git a/pkg/partitionprune/filter_test.go b/pkg/partitionprune/filter_test.go index f02a701c10f61..a97303dd6b88d 100644 --- a/pkg/partitionprune/filter_test.go +++ b/pkg/partitionprune/filter_test.go @@ -52,6 +52,30 @@ func TestFilter(t *testing.T) { want []int wantErr bool }{ + { + name: "different column sharing scan position cannot prune partitions", + filters: []*plan.Expr{makeNamedEqualExpr(0, "hidden_key", 1)}, + metadata: namedRangePartitionMetadata("a"), + want: []int{0, 1, 2}, + }, + { + name: "same column at compact scan position prunes partitions", + filters: []*plan.Expr{makeNamedEqualExpr(1, "range_probe.a", 1)}, + metadata: namedRangePartitionMetadata("a"), + want: []int{1}, + }, + { + name: "dotted column name must not alias partition column", + filters: []*plan.Expr{makeNamedEqualExpr(0, "t.a.b", 1)}, + metadata: namedRangePartitionMetadata("b"), + want: []int{0, 1, 2}, + }, + { + name: "dotted alias cannot be distinguished from dotted partition column", + filters: []*plan.Expr{makeNamedEqualExpr(1, "x.a.b", 1)}, + metadata: namedRangePartitionMetadata("a.b"), + want: []int{0, 1, 2}, + }, { name: "empty filters", filters: []*plan.Expr{}, @@ -478,6 +502,23 @@ func makeEqualExpr(colPos int32, value int64) *plan.Expr { } } +func makeNamedEqualExpr(colPos int32, name string, value int64) *plan.Expr { + expr := makeEqualExpr(colPos, value) + expr.GetF().Args[0].GetCol().Name = name + return expr +} + +func namedRangePartitionMetadata(name string) partition.PartitionMetadata { + return partition.PartitionMetadata{ + Method: partition.PartitionMethod_Range, + Partitions: []partition.Partition{ + {Position: 0, Expr: newTestRangeExpr(name, 0)}, + {Position: 1, Expr: newTestRangeExpr(name, 1)}, + {Position: 2, Expr: newTestRangeExpr(name, 2)}, + }, + } +} + func makeEqualExprInt32(colPos int32, value int32) *plan.Expr { return &plan.Expr{ Typ: plan.Type{Id: int32(types.T_bool)}, diff --git a/pkg/sql/plan/expr_opt_test.go b/pkg/sql/plan/expr_opt_test.go index 29acf25a49d2f..c7f93f1cc3315 100644 --- a/pkg/sql/plan/expr_opt_test.go +++ b/pkg/sql/plan/expr_opt_test.go @@ -190,14 +190,21 @@ func TestLeadingCompositeRangeReachableFromSQL(t *testing.T) { table := makeExprOptCompositeClusterKeyTableDef() table.Name = "range_probe" table.TblId = 29507 + for i, col := range table.Cols { + col.Seqnum = uint32(i + 1) + } ctx.tables[table.Name] = table ctx.objects[table.Name] = &planpb.ObjectRef{Obj: int64(table.TblId), ObjName: table.Name, SchemaName: "tpch"} - for _, sql := range []string{ - "select a from range_probe where a >= 10 and a < 20", - "select a from range_probe where a > 10", + for _, tc := range []struct { + sql string + readCols []string + }{ + {"select a from range_probe where a >= 10 and a < 20", []string{"a"}}, + {"select a from range_probe where a > 10", []string{"a"}}, + {"select b from range_probe where a > 10", []string{"a", "b"}}, } { - t.Run(sql, func(t *testing.T) { - stmt, err := mysql.ParseOne(ctx.GetContext(), sql, 1) + t.Run(tc.sql, func(t *testing.T) { + stmt, err := mysql.ParseOne(ctx.GetContext(), tc.sql, 1) require.NoError(t, err) built, err := BuildPlan(ctx, stmt, false) require.NoError(t, err) @@ -209,12 +216,69 @@ func TestLeadingCompositeRangeReachableFromSQL(t *testing.T) { found = true require.NotEmpty(t, node.FilterList) requireFuncNames(t, node.BlockFilterList, "prefix_in_range") + readCols := make([]string, len(node.TableDef.Cols)) + for i, col := range node.TableDef.Cols { + readCols[i] = col.Name + } + require.Equal(t, tc.readCols, readCols, "block-only composite key must not become a reader attribute") + blockKey := node.BlockFilterList[0].GetF().Args[0].GetCol() + require.Equal(t, table.Name+"."+table.ClusterBy.Name, blockKey.Name) + require.GreaterOrEqual(t, blockKey.ColPos, int32(len(readCols))) + columnMap := make(map[int]int) + blockFilters := node.BlockFilterList + if len(readCols) == 2 { + // Runtime filters use compact row positions in this same map. + rowCol := makeExprOptInt64Col(0, 1, table.Name+".b") + blockFilters = append([]*planpb.Expr{rowCol}, blockFilters...) + } + GetColumnMapByExprs(blockFilters, table, columnMap) + require.Equal(t, int(table.Cols[table.Name2ColIndex[table.ClusterBy.Name]].Seqnum), columnMap[int(blockKey.ColPos)]) + if len(readCols) == 2 { + require.Equal(t, int(table.Cols[table.Name2ColIndex["b"]].Seqnum), columnMap[1]) + } } require.True(t, found) }) } } +func TestCompositePartBlockFilterDoesNotReadRewrittenPart(t *testing.T) { + ctx := NewMockCompilerContext(true) + table := makeExprOptCompositeClusterKeyTableDef() + table.Name = "part_probe" + table.TblId = 29508 + ctx.tables[table.Name] = table + ctx.objects[table.Name] = &planpb.ObjectRef{Obj: int64(table.TblId), ObjName: table.Name, SchemaName: "tpch"} + stmt, err := mysql.ParseOne(ctx.GetContext(), "select a from part_probe where a = 10 and b = 20", 1) + require.NoError(t, err) + built, err := BuildPlan(ctx, stmt, false) + require.NoError(t, err) + for _, node := range built.GetQuery().Nodes { + if node.NodeType != planpb.Node_TABLE_SCAN || node.TableDef.Name != table.Name { + continue + } + readCols := make([]string, len(node.TableDef.Cols)) + for i, col := range node.TableDef.Cols { + readCols[i] = col.Name + } + require.NotContains(t, readCols, "b") + require.Contains(t, readCols, table.ClusterBy.Name, "row predicate still needs the compound key") + require.NotEmpty(t, node.BlockFilterList) + partFilter := false + for _, filter := range node.BlockFilterList { + for _, arg := range filter.GetF().Args { + if col := arg.GetCol(); col != nil && col.Name == table.Name+".b" { + partFilter = true + require.GreaterOrEqual(t, col.ColPos, int32(len(readCols))) + } + } + } + require.True(t, partFilter, "the omitted part must still contribute a metadata filter") + return + } + t.Fatal("table scan not found") +} + func TestDoMergeFiltersOnCompositeKeySupportsFoldedInVector(t *testing.T) { testCases := []struct { name string diff --git a/pkg/sql/plan/query_builder.go b/pkg/sql/plan/query_builder.go index 97a74dbbebbd6..5532c6f1e49db 100644 --- a/pkg/sql/plan/query_builder.go +++ b/pkg/sql/plan/query_builder.go @@ -1258,8 +1258,19 @@ func (builder *QueryBuilder) remapAllColRefsForConsumer( increaseRefCnt(expr, 1, colRefCnt) } - for _, expr := range node.BlockFilterList { - increaseRefCnt(expr, 1, colRefCnt) + var blockColRefs map[[2]int32]int + if node.NodeType == plan.Node_TABLE_SCAN && len(node.BlockFilterList) > 0 { + // Block filters inspect relation metadata; they do not consume row + // vectors. Keep their columns out of the scan reader unless another + // expression also needs those columns as row data. + blockColRefs = make(map[[2]int32]int) + for _, expr := range node.BlockFilterList { + increaseRefCnt(expr, 1, blockColRefs) + } + } else { + for _, expr := range node.BlockFilterList { + increaseRefCnt(expr, 1, colRefCnt) + } } for _, rfSpec := range node.RuntimeFilterProbeList { @@ -1284,6 +1295,7 @@ func (builder *QueryBuilder) remapAllColRefsForConsumer( } colTag := node.BindingTags[0] + originalCols := node.TableDef.Cols newTableDef := CloneTableDefForPlan(node.TableDef, false) // An external scan that reports parse errors must read the whole @@ -1326,6 +1338,26 @@ func (builder *QueryBuilder) remapAllColRefsForConsumer( for localIdx, global := range internalRemapping.localToGlobal { colMap[[2]int32{0, int32(localIdx)}] = global } + blockColMap := colMap + if blockColRefs != nil { + blockColMap = maps.Clone(colMap) + // Share positions with row columns. Give omitted metadata columns + // separate slots so combined runtime and block filters cannot map + // the same ColPos to different physical columns. + nextPos := int32(len(internalRemapping.localToGlobal)) + for i := range originalCols { + globalRef := [2]int32{colTag, int32(i)} + if blockColRefs[globalRef] == 0 { + continue + } + if _, ok := blockColMap[globalRef]; ok { + continue + } + blockColMap[globalRef] = [2]int32{0, nextPos} + blockColMap[[2]int32{0, nextPos}] = globalRef + nextPos++ + } + } remapInfo.tip = "FilterList" remapInfo.interRemapping = internalRemapping @@ -1340,9 +1372,11 @@ func (builder *QueryBuilder) remapAllColRefsForConsumer( remapInfo.tip = "BlockFilterList" for idx, expr := range node.BlockFilterList { - increaseRefCnt(expr, -1, colRefCnt) + if blockColRefs == nil { + increaseRefCnt(expr, -1, colRefCnt) + } remapInfo.srcExprIdx = idx - err := builder.remapColRefForExpr(expr, colMap, &remapInfo) + err := builder.remapColRefForExpr(expr, blockColMap, &remapInfo) if err != nil { return nil, err } diff --git a/pkg/vm/engine/readutil/expr_util.go b/pkg/vm/engine/readutil/expr_util.go index 8cdb2778edef7..5ef2d8c323a9a 100644 --- a/pkg/vm/engine/readutil/expr_util.go +++ b/pkg/vm/engine/readutil/expr_util.go @@ -130,12 +130,15 @@ func getColDefByName(expr *plan.Expr, name string, colPos int32, tableDef *plan. pos = tableDef.Name2ColIndex[name] } common.DoIfDebugEnabled(func() { - if name != tableDef.Cols[colPos].Name { + // ColPos is local to the scan (and can be a metadata-only slot), + // while tableDef is the full relation schema. Validate the name used + // for resolution instead of indexing this schema with ColPos. + if int(pos) >= len(tableDef.Cols) || tableDef.Cols[pos].Name != name[strings.LastIndexByte(name, '.')+1:] { logutil.Error( "Bad-ColExpr", zap.String("col-name", name), - zap.Int32("col-actual-pos", colPos), - zap.Int32("col-expected-pos", pos), + zap.Int32("scan-col-pos", colPos), + zap.Int32("relation-col-pos", pos), zap.String("col-expr", plan2.FormatExpr(expr, plan2.FormatOption{})), ) } From 9ce8518dd92a79034d5f30fea50c85e110bea116 Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Wed, 30 Sep 2026 08:20:52 +0800 Subject: [PATCH 3/3] test: align composite prefix pruning explain expectations --- test/distributed/cases/join/leftjoin.result | 4 ++-- test/distributed/cases/optimizer/blockfilter.result | 12 ++++++------ 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/test/distributed/cases/join/leftjoin.result b/test/distributed/cases/join/leftjoin.result index ab7fc3460d281..02f577c1f7b40 100644 --- a/test/distributed/cases/join/leftjoin.result +++ b/test/distributed/cases/join/leftjoin.result @@ -63,7 +63,7 @@ explain select t3.c1 from t3 left join t1 on t1.c1 =t3.c1 and t1.c1 > t3.c2 wher Project 𝄀 -> Table Scan on d1.t3 𝄀 Filter Cond: (t3.c1 < 10) 𝄀 - Block Filter Cond: (t3.c1 < 10) + Block Filter Cond: (t3.c1 < 10), prefix_in_range(t3.__mo_cpkey_col) explain select t1.c1,t3.c1 from t3 left join t1 on t1.c1 =t3.c1 where t3.c1<10; ➤ TP QUERY PLAN[12,0,0] 𝄀 Project 𝄀 @@ -72,7 +72,7 @@ Project 𝄀 Join Cond: (t3.c1 = t1.c1) 𝄀 -> Table Scan on d1.t3 𝄀 Filter Cond: (t3.c1 < 10) 𝄀 - Block Filter Cond: (t3.c1 < 10) 𝄀 + Block Filter Cond: (t3.c1 < 10), prefix_in_range(t3.__mo_cpkey_col) 𝄀 -> Table Scan on d1.t1 𝄀 Filter Cond: (t1.c1 < 10) 𝄀 Block Filter Cond: (t1.c1 < 10) diff --git a/test/distributed/cases/optimizer/blockfilter.result b/test/distributed/cases/optimizer/blockfilter.result index ca9dd58de9e13..08b55d685b7aa 100644 --- a/test/distributed/cases/optimizer/blockfilter.result +++ b/test/distributed/cases/optimizer/blockfilter.result @@ -59,7 +59,7 @@ Project 𝄀 Aggregate Functions: starcount(1) 𝄀 -> Table Scan on d1.t1 𝄀 Filter Cond: (t1.c1 > 10) 𝄀 - Block Filter Cond: (t1.c1 > 10) + Block Filter Cond: (t1.c1 > 10), prefix_in_range(t1.__mo_cbkey_002c1002c2) select count(*) from t1 where c1 > 10; ➤ count(*)[-5,64,0] 𝄀 99890 @@ -80,8 +80,8 @@ Project 𝄀 -> Aggregate 𝄀 Aggregate Functions: starcount(1) 𝄀 -> Table Scan on d1.t1 𝄀 - Filter Cond: prefix_between(t1.__mo_cbkey_002c1002c2) 𝄀 - Block Filter Cond: prefix_between(t1.__mo_cbkey_002c1002c2), t1.c1 BETWEEN 1 AND 5 + Filter Cond: t1.c1 BETWEEN 1 AND 5 𝄀 + Block Filter Cond: t1.c1 BETWEEN 1 AND 5, prefix_between(t1.__mo_cbkey_002c1002c2) select count(*) from t1 where c1 between 1 and 5; ➤ count(*)[-5,64,0] 𝄀 50 @@ -147,7 +147,7 @@ Project 𝄀 Aggregate Functions: starcount(1) 𝄀 -> Table Scan on d1.t2 𝄀 Filter Cond: (t2.c1 > 10) 𝄀 - Block Filter Cond: (t2.c1 > 10) + Block Filter Cond: (t2.c1 > 10), prefix_in_range(t2.__mo_cpkey_col) select count(*) from t2 where c1 > 10; ➤ count(*)[-5,64,0] 𝄀 890000 @@ -168,8 +168,8 @@ Project 𝄀 -> Aggregate 𝄀 Aggregate Functions: starcount(1) 𝄀 -> Table Scan on d1.t2 𝄀 - Filter Cond: prefix_between(t2.__mo_cpkey_col) 𝄀 - Block Filter Cond: prefix_between(t2.__mo_cpkey_col), t2.c1 BETWEEN 1 AND 5 + Filter Cond: t2.c1 BETWEEN 1 AND 5 𝄀 + Block Filter Cond: t2.c1 BETWEEN 1 AND 5, prefix_between(t2.__mo_cpkey_col) select count(*) from t2 where c1 between 1 and 5; ➤ count(*)[-5,64,0] 𝄀 50000