fix(frontend): publish prepared result metadata after schema retry

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()
 }

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) {

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)

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

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

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,30 @@ 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
+		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 {
+				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 +442,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 +458,19 @@ 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
+			}
+			queryResultFinalized = true
+		}
 		// 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

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()