From 41c622ed363729404870df11e9dae66e543bf938 Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Sun, 9 Aug 2026 19:54:51 +0800 Subject: [PATCH 1/2] fix(frontend): preserve logical result cardinality --- pkg/frontend/mysql_cmd_executor.go | 4 +- pkg/frontend/mysql_protocol.go | 2 +- pkg/frontend/query_result.go | 119 +++- pkg/frontend/result_cardinality_test.go | 529 ++++++++++++++++++ .../prepared_projection_cardinality.result | 60 ++ .../prepared_projection_cardinality.sql | 49 ++ 6 files changed, 756 insertions(+), 7 deletions(-) create mode 100644 pkg/frontend/result_cardinality_test.go create mode 100644 test/distributed/cases/prepare/prepared_projection_cardinality.result create mode 100644 test/distributed/cases/prepare/prepared_projection_cardinality.sql diff --git a/pkg/frontend/mysql_cmd_executor.go b/pkg/frontend/mysql_cmd_executor.go index ef3a180c293d1..c3d721fa13d4c 100644 --- a/pkg/frontend/mysql_cmd_executor.go +++ b/pkg/frontend/mysql_cmd_executor.go @@ -762,8 +762,8 @@ func getDataFromPipeline(obj FeSession, execCtx *ExecCtx, bat *batch.Batch, crs } tTime := time.Since(begin) n := 0 - if !isPerformStatement(execCtx.stmt) && bat != nil && bat.Vecs[0] != nil { - n = bat.Vecs[0].Length() + if !isPerformStatement(execCtx.stmt) && bat != nil { + n = bat.RowCount() ses.sentRows.Add(int64(n)) } diff --git a/pkg/frontend/mysql_protocol.go b/pkg/frontend/mysql_protocol.go index 9e6ed4678c3e2..51d219073b9da 100644 --- a/pkg/frontend/mysql_protocol.go +++ b/pkg/frontend/mysql_protocol.go @@ -366,7 +366,7 @@ func (mp *MysqlProtocolImpl) GetBool(id PropertyID) bool { } func (mp *MysqlProtocolImpl) Write(execCtx *ExecCtx, crs *perfcounter.CounterSet, bat *batch.Batch) error { - n := bat.Vecs[0].Length() + n := bat.RowCount() //TODO: remove this MRS here //Create a new temporary result set per pipeline thread. mrs := MysqlResultSet{} diff --git a/pkg/frontend/query_result.go b/pkg/frontend/query_result.go index c77f1649cdcba..cffb651df7b92 100644 --- a/pkg/frontend/query_result.go +++ b/pkg/frontend/query_result.go @@ -109,11 +109,114 @@ func initQueryResulConfig(ctx context.Context, ses *Session) error { return err } +func validateQueryResultBatchForPersistence(bat *batch.Batch) (bool, error) { + if bat == nil { + return false, moerr.NewInternalErrorNoCtx("invalid nil query result batch") + } + rows := bat.RowCount() + if rows < 0 { + return false, moerr.NewInternalErrorNoCtxf( + "invalid query result batch row count %d", rows) + } + if len(bat.Vecs) == 0 { + return false, moerr.NewInternalErrorNoCtx( + "invalid query result batch without vectors") + } + + needsNormalization := false + for i, vec := range bat.Vecs { + if vec == nil { + return false, moerr.NewInternalErrorNoCtxf( + "invalid query result batch: vector %d is nil", i) + } + if vec.Length() < 0 { + return false, moerr.NewInternalErrorNoCtxf( + "invalid query result batch: vector %d has negative length %d", + i, vec.Length()) + } + if vec.Length() == rows { + continue + } + // A non-empty const physically owns one value and can be broadcast to + // any requested logical row range. Flat vectors and empty constants do + // not have storage for a mismatched positive row range. + if !vec.IsConst() || (rows > 0 && vec.Length() == 0) { + return false, moerr.NewInternalErrorNoCtxf( + "invalid query result batch: vector %d length %d does not match row count %d", + i, vec.Length(), rows) + } + needsNormalization = true + } + return needsNormalization, nil +} + +// normalizeQueryResultBatchForPersistence returns an object-writer view whose +// vector lengths match the batch's logical row count. Const vectors are +// broadcast values, so only a compact const whose logical length differs needs +// to be duplicated; aligned vectors remain borrowed from the executor batch. +// The returned release function owns only those duplicated vectors. +func normalizeQueryResultBatchForPersistence( + bat *batch.Batch, + mp *mpool.MPool, +) (*batch.Batch, func(), error) { + needsNormalization, err := validateQueryResultBatchForPersistence(bat) + if err != nil { + return nil, nil, err + } + return normalizeValidatedQueryResultBatchForPersistence(bat, mp, needsNormalization) +} + +func normalizeValidatedQueryResultBatchForPersistence( + bat *batch.Batch, + mp *mpool.MPool, + needsNormalization bool, +) (*batch.Batch, func(), error) { + if !needsNormalization { + return bat, nil, nil + } + if mp == nil { + return nil, nil, moerr.NewInternalErrorNoCtx( + "cannot normalize query result batch without a memory pool") + } + + rows := bat.RowCount() + normalized := batch.NewWithSize(len(bat.Vecs)) + normalized.Attrs = bat.Attrs + normalized.SetRowCount(rows) + release := func() { + for i, vec := range normalized.Vecs { + if vec != nil && vec != bat.Vecs[i] { + vec.Free(mp) + } + normalized.Vecs[i] = nil + } + normalized.Vecs = nil + normalized.Attrs = nil + normalized.SetRowCount(0) + } + + for i, vec := range bat.Vecs { + if vec.Length() == rows { + normalized.Vecs[i] = vec + continue + } + dup, err := vec.Dup(mp) + if err != nil { + release() + return nil, nil, err + } + dup.SetLength(rows) + normalized.Vecs[i] = dup + } + return normalized, release, nil +} + func saveBatch(ctx context.Context, ses *Session, bat *batch.Batch) error { - n := uint64(0) - if bat != nil && bat.Vecs[0] != nil { - n = uint64(bat.Vecs[0].Length()) + needsNormalization, err := validateQueryResultBatchForPersistence(bat) + if err != nil { + return err } + n := uint64(bat.RowCount()) ses.queryRowCount += n s := ses.curResultSize + float64(bat.Size())/(1024*1024) @@ -121,6 +224,14 @@ func saveBatch(ctx context.Context, ses *Session, bat *batch.Batch) error { ses.Debug(ctx, "open save query result", zap.Float64("current result size:", s)) return nil } + persistedBat, release, err := normalizeValidatedQueryResultBatchForPersistence( + bat, ses.GetMemPool(), needsNormalization) + if err != nil { + return err + } + if release != nil { + defer release() + } fs := getPu(ses.GetService()).FileService // write query result path := catalog.BuildQueryResultPath(ses.GetTenantInfo().GetTenant(), uuid.UUID(ses.GetStmtId()).String(), ses.GetIncBlockIdx()) @@ -129,7 +240,7 @@ func saveBatch(ctx context.Context, ses *Session, bat *batch.Batch) error { if err != nil { return err } - _, err = writer.Write(bat) + _, err = writer.Write(persistedBat) if err != nil { return err } diff --git a/pkg/frontend/result_cardinality_test.go b/pkg/frontend/result_cardinality_test.go new file mode 100644 index 0000000000000..23ed7ec8e70db --- /dev/null +++ b/pkg/frontend/result_cardinality_test.go @@ -0,0 +1,529 @@ +// Copyright 2026 Matrix Origin +// +// 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 frontend + +import ( + "context" + "errors" + "testing" + + "github.com/golang/mock/gomock" + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + "github.com/matrixorigin/matrixone/pkg/catalog" + "github.com/matrixorigin/matrixone/pkg/common/mpool" + "github.com/matrixorigin/matrixone/pkg/config" + "github.com/matrixorigin/matrixone/pkg/container/batch" + "github.com/matrixorigin/matrixone/pkg/container/types" + "github.com/matrixorigin/matrixone/pkg/container/vector" + "github.com/matrixorigin/matrixone/pkg/defines" + "github.com/matrixorigin/matrixone/pkg/fileservice" + "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/sql/colexec/external" + "github.com/matrixorigin/matrixone/pkg/sql/parsers/tree" + "github.com/matrixorigin/matrixone/pkg/testutil" + "github.com/matrixorigin/matrixone/pkg/util/trace/impl/motrace/statistic" + "github.com/matrixorigin/matrixone/pkg/vm/process" +) + +func frontendResultBatch(mp *mpool.MPool, first *vector.Vector, rows int) *batch.Batch { + totals := make([]int64, rows) + for i := range totals { + totals[i] = int64((i + 1) * 10) + } + bat := batch.NewWithSize(2) + bat.Vecs[0] = first + bat.Vecs[1] = testutil.MakeInt64Vector(totals, nil, mp) + bat.SetRowCount(rows) + return bat +} + +func TestMysqlProtocolWriteUsesBatchLogicalRowCount(t *testing.T) { + sv, err := getSystemVariables("test/system_vars_config.toml") + require.NoError(t, err) + pu := config.NewParameterUnit(sv, nil, nil, nil) + setPu("", pu) + setSessionAlloc("", NewLeakCheckAllocator()) + + ioSession, err := NewIOSession(&testConn{}, pu, "") + require.NoError(t, err) + proto := NewMysqlClientProtocol("", 0, ioSession, 1024, sv) + t.Cleanup(proto.Close) + + ctx := context.Background() + ses := NewSession(ctx, "", proto, nil) + ses.SetCmd(COM_QUERY) + ses.mrs = &MysqlResultSet{} + for _, name := range []string{"projection_value", "total"} { + column := &MysqlColumn{} + column.SetName(name) + column.SetColumnType(defines.MYSQL_TYPE_LONGLONG) + ses.mrs.AddColumn(column) + } + proto.ses = ses + execCtx := &ExecCtx{reqCtx: ctx, ses: ses} + + mp := mpool.MustNewZero() + t.Cleanup(func() { require.Zero(t, mp.CurrNB()) }) + tests := []struct { + name string + rows int + first func() *vector.Vector + }{ + { + name: "flat control", + rows: 3, + first: func() *vector.Vector { + return testutil.MakeInt64Vector([]int64{7, 8, 9}, nil, mp) + }, + }, + { + name: "ordinary constant", + rows: 3, + first: func() *vector.Vector { + vec, newErr := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 3, mp) + require.NoError(t, newErr) + return vec + }, + }, + { + name: "broadcast constant", + rows: 3, + first: func() *vector.Vector { + vec, newErr := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 1, mp) + require.NoError(t, newErr) + return vec + }, + }, + { + name: "broadcast constant null", + rows: 2, + first: func() *vector.Vector { + return vector.NewConstNull(types.T_int64.ToType(), 1, mp) + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + bat := frontendResultBatch(mp, test.first(), test.rows) + defer bat.Clean(mp) + + before := proto.tcpConn.sequenceId + require.NoError(t, proto.Write(execCtx, nil, bat)) + require.Equal(t, uint8(test.rows), proto.tcpConn.sequenceId-before) + }) + } +} + +func TestGetDataFromPipelineUsesBatchLogicalRowCount(t *testing.T) { + ctx := statistic.ContextWithStatsInfo(context.Background(), &statistic.StatsInfo{}) + ses := &Session{feSessionImpl: feSessionImpl{respr: &NullResp{}}} + execCtx := &ExecCtx{reqCtx: ctx, ses: ses} + mp := mpool.MustNewZero() + t.Cleanup(func() { require.Zero(t, mp.CurrNB()) }) + + first, err := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 1, mp) + require.NoError(t, err) + bat := frontendResultBatch(mp, first, 3) + require.NoError(t, getDataFromPipeline(ses, execCtx, bat, nil)) + require.Equal(t, int64(3), ses.sentRows.Load()) + bat.Clean(mp) + + nullBatch := frontendResultBatch(mp, vector.NewConstNull(types.T_int64.ToType(), 1, mp), 2) + require.NoError(t, getDataFromPipeline(ses, execCtx, nullBatch, nil)) + require.Equal(t, int64(5), ses.sentRows.Load()) + nullBatch.Clean(mp) +} + +func TestSaveBatchUsesBatchLogicalRowCount(t *testing.T) { + ses := &Session{limitResultSize: -1} + mp := mpool.MustNewZero() + t.Cleanup(func() { require.Zero(t, mp.CurrNB()) }) + + first, err := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 1, mp) + require.NoError(t, err) + bat := frontendResultBatch(mp, first, 3) + require.NoError(t, saveBatch(context.Background(), ses, bat)) + require.Equal(t, uint64(3), ses.queryRowCount) + bat.Clean(mp) + + nullBatch := frontendResultBatch(mp, vector.NewConstNull(types.T_int64.ToType(), 1, mp), 2) + require.NoError(t, saveBatch(context.Background(), ses, nullBatch)) + require.Equal(t, uint64(5), ses.queryRowCount) + nullBatch.Clean(mp) +} + +func TestSaveBatchValidatesBeforeAccountingAndSizing(t *testing.T) { + tests := []struct { + name string + bat func() *batch.Batch + }{ + { + name: "nil batch", + bat: func() *batch.Batch { return nil }, + }, + { + name: "negative row count", + bat: func() *batch.Batch { + bat := batch.NewWithSize(1) + bat.Vecs[0] = vector.NewVec(types.T_int64.ToType()) + bat.SetRowCount(-1) + return bat + }, + }, + { + name: "zero vectors", + bat: func() *batch.Batch { + bat := batch.NewWithSize(0) + bat.SetRowCount(1) + return bat + }, + }, + { + name: "nil vector", + bat: func() *batch.Batch { + bat := batch.NewWithSize(1) + bat.SetRowCount(1) + return bat + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ses := &Session{limitResultSize: -1, queryRowCount: 41} + bat := test.bat() + err := saveBatch(context.Background(), ses, bat) + require.Error(t, err) + require.Equal(t, uint64(41), ses.queryRowCount) + }) + } +} + +func TestNormalizeQueryResultBatchForPersistence(t *testing.T) { + mp := mpool.MustNewZero() + t.Cleanup(func() { require.Zero(t, mp.CurrNB()) }) + + t.Run("aligned vectors are a no-op", func(t *testing.T) { + constant, err := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 3, mp) + require.NoError(t, err) + bat := batch.NewWithSize(2) + bat.Attrs = []string{"constant", "flat"} + bat.Vecs[0] = constant + bat.Vecs[1] = testutil.MakeInt64Vector([]int64{10, 20, 30}, nil, mp) + bat.SetRowCount(3) + defer bat.Clean(mp) + + normalized, release, err := normalizeQueryResultBatchForPersistence(bat, mp) + require.NoError(t, err) + require.Same(t, bat, normalized) + require.Nil(t, release) + require.Same(t, bat.Vecs[0], normalized.Vecs[0]) + require.Same(t, bat.Vecs[1], normalized.Vecs[1]) + }) + + t.Run("first and non-first constants are normalized", func(t *testing.T) { + firstConstant, err := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 1, mp) + require.NoError(t, err) + laterConstant, err := vector.NewConstFixed(types.T_int64.ToType(), int64(11), 1, mp) + require.NoError(t, err) + constantNull := vector.NewConstNull(types.T_int64.ToType(), 1, mp) + bat := batch.NewWithSize(4) + bat.Attrs = []string{"first_constant", "flat", "later_constant", "constant_null"} + bat.Vecs[0] = firstConstant + bat.Vecs[1] = testutil.MakeInt64Vector([]int64{10, 20, 30}, nil, mp) + bat.Vecs[2] = laterConstant + bat.Vecs[3] = constantNull + bat.SetRowCount(3) + defer bat.Clean(mp) + before := mp.CurrNB() + + normalized, release, err := normalizeQueryResultBatchForPersistence(bat, mp) + require.NoError(t, err) + require.NotSame(t, bat, normalized) + require.NotNil(t, release) + require.Equal(t, bat.Attrs, normalized.Attrs) + require.NotSame(t, bat.Vecs[0], normalized.Vecs[0]) + require.Same(t, bat.Vecs[1], normalized.Vecs[1]) + require.NotSame(t, bat.Vecs[2], normalized.Vecs[2]) + require.NotSame(t, bat.Vecs[3], normalized.Vecs[3]) + for _, vec := range normalized.Vecs { + require.Equal(t, 3, vec.Length()) + } + require.Equal(t, []int{1, 3, 1, 1}, []int{ + bat.Vecs[0].Length(), bat.Vecs[1].Length(), bat.Vecs[2].Length(), bat.Vecs[3].Length(), + }) + require.Equal(t, 3, bat.RowCount()) + require.Equal(t, int64(7), vector.GetFixedAtWithTypeCheck[int64](normalized.Vecs[0], 2)) + require.Equal(t, int64(11), vector.GetFixedAtWithTypeCheck[int64](normalized.Vecs[2], 2)) + require.True(t, normalized.Vecs[3].IsNull(2)) + + release() + require.Equal(t, before, mp.CurrNB()) + require.Equal(t, []int{1, 3, 1, 1}, []int{ + bat.Vecs[0].Length(), bat.Vecs[1].Length(), bat.Vecs[2].Length(), bat.Vecs[3].Length(), + }) + }) + + t.Run("zero rows shorten only the persistence view", func(t *testing.T) { + constant, err := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 1, mp) + require.NoError(t, err) + bat := batch.NewWithSize(2) + bat.Vecs[0] = testutil.MakeInt64Vector(nil, nil, mp) + bat.Vecs[1] = constant + bat.SetRowCount(0) + defer bat.Clean(mp) + before := mp.CurrNB() + + normalized, release, err := normalizeQueryResultBatchForPersistence(bat, mp) + require.NoError(t, err) + require.NotNil(t, release) + require.Same(t, bat.Vecs[0], normalized.Vecs[0]) + require.NotSame(t, bat.Vecs[1], normalized.Vecs[1]) + require.Zero(t, normalized.RowCount()) + require.Zero(t, normalized.Vecs[0].Length()) + require.Zero(t, normalized.Vecs[1].Length()) + require.Equal(t, 1, bat.Vecs[1].Length()) + + release() + require.Equal(t, before, mp.CurrNB()) + require.Equal(t, 1, bat.Vecs[1].Length()) + }) + + t.Run("invalid row shapes are rejected without mutation", func(t *testing.T) { + tests := []struct { + name string + vec func() *vector.Vector + }{ + { + name: "short flat", + vec: func() *vector.Vector { + return testutil.MakeInt64Vector([]int64{7}, nil, mp) + }, + }, + { + name: "empty constant", + vec: func() *vector.Vector { + return vector.NewConstNull(types.T_int64.ToType(), 0, mp) + }, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + vec := test.vec() + bat := batch.NewWithSize(1) + bat.Vecs[0] = vec + bat.SetRowCount(3) + defer bat.Clean(mp) + before := mp.CurrNB() + originalLength := vec.Length() + + normalized, release, err := normalizeQueryResultBatchForPersistence(bat, mp) + require.Error(t, err) + require.Nil(t, normalized) + require.Nil(t, release) + require.Equal(t, originalLength, vec.Length()) + require.Equal(t, 3, bat.RowCount()) + require.Equal(t, before, mp.CurrNB()) + }) + } + }) + + t.Run("duplicate failure releases prior replacements", func(t *testing.T) { + limited, err := mpool.NewMPool(t.Name(), 1<<20, mpool.NoFixed) + require.NoError(t, err) + defer mpool.DeleteMPool(limited) + + payload := make([]byte, 600<<10) + bat := batch.NewWithSize(2) + for i := range bat.Vecs { + vec := vector.NewOffHeapVecWithType(types.T_text.ToType()) + require.NoError(t, vector.AppendBytes(vec, payload, false, mp)) + vec.SetClass(vector.CONSTANT) + bat.Vecs[i] = vec + } + bat.SetRowCount(3) + defer bat.Clean(mp) + + normalized, release, err := normalizeQueryResultBatchForPersistence(bat, limited) + require.Error(t, err) + require.Nil(t, normalized) + require.Nil(t, release) + require.Zero(t, limited.CurrNB()) + require.Equal(t, 3, bat.RowCount()) + require.Equal(t, 1, bat.Vecs[0].Length()) + require.Equal(t, 1, bat.Vecs[1].Length()) + }) +} + +type queryResultFailingWriteFS struct { + fileservice.FileService +} + +func (fs *queryResultFailingWriteFS) Write(context.Context, fileservice.IOVector) error { + return errors.New("injected query result write failure") +} + +func TestSaveBatchReleasesNormalizedVectorsOnWriteError(t *testing.T) { + ctrl := gomock.NewController(t) + ses := newTestSession(t, ctrl) + defer ses.Close() + ses.limitResultSize = 64 + ses.SetStmtId(uuid.New()) + + pu := getPu(ses.GetService()) + originalFS := pu.FileService + pu.FileService = &queryResultFailingWriteFS{FileService: originalFS} + defer func() { pu.FileService = originalFS }() + + constant, err := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 1, ses.GetMemPool()) + require.NoError(t, err) + bat := frontendResultBatch(ses.GetMemPool(), constant, 3) + defer bat.Clean(ses.GetMemPool()) + before := ses.GetMemPool().CurrNB() + + err = saveBatch(context.Background(), ses, bat) + require.ErrorContains(t, err, "injected query result write failure") + require.Equal(t, before, ses.GetMemPool().CurrNB()) + require.Equal(t, 3, bat.RowCount()) + require.Equal(t, 1, bat.Vecs[0].Length()) + require.Equal(t, 3, bat.Vecs[1].Length()) +} + +func TestSaveBatchPersistsBatchLogicalCardinalityForResultScan(t *testing.T) { + tests := []struct { + name string + rows int + makeBatch func(*mpool.MPool) *batch.Batch + check func(*testing.T, *batch.Batch) + }{ + { + name: "broadcast constant first", + rows: 3, + makeBatch: func(mp *mpool.MPool) *batch.Batch { + constant, err := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 1, mp) + require.NoError(t, err) + return frontendResultBatch(mp, constant, 3) + }, + check: func(t *testing.T, bat *batch.Batch) { + require.Equal(t, []int64{7, 7, 7}, vector.MustFixedColWithTypeCheck[int64](bat.Vecs[0])) + require.Equal(t, []int64{10, 20, 30}, vector.MustFixedColWithTypeCheck[int64](bat.Vecs[1])) + }, + }, + { + name: "broadcast constant null first", + rows: 3, + makeBatch: func(mp *mpool.MPool) *batch.Batch { + return frontendResultBatch(mp, vector.NewConstNull(types.T_int64.ToType(), 1, mp), 3) + }, + check: func(t *testing.T, bat *batch.Batch) { + for row := 0; row < 3; row++ { + require.True(t, bat.Vecs[0].IsNull(uint64(row))) + } + require.Equal(t, []int64{10, 20, 30}, vector.MustFixedColWithTypeCheck[int64](bat.Vecs[1])) + }, + }, + { + name: "broadcast constant after flat first", + rows: 3, + makeBatch: func(mp *mpool.MPool) *batch.Batch { + constant, err := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 1, mp) + require.NoError(t, err) + bat := batch.NewWithSize(2) + bat.Vecs[0] = testutil.MakeInt64Vector([]int64{10, 20, 30}, nil, mp) + bat.Vecs[1] = constant + bat.SetRowCount(3) + return bat + }, + check: func(t *testing.T, bat *batch.Batch) { + require.Equal(t, []int64{10, 20, 30}, vector.MustFixedColWithTypeCheck[int64](bat.Vecs[0])) + require.Equal(t, []int64{7, 7, 7}, vector.MustFixedColWithTypeCheck[int64](bat.Vecs[1])) + }, + }, + { + name: "zero logical rows", + rows: 0, + makeBatch: func(mp *mpool.MPool) *batch.Batch { + constant, err := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 1, mp) + require.NoError(t, err) + return frontendResultBatch(mp, constant, 0) + }, + check: func(t *testing.T, bat *batch.Batch) { + require.Zero(t, bat.Vecs[0].Length()) + require.Zero(t, bat.Vecs[1].Length()) + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + ses := newTestSession(t, ctrl) + defer ses.Close() + ses.limitResultSize = 64 + stmtID := uuid.New() + ses.SetStmtId(stmtID) + + input := test.makeBatch(ses.GetMemPool()) + defer input.Clean(ses.GetMemPool()) + require.NoError(t, saveBatch(context.Background(), ses, input)) + + path := catalog.BuildQueryResultPath( + ses.GetTenantInfo().GetTenant(), stmtID.String(), 1) + param := &external.ExternalParam{ + ExParamConst: external.ExParamConst{ + Attrs: []plan.ExternAttr{ + {ColName: "c0", ColIndex: 0}, + {ColName: "c1", ColIndex: 1}, + }, + Cols: []*plan.ColDef{ + {Typ: plan.Type{Id: int32(types.T_int64)}}, + {Typ: plan.Type{Id: int32(types.T_int64)}}, + }, + Extern: &tree.ExternParam{ExParam: tree.ExParam{ + FileService: getPu(ses.GetService()).FileService, + }}, + }, + ExParam: external.ExParam{ + Fileparam: &external.ExFileparam{Filepath: path}, + Filter: &external.FilterParam{}, + }, + } + proc := testutil.NewProcess(t) + defer proc.Free() + reader := external.NewZonemapReader(param, proc) + _, err := reader.Open(param, proc) + require.NoError(t, err) + + output := batch.NewWithSize(2) + output.Vecs[0] = vector.NewVec(types.T_int64.ToType()) + output.Vecs[1] = vector.NewVec(types.T_int64.ToType()) + defer output.Clean(proc.Mp()) + finished, err := reader.ReadBatch( + context.Background(), output, proc, + process.NewAnalyzer(0, false, false, "result scan"), + ) + require.NoError(t, err) + require.True(t, finished) + require.NoError(t, reader.Close()) + require.Equal(t, test.rows, output.RowCount()) + require.Equal(t, test.rows, output.Vecs[0].Length()) + require.Equal(t, test.rows, output.Vecs[1].Length()) + test.check(t, output) + }) + } +} diff --git a/test/distributed/cases/prepare/prepared_projection_cardinality.result b/test/distributed/cases/prepare/prepared_projection_cardinality.result new file mode 100644 index 0000000000000..5fc018559d47a --- /dev/null +++ b/test/distributed/cases/prepare/prepared_projection_cardinality.result @@ -0,0 +1,60 @@ +DROP DATABASE IF EXISTS prepared_projection_cardinality; +CREATE DATABASE prepared_projection_cardinality; +USE prepared_projection_cardinality; +CREATE TABLE metric_rows(bucket VARCHAR(20), value_col INT); +INSERT INTO metric_rows VALUES ('a', 10), ('b', 20), ('c', 30); +SET save_query_result = ON; +PREPARE parameter_first FROM +'SELECT ? AS projection_value, SUM(value_col) AS total FROM metric_rows GROUP BY bucket ORDER BY total'; +SET @projection_value = 7; +EXECUTE parameter_first USING @projection_value; +➤ projection_value[12,0,0] ¦ total[-5,64,0] 𝄀 +7 ¦ 10 𝄀 +7 ¦ 20 𝄀 +7 ¦ 30 +SELECT * FROM result_scan(last_query_id()) AS saved_parameter_first; +➤ projection_value[12,0,0] ¦ total[-5,64,0] 𝄀 +7 ¦ 10 𝄀 +7 ¦ 20 𝄀 +7 ¦ 30 +SET @projection_value = NULL; +EXECUTE parameter_first USING @projection_value; +➤ projection_value[12,0,0] ¦ total[-5,64,0] 𝄀 +null ¦ 10 𝄀 +null ¦ 20 𝄀 +null ¦ 30 +SELECT * FROM result_scan(last_query_id()) AS saved_parameter_null; +➤ projection_value[12,0,0] ¦ total[-5,64,0] 𝄀 +null ¦ 10 𝄀 +null ¦ 20 𝄀 +null ¦ 30 +DEALLOCATE PREPARE parameter_first; +PREPARE parameter_second FROM +'SELECT bucket, ? AS projection_value, SUM(value_col) AS total FROM metric_rows GROUP BY bucket ORDER BY total'; +SET @projection_value = 11; +EXECUTE parameter_second USING @projection_value; +➤ bucket[12,20,0] ¦ projection_value[12,0,0] ¦ total[-5,64,0] 𝄀 +a ¦ 11 ¦ 10 𝄀 +b ¦ 11 ¦ 20 𝄀 +c ¦ 11 ¦ 30 +SELECT * FROM result_scan(last_query_id()) AS saved_parameter_second; +➤ bucket[12,20,0] ¦ projection_value[12,0,0] ¦ total[-5,64,0] 𝄀 +a ¦ 11 ¦ 10 𝄀 +b ¦ 11 ¦ 20 𝄀 +c ¦ 11 ¦ 30 +DEALLOCATE PREPARE parameter_second; +SET save_query_result = OFF; +SELECT 7 AS projection_value, SUM(value_col) AS total +FROM metric_rows GROUP BY bucket ORDER BY total; +➤ projection_value[-5,64,0] ¦ total[-5,64,0] 𝄀 +7 ¦ 10 𝄀 +7 ¦ 20 𝄀 +7 ¦ 30 +PREPARE parameter_group FROM +'SELECT ? AS projection_value, SUM(value_col) AS total FROM metric_rows GROUP BY ?'; +SET @projection_value = 13, @group_value = 17; +EXECUTE parameter_group USING @projection_value, @group_value; +➤ projection_value[12,0,0] ¦ total[-5,64,0] 𝄀 +13 ¦ 60 +DEALLOCATE PREPARE parameter_group; +DROP DATABASE prepared_projection_cardinality; diff --git a/test/distributed/cases/prepare/prepared_projection_cardinality.sql b/test/distributed/cases/prepare/prepared_projection_cardinality.sql new file mode 100644 index 0000000000000..b859c3586c522 --- /dev/null +++ b/test/distributed/cases/prepare/prepared_projection_cardinality.sql @@ -0,0 +1,49 @@ +-- @case +-- @desc:Prepared constant projections preserve grouped result cardinality +-- @label:bvt +-- @metacmp(false) + +DROP DATABASE IF EXISTS prepared_projection_cardinality; +CREATE DATABASE prepared_projection_cardinality; +USE prepared_projection_cardinality; + +CREATE TABLE metric_rows(bucket VARCHAR(20), value_col INT); +INSERT INTO metric_rows VALUES ('a', 10), ('b', 20), ('c', 30); +SET save_query_result = ON; + +-- A parameter-backed constant is physically one value, but represents every +-- logical row in the grouped output batch. +PREPARE parameter_first FROM + 'SELECT ? AS projection_value, SUM(value_col) AS total FROM metric_rows GROUP BY bucket ORDER BY total'; +SET @projection_value = 7; +EXECUTE parameter_first USING @projection_value; +SELECT * FROM result_scan(last_query_id()) AS saved_parameter_first; + +-- Reuse the same statement and cover const-null broadcast semantics. +SET @projection_value = NULL; +EXECUTE parameter_first USING @projection_value; +SELECT * FROM result_scan(last_query_id()) AS saved_parameter_null; +DEALLOCATE PREPARE parameter_first; + +-- Moving the same parameter away from the first column must not affect row +-- cardinality. +PREPARE parameter_second FROM + 'SELECT bucket, ? AS projection_value, SUM(value_col) AS total FROM metric_rows GROUP BY bucket ORDER BY total'; +SET @projection_value = 11; +EXECUTE parameter_second USING @projection_value; +SELECT * FROM result_scan(last_query_id()) AS saved_parameter_second; +DEALLOCATE PREPARE parameter_second; +SET save_query_result = OFF; + +-- Nearest non-prepared representation control. +SELECT 7 AS projection_value, SUM(value_col) AS total +FROM metric_rows GROUP BY bucket ORDER BY total; + +-- Preserve the broadcast hashmap path fixed by #26843. +PREPARE parameter_group FROM + 'SELECT ? AS projection_value, SUM(value_col) AS total FROM metric_rows GROUP BY ?'; +SET @projection_value = 13, @group_value = 17; +EXECUTE parameter_group USING @projection_value, @group_value; +DEALLOCATE PREPARE parameter_group; + +DROP DATABASE prepared_projection_cardinality; From 0bfa799584b300ea31b4402389ebc148149c706d Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Sun, 9 Aug 2026 21:20:08 +0800 Subject: [PATCH 2/2] fix(executor): preserve logical result cardinality --- pkg/frontend/query_result.go | 8 +- pkg/frontend/result_cardinality_test.go | 114 +++++++---- pkg/sql/colexec/apply/apply.go | 10 + pkg/sql/colexec/apply/apply_test.go | 191 ++++++++++++++++++ .../cases/function/func_misc_inet.result | 1 + .../cases/function/func_misc_inet6.result | Bin 4224 -> 4259 bytes 6 files changed, 280 insertions(+), 44 deletions(-) diff --git a/pkg/frontend/query_result.go b/pkg/frontend/query_result.go index cffb651df7b92..924768e00c16d 100644 --- a/pkg/frontend/query_result.go +++ b/pkg/frontend/query_result.go @@ -137,10 +137,10 @@ func validateQueryResultBatchForPersistence(bat *batch.Batch) (bool, error) { if vec.Length() == rows { continue } - // A non-empty const physically owns one value and can be broadcast to - // any requested logical row range. Flat vectors and empty constants do - // not have storage for a mismatched positive row range. - if !vec.IsConst() || (rows > 0 && vec.Length() == 0) { + // A non-empty const physically owns one value, while an empty const is + // the payload-free representation of const null. Both can be broadcast + // to the batch's logical row count; mismatched flat vectors cannot. + if !vec.IsConst() || (vec.Length() == 0 && !vec.IsConstNull()) { return false, moerr.NewInternalErrorNoCtxf( "invalid query result batch: vector %d length %d does not match row count %d", i, vec.Length(), rows) diff --git a/pkg/frontend/result_cardinality_test.go b/pkg/frontend/result_cardinality_test.go index 23ed7ec8e70db..393977571219c 100644 --- a/pkg/frontend/result_cardinality_test.go +++ b/pkg/frontend/result_cardinality_test.go @@ -58,7 +58,8 @@ func TestMysqlProtocolWriteUsesBatchLogicalRowCount(t *testing.T) { setPu("", pu) setSessionAlloc("", NewLeakCheckAllocator()) - ioSession, err := NewIOSession(&testConn{}, pu, "") + conn := &testConn{} + ioSession, err := NewIOSession(conn, pu, "") require.NoError(t, err) proto := NewMysqlClientProtocol("", 0, ioSession, 1024, sv) t.Cleanup(proto.Close) @@ -82,6 +83,7 @@ func TestMysqlProtocolWriteUsesBatchLogicalRowCount(t *testing.T) { name string rows int first func() *vector.Vector + null bool }{ { name: "flat control", @@ -114,6 +116,15 @@ func TestMysqlProtocolWriteUsesBatchLogicalRowCount(t *testing.T) { first: func() *vector.Vector { return vector.NewConstNull(types.T_int64.ToType(), 1, mp) }, + null: true, + }, + { + name: "empty broadcast constant null", + rows: 1, + first: func() *vector.Vector { + return vector.NewConstNull(types.T_int64.ToType(), 0, mp) + }, + null: true, }, } @@ -123,8 +134,17 @@ func TestMysqlProtocolWriteUsesBatchLogicalRowCount(t *testing.T) { defer bat.Clean(mp) before := proto.tcpConn.sequenceId + wireStart := len(conn.data) require.NoError(t, proto.Write(execCtx, nil, bat)) require.Equal(t, uint8(test.rows), proto.tcpConn.sequenceId-before) + require.NoError(t, proto.tcpConn.Flush()) + if test.null && test.rows == 1 { + packet := conn.data[wireStart:] + require.GreaterOrEqual(t, len(packet), 5) + payloadLength := int(packet[0]) | int(packet[1])<<8 | int(packet[2])<<16 + require.Equal(t, payloadLength+4, len(packet)) + require.Equal(t, byte(0xfb), packet[4]) + } }) } } @@ -143,7 +163,7 @@ func TestGetDataFromPipelineUsesBatchLogicalRowCount(t *testing.T) { require.Equal(t, int64(3), ses.sentRows.Load()) bat.Clean(mp) - nullBatch := frontendResultBatch(mp, vector.NewConstNull(types.T_int64.ToType(), 1, mp), 2) + nullBatch := frontendResultBatch(mp, vector.NewConstNull(types.T_int64.ToType(), 0, mp), 2) require.NoError(t, getDataFromPipeline(ses, execCtx, nullBatch, nil)) require.Equal(t, int64(5), ses.sentRows.Load()) nullBatch.Clean(mp) @@ -161,7 +181,7 @@ func TestSaveBatchUsesBatchLogicalRowCount(t *testing.T) { require.Equal(t, uint64(3), ses.queryRowCount) bat.Clean(mp) - nullBatch := frontendResultBatch(mp, vector.NewConstNull(types.T_int64.ToType(), 1, mp), 2) + nullBatch := frontendResultBatch(mp, vector.NewConstNull(types.T_int64.ToType(), 0, mp), 2) require.NoError(t, saveBatch(context.Background(), ses, nullBatch)) require.Equal(t, uint64(5), ses.queryRowCount) nullBatch.Clean(mp) @@ -279,6 +299,31 @@ func TestNormalizeQueryResultBatchForPersistence(t *testing.T) { }) }) + t.Run("empty constant null is normalized on an owned duplicate", func(t *testing.T) { + constantNull := vector.NewConstNull(types.T_int64.ToType(), 0, mp) + bat := batch.NewWithSize(2) + bat.Vecs[0] = constantNull + bat.Vecs[1] = testutil.MakeInt64Vector([]int64{10}, nil, mp) + bat.SetRowCount(1) + defer bat.Clean(mp) + before := mp.CurrNB() + + normalized, release, err := normalizeQueryResultBatchForPersistence(bat, mp) + require.NoError(t, err) + require.NotSame(t, bat, normalized) + require.NotNil(t, release) + require.NotSame(t, bat.Vecs[0], normalized.Vecs[0]) + require.Same(t, bat.Vecs[1], normalized.Vecs[1]) + require.Equal(t, 1, normalized.RowCount()) + require.Equal(t, 1, normalized.Vecs[0].Length()) + require.True(t, normalized.Vecs[0].IsNull(0)) + require.Zero(t, bat.Vecs[0].Length()) + + release() + require.Equal(t, before, mp.CurrNB()) + require.Zero(t, bat.Vecs[0].Length()) + }) + t.Run("zero rows shorten only the persistence view", func(t *testing.T) { constant, err := vector.NewConstFixed(types.T_int64.ToType(), int64(7), 1, mp) require.NoError(t, err) @@ -304,43 +349,21 @@ func TestNormalizeQueryResultBatchForPersistence(t *testing.T) { require.Equal(t, 1, bat.Vecs[1].Length()) }) - t.Run("invalid row shapes are rejected without mutation", func(t *testing.T) { - tests := []struct { - name string - vec func() *vector.Vector - }{ - { - name: "short flat", - vec: func() *vector.Vector { - return testutil.MakeInt64Vector([]int64{7}, nil, mp) - }, - }, - { - name: "empty constant", - vec: func() *vector.Vector { - return vector.NewConstNull(types.T_int64.ToType(), 0, mp) - }, - }, - } - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - vec := test.vec() - bat := batch.NewWithSize(1) - bat.Vecs[0] = vec - bat.SetRowCount(3) - defer bat.Clean(mp) - before := mp.CurrNB() - originalLength := vec.Length() - - normalized, release, err := normalizeQueryResultBatchForPersistence(bat, mp) - require.Error(t, err) - require.Nil(t, normalized) - require.Nil(t, release) - require.Equal(t, originalLength, vec.Length()) - require.Equal(t, 3, bat.RowCount()) - require.Equal(t, before, mp.CurrNB()) - }) - } + t.Run("short flat vector is rejected without mutation", func(t *testing.T) { + vec := testutil.MakeInt64Vector([]int64{7}, nil, mp) + bat := batch.NewWithSize(1) + bat.Vecs[0] = vec + bat.SetRowCount(3) + defer bat.Clean(mp) + before := mp.CurrNB() + + normalized, release, err := normalizeQueryResultBatchForPersistence(bat, mp) + require.Error(t, err) + require.Nil(t, normalized) + require.Nil(t, release) + require.Equal(t, 1, vec.Length()) + require.Equal(t, 3, bat.RowCount()) + require.Equal(t, before, mp.CurrNB()) }) t.Run("duplicate failure releases prior replacements", func(t *testing.T) { @@ -437,6 +460,17 @@ func TestSaveBatchPersistsBatchLogicalCardinalityForResultScan(t *testing.T) { require.Equal(t, []int64{10, 20, 30}, vector.MustFixedColWithTypeCheck[int64](bat.Vecs[1])) }, }, + { + name: "empty broadcast constant null first", + rows: 1, + makeBatch: func(mp *mpool.MPool) *batch.Batch { + return frontendResultBatch(mp, vector.NewConstNull(types.T_int64.ToType(), 0, mp), 1) + }, + check: func(t *testing.T, bat *batch.Batch) { + require.True(t, bat.Vecs[0].IsNull(0)) + require.Equal(t, []int64{10}, vector.MustFixedColWithTypeCheck[int64](bat.Vecs[1])) + }, + }, { name: "broadcast constant after flat first", rows: 3, diff --git a/pkg/sql/colexec/apply/apply.go b/pkg/sql/colexec/apply/apply.go index bfdc89de71976..4c3f34f0c849a 100644 --- a/pkg/sql/colexec/apply/apply.go +++ b/pkg/sql/colexec/apply/apply.go @@ -161,6 +161,16 @@ func (ctr *container) probe(ap *Apply, proc *process.Process, result *vm.CallRes return err } } + ctr.rbat.AddRowCount(1) + if ctr.rbat.RowCount() >= colexec.DefaultBatchSize { + ctr.batIdx = i + 1 + ctr.tfFinish = true + if ctr.batIdx >= count { + ctr.inbat = nil + } + result.Batch = ctr.rbat + return nil + } } ctr.tfFinish = true break diff --git a/pkg/sql/colexec/apply/apply_test.go b/pkg/sql/colexec/apply/apply_test.go index 4b3f7b268ccaf..c394ccd015716 100644 --- a/pkg/sql/colexec/apply/apply_test.go +++ b/pkg/sql/colexec/apply/apply_test.go @@ -19,7 +19,15 @@ import ( "testing" "github.com/matrixorigin/matrixone/pkg/common/moerr" + "github.com/matrixorigin/matrixone/pkg/container/batch" + "github.com/matrixorigin/matrixone/pkg/container/types" + "github.com/matrixorigin/matrixone/pkg/container/vector" + "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/sql/colexec" + "github.com/matrixorigin/matrixone/pkg/sql/colexec/table_function" + plan2 "github.com/matrixorigin/matrixone/pkg/sql/plan" "github.com/matrixorigin/matrixone/pkg/testutil" + "github.com/matrixorigin/matrixone/pkg/vm/process" "github.com/stretchr/testify/require" ) @@ -66,3 +74,186 @@ func TestNilTableFunctionLifecycle(t *testing.T) { arg.Free(proc, false, nil) }) } + +func TestApplyMaintainsOuterNullCardinality(t *testing.T) { + tests := []struct { + name string + applyType int + wantIDs []int32 + wantResults []int64 + wantNulls int + }{ + { + name: "outer preserves unmatched row", + applyType: OUTER, + wantIDs: []int32{1, 1, 1, 2}, + wantResults: []int64{1, 2, 3, 0}, + wantNulls: 1, + }, + { + name: "cross omits unmatched row", + applyType: CROSS, + wantIDs: []int32{1, 1, 1}, + wantResults: []int64{1, 2, 3}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + proc := testutil.NewProc(t) + t.Cleanup(func() { + proc.Free() + require.Zero(t, proc.Mp().CurrNB()) + }) + arg := newGenerateSeriesApply( + t, + proc, + test.applyType, + []int32{1, 2}, + []int32{1, 1}, + []int32{3, -1}, + ) + + result, err := arg.Call(proc) + require.NoError(t, err) + require.Equal(t, len(test.wantIDs), result.Batch.RowCount()) + require.Equal(t, test.wantIDs, vector.MustFixedColWithTypeCheck[int32](result.Batch.Vecs[0])) + require.Equal(t, test.wantResults, vector.MustFixedColWithTypeCheck[int64](result.Batch.Vecs[1])) + require.Equal(t, test.wantNulls, result.Batch.Vecs[1].GetNulls().Count()) + }) + } +} + +func TestOuterApplyResumesAfterNullExtendedBatchBoundary(t *testing.T) { + proc := testutil.NewProc(t) + t.Cleanup(func() { + proc.Free() + require.Zero(t, proc.Mp().CurrNB()) + }) + + inputRows := colexec.DefaultBatchSize + 1 + ids := make([]int32, inputRows) + starts := make([]int32, inputRows) + ends := make([]int32, inputRows) + for i := range inputRows { + ids[i] = int32(i) + starts[i] = 1 + ends[i] = -1 + } + starts[inputRows-1] = 7 + ends[inputRows-1] = 7 + arg := newGenerateSeriesApply(t, proc, OUTER, ids, starts, ends) + + first, err := arg.Call(proc) + require.NoError(t, err) + require.Equal(t, colexec.DefaultBatchSize, first.Batch.RowCount()) + firstIDs := vector.MustFixedColWithTypeCheck[int32](first.Batch.Vecs[0]) + require.Len(t, firstIDs, colexec.DefaultBatchSize) + require.Equal(t, int32(0), firstIDs[0]) + require.Equal(t, int32(colexec.DefaultBatchSize-1), firstIDs[colexec.DefaultBatchSize-1]) + require.Equal(t, colexec.DefaultBatchSize, first.Batch.Vecs[1].GetNulls().Count()) + + second, err := arg.Call(proc) + require.NoError(t, err) + require.Equal(t, 1, second.Batch.RowCount()) + require.Equal(t, []int32{int32(colexec.DefaultBatchSize)}, vector.MustFixedColWithTypeCheck[int32](second.Batch.Vecs[0])) + require.Equal(t, []int64{7}, vector.MustFixedColWithTypeCheck[int64](second.Batch.Vecs[1])) + require.Zero(t, second.Batch.Vecs[1].GetNulls().Count()) + + done, err := arg.Call(proc) + require.NoError(t, err) + require.Nil(t, done.Batch) +} + +func TestOuterApplyNullExtendsNullUnnestInput(t *testing.T) { + proc := testutil.NewProc(t) + t.Cleanup(func() { + proc.Free() + require.Zero(t, proc.Mp().CurrNB()) + }) + input := batch.NewWithSize(2) + input.Vecs[0] = testutil.MakeInt32Vector([]int32{1, 2}, nil, proc.Mp()) + input.Vecs[1] = testutil.MakeJsonVector([]string{`{"a":1}`, "null"}, []uint64{1}, proc.Mp()) + input.SetRowCount(2) + + tf := table_function.NewArgument() + tf.FuncName = "unnest" + tf.Attrs = []string{"value"} + tf.Rets = []*plan.ColDef{{ + Name: "value", + Typ: plan.Type{Id: int32(types.T_varchar), Width: types.MaxVarcharLen}, + }} + tf.Args = []*plan.Expr{makeColumnExpr(1, types.T_json)} + + arg := NewArgument() + arg.ApplyType = OUTER + arg.Result = []colexec.ResultPos{{Rel: 0, Pos: 0}, {Rel: 1, Pos: 0}} + arg.Typs = []types.Type{types.New(types.T_varchar, types.MaxVarcharLen, 0)} + arg.TableFunction = tf + child := colexec.NewMockOperator().WithBatchs([]*batch.Batch{input}) + arg.AppendChild(child) + t.Cleanup(func() { + arg.Free(proc, false, nil) + child.Free(proc, false, nil) + arg.Release() + }) + + require.NoError(t, arg.Prepare(proc)) + result, err := arg.Call(proc) + require.NoError(t, err) + require.Equal(t, 2, result.Batch.RowCount()) + require.Equal(t, []int32{1, 2}, vector.MustFixedColWithTypeCheck[int32](result.Batch.Vecs[0])) + require.False(t, result.Batch.Vecs[1].IsNull(0)) + require.True(t, result.Batch.Vecs[1].IsNull(1)) +} + +func newGenerateSeriesApply( + t *testing.T, + proc *process.Process, + applyType int, + ids, starts, ends []int32, +) *Apply { + require.Len(t, starts, len(ids)) + require.Len(t, ends, len(ids)) + input := batch.NewWithSize(3) + input.Vecs[0] = testutil.MakeInt32Vector(ids, nil, proc.Mp()) + input.Vecs[1] = testutil.MakeInt32Vector(starts, nil, proc.Mp()) + input.Vecs[2] = testutil.MakeInt32Vector(ends, nil, proc.Mp()) + input.SetRowCount(len(ids)) + + tf := table_function.NewArgument() + tf.FuncName = "generate_series" + tf.Attrs = []string{"result"} + tf.Rets = []*plan.ColDef{{ + Name: "result", + Typ: plan.Type{Id: int32(types.T_int64)}, + }} + tf.Args = []*plan.Expr{ + makeColumnExpr(1, types.T_int32), + makeColumnExpr(2, types.T_int32), + plan2.MakePlan2Int32ConstExprWithType(1), + } + + arg := NewArgument() + arg.ApplyType = applyType + arg.Result = []colexec.ResultPos{{Rel: 0, Pos: 0}, {Rel: 1, Pos: 0}} + arg.Typs = []types.Type{types.T_int64.ToType()} + arg.TableFunction = tf + child := colexec.NewMockOperator().WithBatchs([]*batch.Batch{input}) + arg.AppendChild(child) + t.Cleanup(func() { + arg.Free(proc, false, nil) + child.Free(proc, false, nil) + arg.Release() + }) + + require.NoError(t, arg.Prepare(proc)) + return arg +} + +func makeColumnExpr(pos int32, typ types.T) *plan.Expr { + return &plan.Expr{ + Typ: plan.Type{Id: int32(typ)}, + Expr: &plan.Expr_Col{Col: &plan.ColRef{ColPos: pos}}, + } +} diff --git a/test/distributed/cases/function/func_misc_inet.result b/test/distributed/cases/function/func_misc_inet.result index 16eeea7b9867c..b84a41b2da114 100644 --- a/test/distributed/cases/function/func_misc_inet.result +++ b/test/distributed/cases/function/func_misc_inet.result @@ -51,6 +51,7 @@ zero_from_int 0.0.0.0 SELECT INET_NTOA(NULL) AS null_input; null_input +null SELECT INET_NTOA(167773449) AS uint64_test; uint64_test 10.0.5.9 diff --git a/test/distributed/cases/function/func_misc_inet6.result b/test/distributed/cases/function/func_misc_inet6.result index 681f938b758f833405bd22b4d531c0af65335ef9..01139d1287983bb30c128ae11eaca113ac589ff9 100644 GIT binary patch delta 100 zcmZorT&%bujF}~`G$&_rD6TG U$rspEQTY7q&S3uLKz4mj0CX%L7ytkO delta 71 zcmZ3i*r2!}jCt}xX5q