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
@@ -181,6 +182,7 @@ const (
 	ErrTableMustHaveAVisibleColumn              uint16 = 20474
 	ErrKeyDoesNotExist                          uint16 = 20475
 	ErrMaxPreparedStmtCountReached              uint16 = 20476
+	ErrWrongNumberOfColumnsInSelect             uint16 = 20477
 
 	// Group 5: rpc errors
 	//
@@ -431,6 +433,9 @@ 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"},
+	ErrWrongNumberOfColumnsInSelect: {ER_WRONG_NUMBER_OF_COLUMNS_IN_SELECT, []string{"21000"},
+		"The used SELECT statements have a different number of columns"},
 
 	// Group 4: unexpected state or file io error
 	ErrInvalidState:                             {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "invalid state %s"},
@@ -1529,6 +1534,14 @@ func NewErrSubqueryNo1Row(ctx context.Context) *Error {
 	return newError(ctx, ErrSubqueryNo1Row)
 }
 
+func NewTooManyRows(ctx context.Context) *Error {
+	return newError(ctx, ErrTooManyRows)
+}
+
+func NewWrongNumberOfColumnsInSelect(ctx context.Context) *Error {
+	return newError(ctx, ErrWrongNumberOfColumnsInSelect)
+}
+
 func NewDerivedMustHaveAlias(ctx context.Context) *Error {
 	return newError(ctx, ErrDerivedMustHaveAlias)
 }

pkg/common/moerr/error_test.go

@@ -216,6 +216,36 @@ 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)
+}
+
+func TestErrWrongNumberOfColumnsInSelectContract(t *testing.T) {
+	err := NewWrongNumberOfColumnsInSelect(context.Background())
+	require.Equal(t, ErrWrongNumberOfColumnsInSelect, err.ErrorCode())
+	require.Equal(t, ER_WRONG_NUMBER_OF_COLUMNS_IN_SELECT, err.MySQLCode())
+	require.Equal(t, "21000", err.SqlState())
+	require.Equal(t, "The used SELECT statements have a different number of columns", 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/back_status_stmt.go

@@ -17,6 +17,8 @@ package frontend
 import (
 	"time"
 
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/sql/parsers/tree"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/disttae"
 )
 
@@ -30,11 +32,31 @@ func executeStatusStmtInBack(backSes *backSession,
 	if err != nil {
 		return err
 	}
+	intoVars := 0
+	if st, ok := execCtx.stmt.(*tree.Select); ok {
+		intoVars = len(st.IntoVars)
+	}
+	if intoVars > 0 {
+		if err = validateSelectIntoArity(execCtx.reqCtx, execCtx.cw.Plan(), intoVars); err != nil {
+			return
+		}
+	}
 
 	runBegin := time.Now()
 	if execCtx.runResult, err = execCtx.runner.Run(0); err != nil {
 		return
 	}
+	if intoVars > 0 {
+		if execCtx.selectInto == nil {
+			return moerr.NewInternalError(execCtx.reqCtx, "SELECT INTO user-variable collector is not initialized")
+		}
+		if err = execCtx.selectInto.apply(execCtx.reqCtx, backSes, execCtx.sqlOfStmt); err != nil {
+			return
+		}
+		if st, ok := execCtx.stmt.(*tree.Select); ok {
+			appendSelectIntoDeprecatedWarning(backSes, st.DeprecatedInto)
+		}
+	}
 	if isPerformStatement(execCtx.stmt) && execCtx.runResult != nil {
 		execCtx.runResult.AffectRows = 0
 	}

pkg/frontend/compiler_context.go

@@ -851,7 +851,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
@@ -860,6 +863,31 @@ func (tcc *TxnCompilerContext) ResolveVariable(varName string, isSystemVar, isGl
 	return
 }
 
+// ResolveVariableType returns the type fixed when a user variable was
+// assigned. The value itself may be represented as text for protocol and
+// compatibility reasons (notably DECIMAL), so planner numeric binding must
+// not infer a narrower type from a sibling literal.
+func (tcc *TxnCompilerContext) ResolveVariableType(varName string, isSystemVar, isGlobalVar bool) (plan2.Type, error) {
+	if isSystemVar {
+		return plan2.Type{}, nil
+	}
+	if tcc.execCtx != nil {
+		if value, ok := resolveStoredProcedureVariable(tcc.execCtx.reqCtx, varName); ok {
+			return inferUserDefinedVarType(value), nil
+		}
+	}
+	udVar, err := tcc.GetSession().GetUserDefinedVar(varName)
+	if err != nil {
+		// An unassigned user variable is NULL; TEXT is the neutral binding type
+		// and lets a numeric context perform the normal MySQL coercion.
+		return inferUserDefinedVarType(nil), nil
+	}
+	if udVar.Type.Id != 0 {
+		return udVar.Type, nil
+	}
+	return inferUserDefinedVarType(udVar.Value), nil
+}
+
 func (tcc *TxnCompilerContext) ResolveVariableIsBin(varName string, isSystemVar, _ bool) (bool, error) {
 	if _, ok := resolveStoredProcedureVariable(tcc.execCtx.reqCtx, varName); ok {
 		return false, nil
@@ -869,7 +897,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
 }
@@ -895,7 +925,9 @@ func (tcc *TxnCompilerContext) ResolveVariablePrepareParamKind(
 	}
 	udVar, err := tcc.GetSession().GetUserDefinedVar(varName)
 	if err != nil {
-		return vector.PrepareParamNone, err
+		// See ResolveVariable: an unassigned user variable is NULL and has no
+		// prepared-parameter conversion category.
+		return vector.PrepareParamNone, nil
 	}
 	return udVar.PrepareParamKind, 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
@@ -1265,7 +1265,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
 	}
@@ -1279,14 +1279,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

@@ -190,8 +190,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 {
@@ -487,6 +491,16 @@ func TestResolveVariableIsBinHonorsStoredProcedureScope(t *testing.T) {
 	kind, err = ses.txnCompileCtx.ResolveVariablePrepareParamKind("session_only", false, false)
 	require.NoError(t, err)
 	require.Equal(t, vector.PrepareParamNone, kind)
+
+	value, err = ses.txnCompileCtx.ResolveVariable("missing_user_var", false, false)
+	require.NoError(t, err)
+	require.Nil(t, value)
+	isBin, err = ses.txnCompileCtx.ResolveVariableIsBin("missing_user_var", false, false)
+	require.NoError(t, err)
+	require.False(t, isBin)
+	kind, err = ses.txnCompileCtx.ResolveVariablePrepareParamKind("missing_user_var", false, false)
+	require.NoError(t, err)
+	require.Equal(t, vector.PrepareParamNone, kind)
 }
 
 func TestBuildExecuteUserParamsHonorsStoredProcedureScope(t *testing.T) {

pkg/frontend/mysql_cmd_executor.go

@@ -928,17 +928,19 @@ func doSetVar(
 	var err error = nil
 	var ok bool
 	var userVarIsBin bool
+	var userVarType plan.Type
 	var userVarPrepareParamKind vector.PrepareParamKind
 	type evaluatedAssignment struct {
 		assign                  *tree.VarAssignmentExpr
 		value                   interface{}
 		userVarIsBin            bool
+		valueType               plan.Type
 		userVarPrepareParamKind vector.PrepareParamKind
 	}
 	evaluateAssignment := func(assign *tree.VarAssignmentExpr) (evaluatedAssignment, error) {
 		isBin := false
 		prepareParamKind := vector.PrepareParamNone
-		value, evalErr := getExprValueWithPrepareMeta(
+		value, valueType, evalErr := getExprValueWithPrepareMeta(
 			assign.Value, ses, execCtx, preparedExpression, &prepareParamKind, &isBin)
 		if evalErr != nil {
 			return evaluatedAssignment{}, evalErr
@@ -953,6 +955,7 @@ func doSetVar(
 			assign:                  assign,
 			value:                   value,
 			userVarIsBin:            isBin,
+			valueType:               valueType,
 			userVarPrepareParamKind: prepareParamKind,
 		}, nil
 	}
@@ -992,8 +995,8 @@ func doSetVar(
 				}
 			}
 		} else {
-			err = ses.setUserDefinedVarWithKind(
-				name, value, sql, userVarIsBin, userVarPrepareParamKind)
+			err = ses.setUserDefinedVarWithTypeAndKind(
+				name, value, sql, userVarIsBin, userVarType, userVarPrepareParamKind)
 			if err != nil {
 				return err
 			}
@@ -1006,6 +1009,7 @@ func doSetVar(
 		name := assign.Name
 		value := item.value
 		userVarIsBin = item.userVarIsBin
+		userVarType = item.valueType
 		userVarPrepareParamKind = item.userVarPrepareParamKind
 
 		//TODO : fix SET NAMES after parser is ready
@@ -1200,6 +1204,9 @@ func doShowErrors(ses *Session, execCtx *ExecCtx) error {
 	for i := info.length() - 1; i >= 0; i-- {
 		row := make([]interface{}, 3)
 		row[0] = "Error"
+		if i < len(info.levels) && info.levels[i] != "" {
+			row[0] = info.levels[i]
+		}
 		row[1] = int16(info.codes[i])
 		row[2] = info.msgs[i]
 		mrs.AddRow(row)

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