From 50847b7feb9a50545a4cb6a4056f61ad954526e4 Mon Sep 17 00:00:00 2001 From: bRong Njam Date: Fri, 9 Oct 2026 15:50:07 +0800 Subject: [PATCH 1/2] fix(frontend): publish prepared result metadata from executed schema --- pkg/frontend/computation_wrapper.go | 3 +- pkg/frontend/mysql_cmd_executor.go | 22 +- pkg/frontend/mysql_cmd_executor_test.go | 17 +- pkg/frontend/resp_client.go | 10 +- pkg/frontend/result_row_stmt.go | 76 ++++-- pkg/frontend/routine_test.go | 15 +- pkg/frontend/select_result_metadata_test.go | 249 +++++++++++++++++ pkg/frontend/session.go | 5 + pkg/frontend/status_stmt.go | 2 +- pkg/frontend/types.go | 3 + pkg/sql/compile/compile.go | 6 +- pkg/sql/compile/compile2.go | 2 +- pkg/sql/compile/types.go | 2 +- .../prepared_schema_commit_test.go | 252 ++++++++++++++++++ 14 files changed, 619 insertions(+), 45 deletions(-) create mode 100644 pkg/frontend/select_result_metadata_test.go create mode 100644 pkg/tests/sqlintegration/prepared_schema_commit_test.go diff --git a/pkg/frontend/computation_wrapper.go b/pkg/frontend/computation_wrapper.go index cb188c7a58fe4..5872c0ba39e6d 100644 --- a/pkg/frontend/computation_wrapper.go +++ b/pkg/frontend/computation_wrapper.go @@ -814,7 +814,8 @@ func (cwft *TxnComputationWrapper) completeCompileExecution( } func (cwft *TxnComputationWrapper) syncCompileExecution(runningCompile Compile) { - // Sync the latest plan generation after Run (it may have changed on retry). + // Sync the latest plan generation before publishing metadata or after Run + // (it may have changed on retry). cwft.plan = runningCompile.GetPlan() cwft.planSnapshotTS, cwft.hasPlanSnapshotTS = runningCompile.PlanSnapshotTS() } diff --git a/pkg/frontend/mysql_cmd_executor.go b/pkg/frontend/mysql_cmd_executor.go index 458dc77f987d0..1488e7fb2e85c 100644 --- a/pkg/frontend/mysql_cmd_executor.go +++ b/pkg/frontend/mysql_cmd_executor.go @@ -5668,11 +5668,10 @@ func rebuildStaleCachedStatements(ses FeSession, execCtx *ExecCtx) (err error) { return nil } -func dispatchStmt(ses FeSession, - statsArr *statistic.StatsArray, - execCtx *ExecCtx) (err error) { - ses.EnterFPrint(FPDispatchStmt) - defer ses.ExitFPrint(FPDispatchStmt) +// checkCachedStatementPlan admits or replaces a cached generation before +// physical compilation. Keeping this phase separate from Run lets compile-only +// consumers validate the same generation boundary without publishing results. +func checkCachedStatementPlan(ses FeSession, execCtx *ExecCtx) error { ses.GetTxnCompileCtx().tcw = execCtx.cw //5. check plan within txn if !execCtx.input.isBinaryProtExecute && execCtx.cw.Plan() != nil { @@ -5687,11 +5686,22 @@ func dispatchStmt(ses FeSession, } } if flag { - if err = rebuildStaleCachedStatements(ses, execCtx); err != nil { + if err := rebuildStaleCachedStatements(ses, execCtx); err != nil { return err } } } + return nil +} + +func dispatchStmt(ses FeSession, + statsArr *statistic.StatsArray, + execCtx *ExecCtx) (err error) { + ses.EnterFPrint(FPDispatchStmt) + defer ses.ExitFPrint(FPDispatchStmt) + if err = checkCachedStatementPlan(ses, execCtx); err != nil { + return err + } //6. execute stmt within txn switch sesImpl := ses.(type) { diff --git a/pkg/frontend/mysql_cmd_executor_test.go b/pkg/frontend/mysql_cmd_executor_test.go index 8029080070e4f..86d3ef9bb31f7 100644 --- a/pkg/frontend/mysql_cmd_executor_test.go +++ b/pkg/frontend/mysql_cmd_executor_test.go @@ -11693,13 +11693,6 @@ func TestPreparedGroupConcatFloorCapturedWithoutPhysicalCompile(t *testing.T) { }) } -type statsAdmissionStopResponse struct { - Responser - err error -} - -func (r *statsAdmissionStopResponse) RespPreMeta(*ExecCtx, any) error { return r.err } - func TestOrdinaryCacheStatsAdmissionUsesGenerationBaseline(t *testing.T) { for _, tc := range []struct { name, sql string @@ -11721,8 +11714,6 @@ func TestOrdinaryCacheStatsAdmissionUsesGenerationBaseline(t *testing.T) { ses.SetDatabaseName("tpch") ses.GetTxnCompileCtx().SetDatabase("tpch") defer ses.GetTxnCompileCtx().Close() - stop := fmt.Errorf("stop after compile before result execution") - ec.resper = &statsAdmissionStopResponse{Responser: ses.GetResponser(), err: stop} _, table, err := base.Resolve("tpch", "nation", nil) require.NoError(t, err) installStatsAdmissionStorage(t, ses, table, func() *pbstats.StatsInfo { return ctx.stats }) @@ -11751,8 +11742,12 @@ func TestOrdinaryCacheStatsAdmissionUsesGenerationBaseline(t *testing.T) { require.True(t, cw.planGenerationReused) ec.cw, ec.cws, ec.stmt = cw, cws, cw.GetAst() ctx.stats.TableCnt = tc.rows - err = dispatchStmt(ses, statistic.NewStatsArray(), ec) - require.ErrorIs(t, err, stop) + // Validate the real admission phase before Compile. The storage mock + // cannot execute a pipeline, and metadata is published only during Run. + require.NoError(t, checkCachedStatementPlan(ses, ec)) + compiled, err := cw.Compile(ec, ses.GetOutputCallback(ec)) + require.NoError(t, err) + defer compiled.(*compile.Compile).Release() if tc.rebuild { require.NotSame(t, cached, cw.Plan()) require.False(t, cw.planGenerationReused) diff --git a/pkg/frontend/resp_client.go b/pkg/frontend/resp_client.go index af592afda0514..220747513e8d9 100644 --- a/pkg/frontend/resp_client.go +++ b/pkg/frontend/resp_client.go @@ -232,10 +232,10 @@ func (resper *MysqlResp) RespPreMeta(execCtx *ExecCtx, meta any) (err error) { } func (resper *MysqlResp) RespResult(execCtx *ExecCtx, crs *perfcounter.CounterSet, bat *batch.Batch) (err error) { - // Output.Reset cannot propagate errors from its terminal nil callback. PERFORM - // finalizes saved-result metadata explicitly after runner.Run succeeds, where - // the error can still prevent the OK response. - if isPerformStatement(execCtx.stmt) && bat == nil { + // Output.Reset also invokes nil callbacks on failed attempts and cannot + // propagate their errors. PERFORM and deferred SELECT metadata finalize + // saved results explicitly after Run succeeds with the final schema. + if bat == nil && (isPerformStatement(execCtx.stmt) || execCtx.resultMetadata != nil) { return nil } if bat != nil { @@ -318,7 +318,7 @@ func (resper *MysqlResp) finalizeQueryResult(execCtx *ExecCtx) error { return resper.binWr.Write(execCtx, nil, nil) } -func finalizePerformQueryResult(execCtx *ExecCtx) error { +func finalizeQueryResult(execCtx *ExecCtx) error { finalizer, ok := execCtx.resper.(queryResultFinalizer) if !ok { return nil diff --git a/pkg/frontend/result_row_stmt.go b/pkg/frontend/result_row_stmt.go index 2c8749ea97f79..b099caed00e15 100644 --- a/pkg/frontend/result_row_stmt.go +++ b/pkg/frontend/result_row_stmt.go @@ -18,6 +18,7 @@ import ( "bufio" "context" "strings" + "sync" "time" "go.uber.org/zap" @@ -31,6 +32,7 @@ import ( plan2 "github.com/matrixorigin/matrixone/pkg/sql/plan" "github.com/matrixorigin/matrixone/pkg/sql/plan/explain" "github.com/matrixorigin/matrixone/pkg/sql/schedule" + "github.com/matrixorigin/matrixone/pkg/util/fault" ) func GetExplainColumn(ctx context.Context, explainColName string) ([]*plan2.ColDef, []interface{}, error) { @@ -321,6 +323,40 @@ func freezeResultMetadata(runner ComputationRunner) { } } +// deferredResultMetadata belongs to one statement. A definition retry may +// replace its plan before output starts; once metadata is published, the +// existing compiler guard prevents a retry from changing that visible schema. +type deferredResultMetadata struct { + once sync.Once + initialize func() error + err error +} + +func (metadata *deferredResultMetadata) publish() error { + metadata.once.Do(func() { metadata.err = metadata.initialize() }) + return metadata.err +} + +func publishSelectResultMetadata(ses *Session, execCtx *ExecCtx) error { + if txnCW, ok := execCtx.cw.(*TxnComputationWrapper); ok { + if runningCompile, ok := execCtx.runner.(Compile); ok { + txnCW.syncCompileExecution(runningCompile) + if runningCompile.PlanGenerationRebuilt() { + // Binary execution may retain column packets from the original + // prepared generation. Derive fresh packets from the current plan. + execCtx.prepareColDef = nil + } + } + } + columns, colDefs, err := getSelectColumnsAndResultColumns(execCtx.reqCtx, execCtx.cw) + if err != nil { + return err + } + ses.rs = &plan.ResultColDef{ResultCols: colDefs} + freezeResultMetadata(execCtx.runner) + return execCtx.resper.RespPreMeta(execCtx, columns) +} + // executeResultRowStmt run the statemet that responses result rows func executeResultRowStmt(ses *Session, execCtx *ExecCtx) (err error) { var columns []interface{} @@ -373,22 +409,21 @@ func executeResultRowStmt(ses *Session, execCtx *ExecCtx) (err error) { } switch statement := execCtx.stmt.(type) { case *tree.Select: - - columns, colDefs, err = getSelectColumnsAndResultColumns(execCtx.reqCtx, execCtx.cw) - if err != nil { - ses.Error(execCtx.reqCtx, - "Failed to get columns from computation handler", - zap.Error(err)) - return - } - - ses.rs = &plan.ResultColDef{ResultCols: colDefs} - ses.EnterFPrint(FPResultRowStmtSelect1) defer ses.ExitFPrint(FPResultRowStmtSelect1) - freezeResultMetadata(execCtx.runner) + if txnCW, ok := execCtx.cw.(*TxnComputationWrapper); ok && txnCW.preparedStmt != nil { + // Hold a prepared execution after binding but before any result + // metadata is retained or published for deterministic DDL races. + fault.TriggerFaultWithContext(execCtx.reqCtx, "prepared-result-metadata-bound") + } cursorExecute := execCtx.input != nil && execCtx.input.isCursorExecute if cursorExecute { + columns, colDefs, err = getSelectColumnsAndResultColumns(execCtx.reqCtx, execCtx.cw) + if err != nil { + return err + } + ses.rs = &plan.ResultColDef{ResultCols: colDefs} + freezeResultMetadata(execCtx.runner) // A cursor must retain its metadata before the pipeline starts so // captured batches can be decoded, but its execute terminator must // not be sent until all batches have materialized successfully. @@ -398,9 +433,8 @@ func executeResultRowStmt(ses *Session, execCtx *ExecCtx) (err error) { return moerr.NewInternalError(execCtx.reqCtx, "prepared cursor requires MySQL response writer") } } else { - err = execCtx.resper.RespPreMeta(execCtx, columns) - if err != nil { - return + execCtx.resultMetadata = &deferredResultMetadata{ + initialize: func() error { return publishSelectResultMetadata(ses, execCtx) }, } } @@ -415,6 +449,18 @@ func executeResultRowStmt(ses *Session, execCtx *ExecCtx) (err error) { if _, err = execCtx.runner.Run(0); err != nil { return } + if !cursorExecute { + // Empty results never invoke a positive-row output callback. Publish + // their metadata only after the successful final plan is known. + if err = execCtx.resultMetadata.publish(); err != nil { + return err + } + // Pipeline reset callbacks also run on failed attempts. Finalize a + // saved result only after Run succeeds with its final schema. + if err = finalizeQueryResult(execCtx); err != nil { + return err + } + } // Cursor metadata is retained above for decoding, but its wire response // is emitted by respStreamResultRow after transaction finalization. This // prevents a later autocommit commit error from following a successful diff --git a/pkg/frontend/routine_test.go b/pkg/frontend/routine_test.go index e118e908c4318..e0b5dcd2fa44f 100644 --- a/pkg/frontend/routine_test.go +++ b/pkg/frontend/routine_test.go @@ -38,6 +38,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/common/moerr" moruntime "github.com/matrixorigin/matrixone/pkg/common/runtime" "github.com/matrixorigin/matrixone/pkg/config" + "github.com/matrixorigin/matrixone/pkg/container/batch" "github.com/matrixorigin/matrixone/pkg/container/vector" "github.com/matrixorigin/matrixone/pkg/defines" mock_frontend "github.com/matrixorigin/matrixone/pkg/frontend/test" @@ -46,6 +47,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/pb/query" "github.com/matrixorigin/matrixone/pkg/pb/timestamp" "github.com/matrixorigin/matrixone/pkg/pb/txn" + "github.com/matrixorigin/matrixone/pkg/perfcounter" "github.com/matrixorigin/matrixone/pkg/queryservice" "github.com/matrixorigin/matrixone/pkg/sql/parsers" "github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect" @@ -2409,6 +2411,7 @@ var newMockWrapper = func(ctrl *gomock.Controller, ses *Session, panic(fmt.Sprintf("there is no mysqlResultset for the sql %s", sql)) } uuid, _ := uuid.NewV7() + var execCtx *ExecCtx runner := mock_frontend.NewMockComputationRunner(ctrl) runner.EXPECT().Run(gomock.Any()).DoAndReturn(func(uint64) (*util.RunResult, error) { proto := ses.GetResponser().MysqlRrWr() @@ -2429,6 +2432,13 @@ var newMockWrapper = func(ctrl *gomock.Controller, ses *Session, res.resultX.Store(contextCancel) } } + // This fake writes prebuilt rows directly rather than through the + // pipeline callback, so it must honor the same metadata boundary. + if mrs.GetRowCount() > 0 && execCtx.resultMetadata != nil { + if err = execCtx.resultMetadata.publish(); err != nil { + return nil, err + } + } err = proto.WriteResultSetRow(mrs, mrs.GetRowCount()) if err != nil { logutil.Errorf("flush error %v", err) @@ -2441,7 +2451,10 @@ var newMockWrapper = func(ctrl *gomock.Controller, ses *Session, mcw.EXPECT().GetAst().Return(stmt).AnyTimes() mcw.EXPECT().GetProcess().Return(proc).AnyTimes() mcw.EXPECT().GetColumns(gomock.Any()).Return(columns, nil).AnyTimes() - mcw.EXPECT().Compile(gomock.Any(), gomock.Any()).Return(runner, nil).AnyTimes() + mcw.EXPECT().Compile(gomock.Any(), gomock.Any()).DoAndReturn(func(ec any, _ func(*batch.Batch, *perfcounter.CounterSet) error) (any, error) { + execCtx = ec.(*ExecCtx) + return runner, nil + }).AnyTimes() mcw.EXPECT().GetUUID().Return(uuid[:]).AnyTimes() mcw.EXPECT().RecordExecPlan(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() mcw.EXPECT().RecordCompoundStmt(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() diff --git a/pkg/frontend/select_result_metadata_test.go b/pkg/frontend/select_result_metadata_test.go new file mode 100644 index 0000000000000..f77618181286f --- /dev/null +++ b/pkg/frontend/select_result_metadata_test.go @@ -0,0 +1,249 @@ +// 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" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/matrixorigin/matrixone/pkg/container/batch" + "github.com/matrixorigin/matrixone/pkg/container/types" + "github.com/matrixorigin/matrixone/pkg/defines" + "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/perfcounter" + "github.com/matrixorigin/matrixone/pkg/sql/parsers/tree" + "github.com/matrixorigin/matrixone/pkg/util" +) + +type selectMetadataTestCompile struct { + mockCompile + freeze func() +} + +func (c *selectMetadataTestCompile) FreezeResultMetadata() { c.freeze() } + +type selectMetadataTestResponder struct { + Responser + before func(*ExecCtx, any) error + finish func(*ExecCtx) error +} + +func (r *selectMetadataTestResponder) RespPreMeta(ec *ExecCtx, columns any) error { + return r.before(ec, columns) +} + +func (r *selectMetadataTestResponder) finalizeQueryResult(ec *ExecCtx) error { + return r.finish(ec) +} + +func TestSelectResultMetadataUsesExecutedPlan(t *testing.T) { + for _, rows := range []int{0, 2} { + t.Run(map[int]string{0: "empty", 2: "rows"}[rows], func(t *testing.T) { + original := newResultColumnTestPlan(1) + original.GetQuery().Nodes[0].ProjectList[0].Typ.Id = int32(types.T_int32) + final := newResultColumnTestPlan(1) + stmt := &tree.Select{} + ses := &Session{feSessionImpl: feSessionImpl{txnHandler: &TxnHandler{}}} + ec := &ExecCtx{ + reqCtx: context.Background(), ses: ses, stmt: stmt, + cw: &TxnComputationWrapper{stmt: stmt, plan: original}, + prepareColDef: [][]byte{{0x42}}, + } + var headers, freezes, written, finalized int + runFinished := false + runner := &selectMetadataTestCompile{freeze: func() { freezes++ }} + runner.getPlanFunc = func() *plan.Plan { return final } + runner.planGenerationRebuilt = true + ec.runner = runner + ec.resper = &selectMetadataTestResponder{ + before: func(ec *ExecCtx, columns any) error { + headers++ + require.Equal(t, 1, freezes) + require.Nil(t, ec.prepareColDef) + require.Equal(t, defines.MYSQL_TYPE_LONGLONG, columns.([]any)[0].(Column).ColumnType()) + require.Equal(t, int32(types.T_int64), ses.rs.ResultCols[0].Typ.Id) + require.Equal(t, rows == 0, runFinished) + return nil + }, + finish: func(*ExecCtx) error { + require.True(t, runFinished) + require.Equal(t, 1, headers) + finalized++ + return nil + }, + } + ses.outputCallback = func(_ FeSession, _ *ExecCtx, bat *batch.Batch, _ *perfcounter.CounterSet) error { + if bat != nil && bat.RowCount() > 0 { + require.Equal(t, 1, headers, "metadata must precede every row") + written += bat.RowCount() + } + return nil + } + callback := ses.GetOutputCallback(ec) + runner.runFunc = func(uint64) (*util.RunResult, error) { + // A failed attempt can emit terminal/empty callbacks before retry. + require.NoError(t, callback(nil, nil)) + require.NoError(t, callback(batch.NewWithSize(0), nil)) + require.Zero(t, headers) + require.Zero(t, freezes) + for range rows { + bat := batch.NewWithSize(0) + bat.SetRowCount(1) + require.NoError(t, callback(bat, nil)) + } + runFinished = true + return &util.RunResult{}, nil + } + require.NoError(t, executeResultRowStmt(ses, ec)) + require.Equal(t, rows, written) + require.Equal(t, 1, headers) + require.Equal(t, 1, freezes) + require.Equal(t, 1, finalized) + }) + } +} + +func TestSelectResultMetadataFailureBoundaries(t *testing.T) { + wantErr := errors.New("metadata publication failed") + for _, tc := range []struct { + name string + runErr error + metadataErr error + finalizeErr error + }{ + {name: "run failure", runErr: wantErr}, + {name: "cancellation", runErr: context.Canceled}, + {name: "metadata failure", metadataErr: wantErr}, + {name: "saved metadata failure", finalizeErr: wantErr}, + } { + t.Run(tc.name, func(t *testing.T) { + ses := &Session{feSessionImpl: feSessionImpl{txnHandler: &TxnHandler{}}} + stmt := &tree.Select{} + ec := &ExecCtx{reqCtx: context.Background(), ses: ses, stmt: stmt, + cw: &TxnComputationWrapper{stmt: stmt, plan: newResultColumnTestPlan(1)}} + headers, written, finalized := 0, 0, 0 + ec.resper = &selectMetadataTestResponder{ + before: func(*ExecCtx, any) error { headers++; return tc.metadataErr }, + finish: func(*ExecCtx) error { finalized++; return tc.finalizeErr }, + } + ses.outputCallback = func(FeSession, *ExecCtx, *batch.Batch, *perfcounter.CounterSet) error { + written++ + return nil + } + callback := ses.GetOutputCallback(ec) + ec.runner = &mockCompile{runFunc: func(uint64) (*util.RunResult, error) { + if tc.runErr != nil { + return nil, tc.runErr + } + if tc.finalizeErr != nil { + return &util.RunResult{}, nil + } + bat := batch.NewWithSize(0) + bat.SetRowCount(1) + first := callback(bat, nil) + require.ErrorIs(t, callback(bat, nil), first) + return nil, first + }, getPlanFunc: func() *plan.Plan { return ec.cw.Plan() }} + err := executeResultRowStmt(ses, ec) + if tc.runErr != nil { + require.ErrorIs(t, err, tc.runErr) + require.Zero(t, headers) + } else if tc.metadataErr != nil { + require.ErrorIs(t, err, tc.metadataErr) + require.Equal(t, 1, headers) + } else { + require.ErrorIs(t, err, tc.finalizeErr) + require.Equal(t, 1, headers) + } + require.Zero(t, written) + if tc.finalizeErr != nil { + require.Equal(t, 1, finalized) + } else { + require.Zero(t, finalized) + } + // A later statement must never inherit this publisher's cached error. + ec.beginStatementGeneration(&UserInput{}) + require.Nil(t, ec.resultMetadata) + require.NoError(t, callback(batch.NewWithSize(0), nil)) + require.Equal(t, 1, written) + }) + } +} + +func TestSelectResultMetadataConcurrentCallbacks(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + started, release := make(chan struct{}), make(chan struct{}) + var headers, rows atomic.Int32 + var published atomic.Bool + ec := &ExecCtx{resultMetadata: &deferredResultMetadata{initialize: func() error { + headers.Add(1) + close(started) + select { + case <-release: + published.Store(true) + return nil + case <-ctx.Done(): + return ctx.Err() + } + }}} + ses := &Session{} + ses.outputCallback = func(FeSession, *ExecCtx, *batch.Batch, *perfcounter.CounterSet) error { + if !published.Load() { + return errors.New("row preceded metadata") + } + rows.Add(1) + return nil + } + callback := ses.GetOutputCallback(ec) + bat := batch.NewWithSize(0) + bat.SetRowCount(1) + var workers sync.WaitGroup + var errs [2]error + workers.Go(func() { errs[0] = callback(bat, nil) }) + select { + case <-started: + case <-ctx.Done(): + workers.Wait() + t.Fatal(ctx.Err()) + } + workers.Go(func() { errs[1] = callback(bat, nil) }) + close(release) + workers.Wait() + for _, err := range errs { + require.NoError(t, err) + } + require.Equal(t, int32(1), headers.Load()) + require.Equal(t, int32(2), rows.Load()) +} + +func TestSelectResultSaverFinalizesAfterRun(t *testing.T) { + wantErr := errors.New("saved metadata write failed") + saver := &performTestBinaryWriter{err: wantErr} + resper := &MysqlResp{binWr: saver} + ec := &ExecCtx{stmt: &tree.Select{}, resultMetadata: &deferredResultMetadata{}} + // A retry/reset nil callback cannot finalize saved metadata prematurely. + require.NoError(t, resper.RespResult(ec, nil, nil)) + require.Zero(t, saver.calls) + ec.resper = resper + require.ErrorIs(t, finalizeQueryResult(ec), wantErr) + require.Equal(t, 1, saver.calls) +} diff --git a/pkg/frontend/session.go b/pkg/frontend/session.go index ad06a9f8eef33..26590b30a62ed 100644 --- a/pkg/frontend/session.go +++ b/pkg/frontend/session.go @@ -2620,6 +2620,11 @@ func (ses *Session) GetOutputCallback(execCtx *ExecCtx) func(*batch.Batch, *perf ses.mu.Lock() defer ses.mu.Unlock() return func(bat *batch.Batch, crs *perfcounter.CounterSet) error { + if execCtx != nil && execCtx.resultMetadata != nil && bat != nil && bat.RowCount() > 0 { + if err := execCtx.resultMetadata.publish(); err != nil { + return err + } + } if execCtx != nil && execCtx.input != nil && execCtx.input.isCursorExecute { if err := capturePreparedCursorBatch(ses, execCtx, bat); err != nil { return err diff --git a/pkg/frontend/status_stmt.go b/pkg/frontend/status_stmt.go index 41c3b25783516..250fb961d4840 100644 --- a/pkg/frontend/status_stmt.go +++ b/pkg/frontend/status_stmt.go @@ -81,7 +81,7 @@ func executeStatusStmt(ses *Session, execCtx *ExecCtx) (err error) { if execCtx.runResult, err = execCtx.runner.Run(0); err != nil { return } - if err = finalizePerformQueryResult(execCtx); err != nil { + if err = finalizeQueryResult(execCtx); err != nil { return } queryResultFinalized = true diff --git a/pkg/frontend/types.go b/pkg/frontend/types.go index 3fba7ec55fb39..8f2bb125285f4 100644 --- a/pkg/frontend/types.go +++ b/pkg/frontend/types.go @@ -1186,6 +1186,7 @@ type ExecCtx struct { results []ExecResult prepareColDef [][]byte cursorResultSaver StagedBinaryWriter + resultMetadata *deferredResultMetadata returning *returningState selectInto *selectIntoUserVariables isIssue3482 bool @@ -1203,6 +1204,7 @@ type ExecCtx struct { } func (execCtx *ExecCtx) beginStatementGeneration(input *UserInput) { + execCtx.resultMetadata = nil execCtx.effectiveTxnDefaultDatabase = "" execCtx.effectiveTxnStatement = nil execCtx.implicitCommitBefore = false @@ -1289,6 +1291,7 @@ func (execCtx *ExecCtx) Close() { execCtx.resper = nil execCtx.results = nil execCtx.prepareColDef = nil + execCtx.resultMetadata = nil execCtx.selectInto = nil execCtx.rewriteEnabled = false } diff --git a/pkg/sql/compile/compile.go b/pkg/sql/compile/compile.go index 72e0f2deb1911..b844f6eea82d5 100644 --- a/pkg/sql/compile/compile.go +++ b/pkg/sql/compile/compile.go @@ -275,7 +275,7 @@ func (c *Compile) SetPlanGenerationReused(reused bool) { // frontend or another streaming consumer. func (c *Compile) FreezeResultMetadata() { if c != nil { - c.resultMetadataFrozen = true + c.resultMetadataFrozen.Store(true) } } @@ -312,7 +312,7 @@ func (c *Compile) Reset(proc *process.Process, startAt time.Time, fill func(*bat c.clearLoadUniqueIndexPromotion() c.executionGeneration = 0 c.retryTimes = 0 - c.resultMetadataFrozen = false + c.resultMetadataFrozen.Store(false) c.anal.Reset(c.isPrepare, c.IsTpQuery()) if c.lockMeta != nil { @@ -523,7 +523,7 @@ func (c *Compile) clear() { c.planGenerationReused = false c.stringShuffleHashAlgorithm = process.StringShuffleHashLegacy c.stringShuffleHashAlgorithmFrozen = false - c.resultMetadataFrozen = false + c.resultMetadataFrozen.Store(false) c.planGenerationRebuilt = false c.sequenceState = sequenceStatementState{} diff --git a/pkg/sql/compile/compile2.go b/pkg/sql/compile/compile2.go index 695443dde5150..28623f7337b04 100644 --- a/pkg/sql/compile/compile2.go +++ b/pkg/sql/compile/compile2.go @@ -1226,7 +1226,7 @@ func (c *Compile) validateRetryResultMetadata( // result never reaches the row callback that also checks arity. return moerr.NewWrongNumberOfColumnsInSelect(ctx) } - if !c.resultMetadataFrozen || sameResultMetadata(c.pn, rebuilt) { + if !c.resultMetadataFrozen.Load() || sameResultMetadata(c.pn, rebuilt) { return nil } // Returning the definition-change error from buildRetryCompile is terminal diff --git a/pkg/sql/compile/types.go b/pkg/sql/compile/types.go index ce24a0f9997de..fb15c8e073d20 100644 --- a/pkg/sql/compile/types.go +++ b/pkg/sql/compile/types.go @@ -345,7 +345,7 @@ type Compile struct { // resultMetadataFrozen is set once a streaming consumer has materialized or // sent the current result schema. A definition retry may continue only when // the rebuilt logical plan exposes identical result metadata. - resultMetadataFrozen bool + resultMetadataFrozen atomic.Bool // planGenerationRebuilt is sticky for this Compile. Once a retry rebuilds // its logical plan, any frontend-owned prepared plan or physical topology // from the previous generation must not be reused. diff --git a/pkg/tests/sqlintegration/prepared_schema_commit_test.go b/pkg/tests/sqlintegration/prepared_schema_commit_test.go new file mode 100644 index 0000000000000..142dd1f0124e4 --- /dev/null +++ b/pkg/tests/sqlintegration/prepared_schema_commit_test.go @@ -0,0 +1,252 @@ +// 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 sqlintegration + +import ( + "context" + "database/sql" + "errors" + "fmt" + "testing" + "time" + + "github.com/matrixorigin/matrixone/pkg/embed" + "github.com/matrixorigin/matrixone/pkg/pb/metadata" + "github.com/matrixorigin/matrixone/pkg/pb/txn" + "github.com/matrixorigin/matrixone/pkg/util/fault" + "github.com/stretchr/testify/require" +) + +func TestPreparedSchemaCommitBeforeMetadata(t *testing.T) { + // This pessimistic two-CN scenario owns a disposable generation rather than + // the shared optimistic one-CN fixture used by neighboring tests. + require.NoError(t, embed.CloseSingleCNBaseClusterTests()) + cluster, err := embed.StartTestCluster(embed.WithCNCount(2), embed.WithPreStart(func(op embed.ServiceOperator) { + if op.ServiceType() == metadata.ServiceType_CN { + op.Adjust(func(cfg *embed.ServiceConfig) { cfg.CN.Txn.Mode = txn.TxnMode_Pessimistic.String() }) + } + })) + if cluster != nil { + t.Cleanup(func() { require.NoError(t, cluster.Close()) }) + } + require.NoError(t, err) + ctx, cancel := context.WithTimeout(t.Context(), 120*time.Second) + t.Cleanup(cancel) + open := func(index int) *sql.Conn { + cn, err := cluster.GetCNService(index) + require.NoError(t, err) + db, err := sql.Open("mysql", fmt.Sprintf("dump:111@tcp(127.0.0.1:%d)/", cn.GetServiceConfig().CN.Frontend.Port)) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, db.Close()) }) + conn, err := db.Conn(ctx) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, conn.Close()) }) + return conn + } + first, second := open(0), open(1) + exec := func(t *testing.T, conn *sql.Conn, statement string) { + t.Helper() + _, err := conn.ExecContext(ctx, statement) + require.NoError(t, err, statement) + } + exec(t, first, "create database prepared_schema_commit") + t.Cleanup(func() { + cleanup, stop := context.WithTimeout(context.Background(), 15*time.Second) + defer stop() + _, err := first.ExecContext(cleanup, "rollback") + require.NoError(t, err) + _, err = first.ExecContext(cleanup, "drop database prepared_schema_commit") + require.NoError(t, err) + }) + exec(t, first, "use prepared_schema_commit") + exec(t, second, "use prepared_schema_commit") + for _, protocol := range []string{"text", "binary"} { + for _, scenario := range []string{"committed_before_execute", "empty_after_binding", "rows_after_binding"} { + t.Run(protocol+"/"+scenario, func(t *testing.T) { + table := protocol + "_" + scenario + exec(t, first, "create table "+table+" (a int primary key,b int)") + exec(t, first, "begin") + t.Cleanup(func() { + cleanup, stop := context.WithTimeout(context.Background(), 15*time.Second) + defer stop() + _, err := first.ExecContext(cleanup, "rollback") + require.NoError(t, err) + }) + exec(t, first, "alter table "+table+" modify b bigint") + wantRows := scenario == "rows_after_binding" + if wantRows { + exec(t, first, "insert into "+table+" values (2,2147483648)") + } + statement := "select * from " + table + " where a > ? for update" + var prepared *sql.Stmt + if protocol == "text" { + exec(t, second, "prepare schema_commit from "+statement) + exec(t, second, "set @a=1") + } else { + prepared, err = second.PrepareContext(ctx, statement) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, prepared.Close()) }) + } + queryCtx, stopQuery := context.WithCancel(ctx) + defer stopQuery() + type result struct { + types []string + rows [][2]int64 + err error + } + query := func() result { + var rows *sql.Rows + var err error + if protocol == "text" { + rows, err = second.QueryContext(queryCtx, "execute schema_commit using @a") + } else { + rows, err = prepared.QueryContext(queryCtx, 1) + } + if err != nil { + return result{err: err} + } + defer rows.Close() + columns, err := rows.ColumnTypes() + if err != nil { + return result{err: err} + } + out := result{} + for _, column := range columns { + out.types = append(out.types, column.DatabaseTypeName()) + } + for rows.Next() { + var row [2]int64 + if err := rows.Scan(&row[0], &row[1]); err != nil { + out.err = err + return out + } + out.rows = append(out.rows, row) + } + out.err = errors.Join(rows.Err(), rows.Close()) + return out + } + var got result + if scenario == "committed_before_execute" { + exec(t, first, "commit") + got = query() + } else { + wasEnabled := fault.Status() + fault.Enable() + t.Cleanup(func() { + _, err := fault.RemoveFaultPoint(context.Background(), "prepared-result-metadata-bound") + require.NoError(t, err) + _, err = fault.RemoveFaultPoint(context.Background(), "prepared-result-metadata-waiters") + require.NoError(t, err) + if !wasEnabled { + fault.Disable() + } + }) + require.NoError(t, fault.AddFaultPoint(ctx, "prepared-result-metadata-bound", ":::", "wait", 0, "", false)) + require.NoError(t, fault.AddFaultPoint(ctx, "prepared-result-metadata-waiters", ":::", "getwaiters", 0, "prepared-result-metadata-bound", false)) + done := make(chan result, 1) + go func() { done <- query() }() + joined := false + defer func() { + _, _ = fault.RemoveFaultPoint(context.Background(), "prepared-result-metadata-bound") + stopQuery() + if !joined { + select { + case <-done: + case <-ctx.Done(): + t.Errorf("query cleanup: %v", ctx.Err()) + } + } + }() + require.Eventually(t, func() bool { + waiters, _, exists := fault.TriggerFault("prepared-result-metadata-waiters") + return exists && waiters == 1 + }, 5*time.Second, time.Millisecond) + exec(t, first, "commit") + _, err := fault.RemoveFaultPoint(ctx, "prepared-result-metadata-bound") + require.NoError(t, err) + select { + case got = <-done: + joined = true + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + } + require.NoError(t, got.err) + require.Equal(t, []string{"INT", "BIGINT"}, got.types) + if wantRows { + require.Equal(t, [][2]int64{{2, 2147483648}}, got.rows) + } else { + require.Empty(t, got.rows) + } + if protocol == "text" { + exec(t, second, "deallocate prepare schema_commit") + } + // A later statement cannot inherit a failed/unfinished metadata publisher. + var value int + require.NoError(t, second.QueryRowContext(ctx, "select 7").Scan(&value)) + require.Equal(t, 7, value) + }) + } + } + // Saved metadata has the same publication boundary, including zero rows. + exec(t, second, "set save_query_result=on") + for _, limit := range []string{"0", "1"} { + t.Run("saved_result/limit_"+limit, func(t *testing.T) { + rows, err := second.QueryContext(ctx, "/* save_result */ select b from text_rows_after_binding limit "+limit) + require.NoError(t, err) + defer rows.Close() + columns, err := rows.ColumnTypes() + require.NoError(t, err) + require.Len(t, columns, 1) + require.Equal(t, "BIGINT", columns[0].DatabaseTypeName()) + count := 0 + for rows.Next() { + var value int64 + require.NoError(t, rows.Scan(&value)) + require.Equal(t, int64(2147483648), value) + count++ + } + require.NoError(t, rows.Err()) + require.NoError(t, rows.Close()) + if limit == "0" { + require.Zero(t, count) + } else { + require.Equal(t, 1, count) + } + var saved, generated int + require.NoError(t, second.QueryRowContext(ctx, + "select savedRowCount,queryRowCount from meta_scan(last_query_id()) as r").Scan(&saved, &generated)) + require.Equal(t, count, saved) + require.Equal(t, count, generated) + if count > 0 { + // The persisted schema must also decode the saved BIGINT exactly. + replayed, err := second.QueryContext(ctx, "select b from result_scan(last_query_id(-2)) as r") + require.NoError(t, err) + defer replayed.Close() + columns, err := replayed.ColumnTypes() + require.NoError(t, err) + require.Len(t, columns, 1) + require.Equal(t, "BIGINT", columns[0].DatabaseTypeName()) + require.True(t, replayed.Next()) + var value int64 + require.NoError(t, replayed.Scan(&value)) + require.Equal(t, int64(2147483648), value) + require.False(t, replayed.Next()) + require.NoError(t, replayed.Err()) + require.NoError(t, replayed.Close()) + } + }) + } +} From 6efb5fdfdaad60c2f458963ab143ec5ae5ec812f Mon Sep 17 00:00:00 2001 From: bRong Njam Date: Fri, 9 Oct 2026 23:57:41 +0800 Subject: [PATCH 2/2] fix(frontend): clear saved SELECT state after late failures --- pkg/frontend/query_result.go | 6 +++ pkg/frontend/result_row_stmt.go | 10 +++++ pkg/frontend/select_result_metadata_test.go | 34 +++++++++++++++++ .../prepared_schema_commit_test.go | 38 +++++++++++++++++++ 4 files changed, 88 insertions(+) diff --git a/pkg/frontend/query_result.go b/pkg/frontend/query_result.go index e7e858cb36ef3..4dee6ced436d4 100644 --- a/pkg/frontend/query_result.go +++ b/pkg/frontend/query_result.go @@ -40,6 +40,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/perfcounter" "github.com/matrixorigin/matrixone/pkg/sql/colexec" "github.com/matrixorigin/matrixone/pkg/sql/parsers/tree" + "github.com/matrixorigin/matrixone/pkg/util/fault" "github.com/matrixorigin/matrixone/pkg/util/resource" "github.com/matrixorigin/matrixone/pkg/util/trace/impl/motrace/statistic" "github.com/matrixorigin/matrixone/pkg/vm/process" @@ -383,6 +384,11 @@ func saveQueryResult2(execCtx *ExecCtx, crs *perfcounter.CounterSet, bat *batch. if err := saveBatch(newCtx, ses, bat); err != nil { return err } + // This inert test point fails a producer only after its batch has + // been persisted, before SELECT can publish completion metadata. + if _, _, injected := fault.TriggerFaultWithContext(execCtx.reqCtx, "saved-result-after-batch"); injected { + return moerr.NewInternalError(execCtx.reqCtx, "injected saved-result batch failure") + } } } return nil diff --git a/pkg/frontend/result_row_stmt.go b/pkg/frontend/result_row_stmt.go index b099caed00e15..4deaf87bae4a4 100644 --- a/pkg/frontend/result_row_stmt.go +++ b/pkg/frontend/result_row_stmt.go @@ -417,6 +417,15 @@ func executeResultRowStmt(ses *Session, execCtx *ExecCtx) (err error) { fault.TriggerFaultWithContext(execCtx.reqCtx, "prepared-result-metadata-bound") } cursorExecute := execCtx.input != nil && execCtx.input.isCursorExecute + queryResultFinalized := false + defer func() { + // A producer or terminal step can fail after batches were saved. + // Cursor staging has its own abort owner; ordinary SELECT must clear + // its session state before another statement can save a result. + if !cursorExecute && !queryResultFinalized { + resetQueryResultState(ses) + } + }() if cursorExecute { columns, colDefs, err = getSelectColumnsAndResultColumns(execCtx.reqCtx, execCtx.cw) if err != nil { @@ -460,6 +469,7 @@ func executeResultRowStmt(ses *Session, execCtx *ExecCtx) (err error) { if err = finalizeQueryResult(execCtx); err != nil { return err } + queryResultFinalized = true } // Cursor metadata is retained above for decoding, but its wire response // is emitted by respStreamResultRow after transaction finalization. This diff --git a/pkg/frontend/select_result_metadata_test.go b/pkg/frontend/select_result_metadata_test.go index f77618181286f..346aff8e665f9 100644 --- a/pkg/frontend/select_result_metadata_test.go +++ b/pkg/frontend/select_result_metadata_test.go @@ -247,3 +247,37 @@ func TestSelectResultSaverFinalizesAfterRun(t *testing.T) { require.ErrorIs(t, finalizeQueryResult(ec), wantErr) require.Equal(t, 1, saver.calls) } + +func TestSelectResultFailureResetsSavedState(t *testing.T) { + for _, wantErr := range []error{errors.New("late producer failure"), context.Canceled} { + t.Run(wantErr.Error(), func(t *testing.T) { + ses := &Session{feSessionImpl: feSessionImpl{txnHandler: &TxnHandler{}}} + stmt := &tree.Select{} + ec := &ExecCtx{reqCtx: context.Background(), ses: ses, stmt: stmt, + cw: &TxnComputationWrapper{stmt: stmt, plan: newResultColumnTestPlan(1)}} + ec.resper = &selectMetadataTestResponder{ + before: func(*ExecCtx, any) error { return nil }, + finish: func(*ExecCtx) error { t.Fatal("failed Run must not finalize saved metadata"); return nil }, + } + ses.outputCallback = func(FeSession, *ExecCtx, *batch.Batch, *perfcounter.CounterSet) error { + // Model the session state owned by an already persisted batch. + ses.blockIdx = 1 + ses.p = ec.cw.Plan() + ses.curResultSize = 1 + ses.savedRowCount = 1 + ses.queryRowCount = 1 + return nil + } + callback := ses.GetOutputCallback(ec) + ec.runner = &mockCompile{runFunc: func(uint64) (*util.RunResult, error) { + bat := batch.NewWithSize(0) + bat.SetRowCount(1) + require.NoError(t, callback(bat, nil)) + require.Equal(t, 1, ses.blockIdx) + return nil, wantErr + }, getPlanFunc: func() *plan.Plan { return ec.cw.Plan() }} + require.ErrorIs(t, executeResultRowStmt(ses, ec), wantErr) + requirePerformResultStateReset(t, ses) + }) + } +} diff --git a/pkg/tests/sqlintegration/prepared_schema_commit_test.go b/pkg/tests/sqlintegration/prepared_schema_commit_test.go index 142dd1f0124e4..977771dfcfebd 100644 --- a/pkg/tests/sqlintegration/prepared_schema_commit_test.go +++ b/pkg/tests/sqlintegration/prepared_schema_commit_test.go @@ -249,4 +249,42 @@ func TestPreparedSchemaCommitBeforeMetadata(t *testing.T) { } }) } + t.Run("saved_result/late_failure_and_same_connection_reuse", func(t *testing.T) { + var before, after uint64 + require.NoError(t, second.QueryRowContext(ctx, "select connection_id()").Scan(&before)) + wasEnabled := fault.Status() + fault.Enable() + t.Cleanup(func() { + _, err := fault.RemoveFaultPoint(context.Background(), "saved-result-after-batch") + require.NoError(t, err) + if !wasEnabled { + fault.Disable() + } + }) + require.NoError(t, fault.AddFaultPoint(ctx, "saved-result-after-batch", ":::", "return", 0, "", false)) + rows, queryErr := second.QueryContext(ctx, + "/* save_result */ select b from text_rows_after_binding limit 1") + if queryErr == nil { + defer rows.Close() + for rows.Next() { + var value int64 + require.NoError(t, rows.Scan(&value)) + } + queryErr = errors.Join(rows.Err(), rows.Close()) + } + require.ErrorContains(t, queryErr, "injected saved-result batch failure") + _, err := fault.RemoveFaultPoint(ctx, "saved-result-after-batch") + require.NoError(t, err) + require.NoError(t, second.QueryRowContext(ctx, "select connection_id()").Scan(&after)) + require.Equal(t, before, after) + // A fresh statement owns exactly one saved block; stale blockIdx would + // make its metadata reference a second, nonexistent result object. + var value int64 + require.NoError(t, second.QueryRowContext(ctx, + "/* save_result */ select b from text_rows_after_binding limit 1").Scan(&value)) + require.Equal(t, int64(2147483648), value) + require.NoError(t, second.QueryRowContext(ctx, + "select b from result_scan(last_query_id()) as r").Scan(&value)) + require.Equal(t, int64(2147483648), value) + }) }