fix variables

pkg/common/moerr/error.go

@@ -102,6 +102,7 @@ const (
 	ErrUpdateTableUsed      uint16 = 20322
 	ErrWindowInvalidUse     uint16 = 20323
 	ErrViewSelectTmpTable   uint16 = 20324
+	ErrTooManyRows          uint16 = 20325
 
 	// Group 4: unexpected state and io errors
 	ErrInvalidState                             uint16 = 20400
@@ -431,6 +432,7 @@ var errorMsgRefer = map[uint16]moErrorMsgItem{
 	ErrUpdateTableUsed:      {ER_UPDATE_TABLE_USED, []string{MySQLDefaultSqlState}, "You can't specify target table '%-.192s' for update in FROM clause"},
 	ErrWindowInvalidUse:     {ER_WINDOW_INVALID_WINDOW_FUNC_USE, []string{"HY000"}, "You cannot use the window function '%s' in this context"},
 	ErrViewSelectTmpTable:   {ER_VIEW_SELECT_TMPTABLE, []string{MySQLDefaultSqlState}, "View's SELECT refers to a temporary table '%-.192s'"},
+	ErrTooManyRows:          {ER_TOO_MANY_ROWS, []string{"42000"}, "Result consisted of more than one row"},
 
 	// Group 4: unexpected state or file io error
 	ErrInvalidState:                             {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "invalid state %s"},
@@ -1529,6 +1531,10 @@ func NewErrSubqueryNo1Row(ctx context.Context) *Error {
 	return newError(ctx, ErrSubqueryNo1Row)
 }
 
+func NewTooManyRows(ctx context.Context) *Error {
+	return newError(ctx, ErrTooManyRows)
+}
+
 func NewDerivedMustHaveAlias(ctx context.Context) *Error {
 	return newError(ctx, ErrDerivedMustHaveAlias)
 }

pkg/common/moerr/error_test.go

@@ -216,6 +216,21 @@ func TestErrSubqueryNo1RowContract(t *testing.T) {
 	require.Equal(t, err, decoded)
 }
 
+func TestErrTooManyRowsContract(t *testing.T) {
+	err := NewTooManyRows(context.Background())
+	require.Equal(t, ErrTooManyRows, err.ErrorCode())
+	require.Equal(t, ER_TOO_MANY_ROWS, err.MySQLCode())
+	require.Equal(t, "42000", err.SqlState())
+	require.Equal(t, "Result consisted of more than one row", err.Error())
+
+	data, marshalErr := err.MarshalBinary()
+	require.NoError(t, marshalErr)
+
+	decoded := new(Error)
+	require.NoError(t, decoded.UnmarshalBinary(data))
+	require.Equal(t, err, decoded)
+}
+
 type fakeErr struct {
 }
 

pkg/frontend/compiler_context.go

@@ -808,7 +808,10 @@ func (tcc *TxnCompilerContext) ResolveVariable(varName string, isSystemVar, isGl
 	} else {
 		var udVar *UserDefinedVar
 		if udVar, err = tcc.GetSession().GetUserDefinedVar(varName); err != nil {
-			return nil, err
+			// MySQL creates user variables lazily.  Reading a variable that has
+			// not been assigned yet therefore evaluates to NULL rather than
+			// producing an unknown-variable error.
+			return nil, nil
 		}
 
 		varValue = udVar.Value
@@ -826,7 +829,9 @@ func (tcc *TxnCompilerContext) ResolveVariableIsBin(varName string, isSystemVar,
 	}
 	udVar, err := tcc.GetSession().GetUserDefinedVar(varName)
 	if err != nil {
-		return false, err
+		// See ResolveVariable: an unassigned user variable is NULL and has
+		// no binary-string attribute.
+		return false, nil
 	}
 	return udVar.IsBin, nil
 }

pkg/frontend/computation_wrapper.go

@@ -456,7 +456,7 @@ func (cwft *TxnComputationWrapper) Compile(any any, fill func(*batch.Batch, *per
 			if err = retComp.Reset(
 				cwft.proc,
 				getStatementStartAt(execCtx.reqCtx),
-				compileOutputCallback(cwft.stmt, fill),
+				compileOutputCallback(execCtx, cwft.ses, cwft.stmt, fill),
 				cwft.ses.GetSql(),
 			); err != nil {
 				return nil, err
@@ -1203,7 +1203,7 @@ func createCompile(
 			ctx, ses, ses.GetTxnCompileCtx(), stmt, forcePrepare)
 	})
 
-	err = retCompile.Compile(execCtx.reqCtx, plan, compileOutputCallback(stmt, fill))
+	err = retCompile.Compile(execCtx.reqCtx, plan, compileOutputCallback(execCtx, ses, stmt, fill))
 	if err != nil {
 		return
 	}
@@ -1217,14 +1217,16 @@ func createCompile(
 // callback. Apply the same rule both when compiling a fresh pipeline and when
 // resetting a cached prepared pipeline for another execution.
 func compileOutputCallback(
+	execCtx *ExecCtx,
+	ses FeSession,
 	stmt tree.Statement,
 	fill func(*batch.Batch, *perfcounter.CounterSet) error,
 ) func(*batch.Batch, *perfcounter.CounterSet) error {
 	switch stmt.(type) {
 	case *tree.ExplainAnalyze, *tree.ExplainPhyPlan:
 		return func(*batch.Batch, *perfcounter.CounterSet) error { return nil }
 	default:
-		return fill
+		return selectIntoUserVariablesOutputCallback(execCtx, ses, stmt, fill)
 	}
 }
 

pkg/frontend/computation_wrapper_test.go

@@ -189,8 +189,12 @@ func TestInitExecuteStmtParamPreservesBinaryFlagPerUserVariable(t *testing.T) {
 	isBin, err = ses.txnCompileCtx.ResolveVariableIsBin("system_var", true, false)
 	require.NoError(t, err)
 	require.False(t, isBin)
-	_, err = ses.txnCompileCtx.ResolveVariableIsBin("missing", false, false)
-	require.Error(t, err)
+	value, err := ses.txnCompileCtx.ResolveVariable("missing", false, false)
+	require.NoError(t, err)
+	require.Nil(t, value)
+	isBin, err = ses.txnCompileCtx.ResolveVariableIsBin("missing", false, false)
+	require.NoError(t, err)
+	require.False(t, isBin)
 	cw.proc.SetResolveVariableFunc(func(name string, _, _ bool) (interface{}, error) {
 		variable, err := ses.GetUserDefinedVar(name)
 		if err != nil {

pkg/frontend/prepared_explain_test.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"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"
@@ -185,13 +186,42 @@ func TestCompileOutputCallbackSuppressesExplainPipelineRows(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
+			execCtx := &ExecCtx{reqCtx: context.Background()}
 			called := false
 			fill := func(*batch.Batch, *perfcounter.CounterSet) error {
 				called = true
 				return nil
 			}
-			require.NoError(t, compileOutputCallback(tc.stmt, fill)(nil, nil))
+			require.NoError(t, compileOutputCallback(execCtx, nil, tc.stmt, fill)(nil, nil))
 			require.Equal(t, tc.wantCalled, called)
 		})
 	}
 }
+
+func TestCompileOutputCallbackCapturesSelectIntoUserVariables(t *testing.T) {
+	ctx := context.Background()
+	mp := mpool.MustNewZero()
+	defer mpool.DeleteMPool(mp)
+	ses := &Session{userDefinedVars: make(map[string]*UserDefinedVar)}
+	execCtx := &ExecCtx{reqCtx: ctx}
+	stmt := &tree.Select{IntoVars: []*tree.VarExpr{{Name: "out"}}}
+	called := false
+	fill := func(*batch.Batch, *perfcounter.CounterSet) error {
+		called = true
+		return nil
+	}
+
+	bat := batch.NewWithSize(1)
+	bat.Vecs[0] = vector.NewVec(types.T_int64.ToType())
+	require.NoError(t, vector.AppendFixed(bat.Vecs[0], int64(5), false, mp))
+	bat.SetRowCount(1)
+	defer bat.Clean(mp)
+
+	require.NoError(t, compileOutputCallback(execCtx, ses, stmt, fill)(bat, nil))
+	require.False(t, called)
+	require.NotNil(t, execCtx.selectInto)
+	require.NoError(t, execCtx.selectInto.apply(ctx, ses, "select abs(-5) into @out"))
+	variable, err := ses.GetUserDefinedVar("out")
+	require.NoError(t, err)
+	require.Equal(t, int64(5), variable.Value)
+}

pkg/frontend/select_into_user_variables.go

@@ -0,0 +1,96 @@
+// 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"
+	"sync"
+
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/perfcounter"
+	"github.com/matrixorigin/matrixone/pkg/sql/parsers/tree"
+)
+
+// selectIntoUserVariables collects the result of SELECT ... INTO @var.  The
+// output callback can receive several batches, so assignments are delayed
+// until the complete result is known.  This also keeps existing variables
+// unchanged for a zero-row query, as MySQL does.
+type selectIntoUserVariables struct {
+	mu       sync.Mutex
+	vars     []*tree.VarExpr
+	row      []any
+	rowCount uint64
+}
+
+func newSelectIntoUserVariables(vars []*tree.VarExpr) *selectIntoUserVariables {
+	return &selectIntoUserVariables{vars: vars}
+}
+
+func selectIntoUserVariablesOutputCallback(
+	execCtx *ExecCtx,
+	ses FeSession,
+	stmt tree.Statement,
+	fill func(*batch.Batch, *perfcounter.CounterSet) error,
+) func(*batch.Batch, *perfcounter.CounterSet) error {
+	selectStmt, ok := stmt.(*tree.Select)
+	if !ok || len(selectStmt.IntoVars) == 0 {
+		return fill
+	}
+	if execCtx.selectInto == nil {
+		execCtx.selectInto = newSelectIntoUserVariables(selectStmt.IntoVars)
+	}
+	return func(bat *batch.Batch, _ *perfcounter.CounterSet) error {
+		return execCtx.selectInto.capture(execCtx.reqCtx, ses, bat)
+	}
+}
+
+func (collector *selectIntoUserVariables) capture(ctx context.Context, ses FeSession, bat *batch.Batch) error {
+	if bat == nil || bat.RowCount() == 0 {
+		return nil
+	}
+	if len(bat.Vecs) != len(collector.vars) {
+		return moerr.NewInvalidInputf(ctx,
+			"SELECT INTO has %d expressions for %d user variables", len(bat.Vecs), len(collector.vars))
+	}
+
+	collector.mu.Lock()
+	defer collector.mu.Unlock()
+	if collector.rowCount == 0 {
+		collector.row = make([]any, len(collector.vars))
+		if err := extractRowFromEveryVector(ctx, ses, bat, 0, collector.row, false); err != nil {
+			return err
+		}
+	}
+	collector.rowCount += uint64(bat.RowCount())
+	return nil
+}
+
+func (collector *selectIntoUserVariables) apply(ctx context.Context, ses FeSession, sql string) error {
+	collector.mu.Lock()
+	defer collector.mu.Unlock()
+	if collector.rowCount == 0 {
+		return nil
+	}
+	if collector.rowCount > 1 {
+		return moerr.NewTooManyRows(ctx)
+	}
+	for i, variable := range collector.vars {
+		if err := ses.SetUserDefinedVar(variable.Name, collector.row[i], sql); err != nil {
+			return err
+		}
+	}
+	return nil
+}

pkg/frontend/select_into_user_variables_test.go

@@ -0,0 +1,71 @@
+// 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"
+	"testing"
+
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"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/sql/parsers/tree"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSelectIntoUserVariablesCapturesAndAssignsOneRow(t *testing.T) {
+	ctx := context.Background()
+	mp := mpool.MustNewZero()
+	defer mpool.DeleteMPool(mp)
+	ses := &Session{userDefinedVars: make(map[string]*UserDefinedVar)}
+	collector := newSelectIntoUserVariables([]*tree.VarExpr{{Name: "out"}})
+
+	bat := batch.NewWithSize(1)
+	bat.Vecs[0] = vector.NewVec(types.T_int64.ToType())
+	require.NoError(t, vector.AppendFixed(bat.Vecs[0], int64(5), false, mp))
+	bat.SetRowCount(1)
+	defer bat.Clean(mp)
+
+	require.NoError(t, collector.capture(ctx, ses, bat))
+	require.NoError(t, collector.apply(ctx, ses, "select abs(-5) into @out"))
+	variable, err := ses.GetUserDefinedVar("OUT")
+	require.NoError(t, err)
+	require.Equal(t, int64(5), variable.Value)
+	require.Equal(t, "select abs(-5) into @out", variable.Sql)
+}
+
+func TestSelectIntoUserVariablesZeroOrManyRowsDoNotAssign(t *testing.T) {
+	ctx := context.Background()
+	ses := &Session{userDefinedVars: make(map[string]*UserDefinedVar)}
+	require.NoError(t, ses.SetUserDefinedVar("out", "old", "set @out = 'old'"))
+
+	zeroRows := newSelectIntoUserVariables([]*tree.VarExpr{{Name: "out"}})
+	require.NoError(t, zeroRows.apply(ctx, ses, "select value into @out from empty_table"))
+	variable, err := ses.GetUserDefinedVar("out")
+	require.NoError(t, err)
+	require.Equal(t, "old", variable.Value)
+
+	manyRows := newSelectIntoUserVariables([]*tree.VarExpr{{Name: "out"}})
+	manyRows.row = []any{"new"}
+	manyRows.rowCount = 2
+	err = manyRows.apply(ctx, ses, "select value into @out from two_rows")
+	require.ErrorContains(t, err, "Result consisted of more than one row")
+	require.True(t, moerr.IsMoErrCode(err, moerr.ErrTooManyRows))
+	variable, err = ses.GetUserDefinedVar("out")
+	require.NoError(t, err)
+	require.Equal(t, "old", variable.Value)
+}

pkg/frontend/status_stmt.go

@@ -67,6 +67,22 @@ func executeStatusStmt(ses *Session, execCtx *ExecCtx) (err error) {
 			}
 			return
 		}
+		if len(st.IntoVars) > 0 {
+			runBegin := time.Now()
+			if execCtx.runResult, err = execCtx.runner.Run(0); err != nil {
+				return
+			}
+			if execCtx.selectInto == nil {
+				return moerr.NewInternalError(execCtx.reqCtx, "SELECT INTO user-variable collector is not initialized")
+			}
+			if err = execCtx.selectInto.apply(execCtx.reqCtx, ses, execCtx.sqlOfStmt); err != nil {
+				return
+			}
+			if time.Since(runBegin) > time.Second {
+				ses.Infof(execCtx.reqCtx, "time of Exec.Run : %s", time.Since(runBegin).String())
+			}
+			return
+		}
 		if ep.needExportToFile() {
 			defer ep.Close()
 			columns, err = execCtx.cw.GetColumns(execCtx.reqCtx)

pkg/frontend/types.go

@@ -933,6 +933,7 @@ type ExecCtx struct {
 	results           []ExecResult
 	prepareColDef     [][]byte
 	returning         *returningState
+	selectInto        *selectIntoUserVariables
 	isIssue3482       bool
 	// remapDb is the effective database remap (role/session/inline merged) for
 	// this statement. It is applied at the AST level to qualified references by
@@ -981,6 +982,7 @@ func (execCtx *ExecCtx) Close() {
 	execCtx.resper = nil
 	execCtx.results = nil
 	execCtx.prepareColDef = nil
+	execCtx.selectInto = nil
 	execCtx.rewriteEnabled = false
 }