From 4e573492b3bd1a70c8e2b5de576f02db15c4f83a Mon Sep 17 00:00:00 2001 From: Cao Kai Date: Mon, 3 Aug 2026 11:10:19 +0800 Subject: [PATCH 1/2] refactor(plan): migrate remaining update special cases --- pkg/sql/plan/bind_update.go | 36 +-------------- pkg/sql/plan/update_planner_route.go | 23 +++++----- pkg/sql/plan/update_planner_route_test.go | 53 ++++++++++++++++++++--- 3 files changed, 59 insertions(+), 53 deletions(-) diff --git a/pkg/sql/plan/bind_update.go b/pkg/sql/plan/bind_update.go index 01e2a7cbb1c84..596ea6ad4574c 100644 --- a/pkg/sql/plan/bind_update.go +++ b/pkg/sql/plan/bind_update.go @@ -135,34 +135,7 @@ func (builder *QueryBuilder) bindUpdate(stmt *tree.Update, bindCtx *BindContext) validIndexes, _ := getValidIndexes(tableDef) tableDef.Indexes = validIndexes - var pkAndUkCols = make(map[string]bool) - - if tableDef.Name == catalog.MO_PUBS || tableDef.Name == catalog.MO_SUBS { - for _, colName := range tableDef.Pkey.Names { - pkAndUkCols[colName] = true - } - } - - for _, idxDef := range tableDef.Indexes { - if !idxDef.Unique { - continue - } - - if tableDef.Name == catalog.MO_PUBS || tableDef.Name == catalog.MO_SUBS { - for _, colName := range idxDef.Parts { - pkAndUkCols[catalog.ResolveAlias(colName)] = true - } - } - } - for colName, updateExpr := range dmlCtx.updateCol2Expr[i] { - if pkAndUkCols[colName] { - return 0, newLegacyUpdatePlannerRouteError( - updateRouteReasonPubSubKey, - moerr.NewUnsupportedDML(builder.compCtx.GetContext(), "update pk/uk on pub/sub table"), - ) - } - // Check: cannot update a generated column (unless SET gen_col = DEFAULT) isGenCol := false for _, colDef := range tableDef.Cols { @@ -189,13 +162,6 @@ func (builder *QueryBuilder) bindUpdate(stmt *tree.Update, bindCtx *BindContext) for _, colDef := range tableDef.Cols { if colDef.Name == colName { if isEnumOrSetPlanType(&colDef.Typ) { - if colDef.Typ.AutoIncr { - return 0, newLegacyUpdatePlannerRouteError( - updateRouteReasonAutoIncrement, - moerr.NewUnsupportedDML(builder.compCtx.GetContext(), "auto_increment default value"), - ) - } - updateExpr, err = wrapAstExprForMySQLSpecialType(builder.GetContext(), colDef.Typ, updateExpr) if err != nil { return 0, err @@ -414,7 +380,7 @@ func (builder *QueryBuilder) bindUpdate(stmt *tree.Update, bindCtx *BindContext) if updateAutoIncrCols[i] && len(affectedUpdateChildFks(tableDef, dmlCtx.aliases[i], newColName2Idx)) > 0 { return 0, newLegacyUpdatePlannerRouteError( - updateRouteReasonAutoIncrement, + updateRouteReasonAutoIncrementFK, moerr.NewUnsupportedDML( builder.compCtx.GetContext(), "auto_increment foreign key update", diff --git a/pkg/sql/plan/update_planner_route.go b/pkg/sql/plan/update_planner_route.go index 9b2f4c089d288..75db41125cbf6 100644 --- a/pkg/sql/plan/update_planner_route.go +++ b/pkg/sql/plan/update_planner_route.go @@ -37,18 +37,17 @@ const ( type updatePlannerRouteReason string const ( - updateRouteReasonNone updatePlannerRouteReason = "none" - updateRouteReasonMultiTarget updatePlannerRouteReason = "multi_target" - updateRouteReasonForeignKey updatePlannerRouteReason = "foreign_key" - updateRouteReasonIrregularIndex updatePlannerRouteReason = "irregular_index" - updateRouteReasonPubSubKey updatePlannerRouteReason = "pub_sub_key" - updateRouteReasonAutoIncrement updatePlannerRouteReason = "enum_set_auto_increment" - updateRouteReasonIceberg updatePlannerRouteReason = "iceberg" - updateRouteReasonExternalTable updatePlannerRouteReason = "external_table" - updateRouteReasonTableForm updatePlannerRouteReason = "unsupported_table_form" - updateRouteReasonEmptyTableName updatePlannerRouteReason = "empty_table_name" - updateRouteReasonBinderError updatePlannerRouteReason = "binder_error" - updateRouteReasonUnknown updatePlannerRouteReason = "unknown" + updateRouteReasonNone updatePlannerRouteReason = "none" + updateRouteReasonMultiTarget updatePlannerRouteReason = "multi_target" + updateRouteReasonForeignKey updatePlannerRouteReason = "foreign_key" + updateRouteReasonIrregularIndex updatePlannerRouteReason = "irregular_index" + updateRouteReasonAutoIncrementFK updatePlannerRouteReason = "auto_increment_foreign_key" + updateRouteReasonIceberg updatePlannerRouteReason = "iceberg" + updateRouteReasonExternalTable updatePlannerRouteReason = "external_table" + updateRouteReasonTableForm updatePlannerRouteReason = "unsupported_table_form" + updateRouteReasonEmptyTableName updatePlannerRouteReason = "empty_table_name" + updateRouteReasonBinderError updatePlannerRouteReason = "binder_error" + updateRouteReasonUnknown updatePlannerRouteReason = "unknown" ) type updatePlannerRouteError struct { diff --git a/pkg/sql/plan/update_planner_route_test.go b/pkg/sql/plan/update_planner_route_test.go index a76c892e8a588..20d7c5a66e9c9 100644 --- a/pkg/sql/plan/update_planner_route_test.go +++ b/pkg/sql/plan/update_planner_route_test.go @@ -195,13 +195,28 @@ func TestBindUpdateProducesTypedPlannerRoutes(t *testing.T) { wantReason: updateRouteReasonNone, }, { - name: "pub sub key", + name: "pub primary key", sql: "UPDATE nation SET n_nationkey = 2", prepare: func(mock *MockOptimizer) { mock.ctxt.tables["nation"].Name = catalog.MO_PUBS }, - wantRoute: updatePlannerLegacy, - wantReason: updateRouteReasonPubSubKey, + wantRoute: updatePlannerModern, + wantReason: updateRouteReasonNone, + }, + { + name: "sub unique key", + sql: "UPDATE nation SET n_name = 'x'", + prepare: func(mock *MockOptimizer) { + tableDef := mock.ctxt.tables["nation"] + tableDef.Name = catalog.MO_SUBS + tableDef.Indexes = []*planpb.IndexDef{{ + IndexName: "uk_name", + Unique: true, + Parts: []string{"n_name"}, + }} + }, + wantRoute: updatePlannerModern, + wantReason: updateRouteReasonNone, }, { name: "set auto increment", @@ -212,8 +227,8 @@ func TestBindUpdateProducesTypedPlannerRoutes(t *testing.T) { col.Typ.Enumvalues = "one,two" col.Typ.AutoIncr = true }, - wantRoute: updatePlannerLegacy, - wantReason: updateRouteReasonAutoIncrement, + wantRoute: updatePlannerModern, + wantReason: updateRouteReasonNone, }, } @@ -249,6 +264,32 @@ func TestBindUpdateProducesTypedPlannerRoutes(t *testing.T) { } } +func TestRemainingUpdateSpecialCasesUseModernPlan(t *testing.T) { + t.Run("pub primary key uses multi update", func(t *testing.T) { + mock := NewMockOptimizer(true) + mock.ctxt.tables["nation"].Name = catalog.MO_PUBS + + logicPlan, err := runOneStmt(mock, t, "UPDATE nation SET n_nationkey = 2") + require.NoError(t, err) + require.Equal(t, 1, countUpdateFkPlanNodes(logicPlan.GetQuery(), planpb.Node_MULTI_UPDATE)) + }) + + t.Run("set auto increment keeps type conversion and pre insert", func(t *testing.T) { + mock := NewMockOptimizer(true) + col := mock.ctxt.tables["nation"].Cols[0] + col.Typ.Id = int32(types.T_uint64) + col.Typ.Enumvalues = "one,two" + col.Typ.AutoIncr = true + + logicPlan, err := runOneStmt(mock, t, "UPDATE nation SET n_nationkey = 'two'") + require.NoError(t, err) + query := logicPlan.GetQuery() + require.Equal(t, 1, countUpdateFkPlanNodes(query, planpb.Node_MULTI_UPDATE)) + require.Equal(t, 1, countUpdateFkPlanNodes(query, planpb.Node_PRE_INSERT)) + require.True(t, updateFkPlanContainsFunc(query, moSetCastValueToIndexFun)) + }) +} + func TestBindUpdateForeignKeyRoutingByAffectedColumns(t *testing.T) { prepareEmpDept := func(mock *MockOptimizer) { emp := mock.ctxt.tables["emp"] @@ -376,7 +417,7 @@ func TestBindUpdateForeignKeyRoutingByAffectedColumns(t *testing.T) { require.Error(t, err) route, reason, _ := classifyUpdatePlannerError(err) require.Equal(t, updatePlannerLegacy, route) - require.Equal(t, updateRouteReasonAutoIncrement, reason) + require.Equal(t, updateRouteReasonAutoIncrementFK, reason) }) t.Run("disabled checks keep auto increment child key on modern route", func(t *testing.T) { From cb39c93c3df10cc6a675d60ca915fbac2ed2e08e Mon Sep 17 00:00:00 2001 From: Cao Kai Date: Mon, 3 Aug 2026 12:05:58 +0800 Subject: [PATCH 2/2] test: cover mo_subs update migration --- pkg/tests/issues/issue_26342_test.go | 312 +++++++++++++++++++++++++++ 1 file changed, 312 insertions(+) create mode 100644 pkg/tests/issues/issue_26342_test.go diff --git a/pkg/tests/issues/issue_26342_test.go b/pkg/tests/issues/issue_26342_test.go new file mode 100644 index 0000000000000..8f88e60b3b22d --- /dev/null +++ b/pkg/tests/issues/issue_26342_test.go @@ -0,0 +1,312 @@ +// 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 issues + +import ( + "context" + "database/sql" + "fmt" + "strings" + "testing" + "time" + + "github.com/matrixorigin/matrixone/pkg/embed" + "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/tests/testutils" + "github.com/matrixorigin/matrixone/pkg/util/executor" + "github.com/stretchr/testify/require" +) + +func TestIssue26342MoSubsModernUpdate(t *testing.T) { + runAuthenticatedClusterTest(t, func(c embed.Cluster) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + + cn, err := c.GetCNService(0) + require.NoError(t, err) + port := cn.GetServiceConfig().CN.Frontend.Port + sysDB := openIssue26342DB(t, fmt.Sprintf("sys#root#moadmin:111@tcp(127.0.0.1:%d)/", port)) + defer func() { require.NoError(t, sysDB.Close()) }() + + suffix := time.Now().UnixNano() + tenantName := fmt.Sprintf("acc26342_%d", suffix) + execSQLRequire(t, ctx, sysDB, fmt.Sprintf( + "create account `%s` admin_name 'admin' identified by '111'", tenantName, + )) + defer func() { + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cleanupCancel() + execSQLMaybe(t, cleanupCtx, sysDB, fmt.Sprintf("drop account if exists `%s`", tenantName)) + }() + + tenantID := queryIssue26095AccountID(t, ctx, sysDB, tenantName) + require.NotZero(t, tenantID) + tenantDB := openIssue26342DB(t, fmt.Sprintf( + "%s#admin#accountadmin:111@tcp(127.0.0.1:%d)/", tenantName, port, + )) + defer func() { require.NoError(t, tenantDB.Close()) }() + + prefix := fmt.Sprintf("p26342_%d", suffix) + pubAccount := prefix + "_publisher" + pubA := prefix + "_pub_a" + pubA2 := prefix + "_pub_a2" + pubB := prefix + "_pub_b" + subA := prefix + "_sub_a" + subA2 := prefix + "_sub_a2" + subB := prefix + "_sub_b" + defer func() { + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cleanupCancel() + _ = execIssue26342InternalSQL(cleanupCtx, cn, fmt.Sprintf( + "delete from mo_catalog.mo_subs where pub_account_name = '%s'", pubAccount)) + }() + + assertIssue26342MoSubsDDL(t, ctx, sysDB) + hiddenTable := queryIssue26342MoSubsUniqueIndexTable(t, ctx, sysDB) + hiddenBefore := queryIssue26342TableCount(t, ctx, sysDB, hiddenTable) + + insertIssue26342MoSub(t, ctx, cn, tenantID, tenantName, subA, pubAccount, pubA) + insertIssue26342MoSub(t, ctx, cn, tenantID, tenantName, subB, pubAccount, pubB) + require.Equal(t, 2, queryIssue26342MoSubsCount(t, ctx, sysDB, tenantID, pubAccount)) + require.Equal(t, hiddenBefore+2, queryIssue26342TableCount(t, ctx, sysDB, hiddenTable)) + + logicalPlan, err := execIssue26342InternalSQLWithPlan(ctx, cn, fmt.Sprintf( + "update mo_catalog.mo_subs set pub_name = '%s', sub_name = '%s' "+ + "where pub_account_name = '%s' and pub_name = '%s' and sub_account_id = %d", + pubA2, subA2, pubAccount, pubA, tenantID, + )) + require.NoError(t, err) + require.NotNil(t, logicalPlan) + assertIssue26342MultiUpdateContexts(t, logicalPlan, hiddenTable) + require.Equal(t, 0, queryIssue26342MoSubKeyCount(t, ctx, sysDB, tenantID, pubAccount, pubA, subA)) + require.Equal(t, 1, queryIssue26342MoSubKeyCount(t, ctx, sysDB, tenantID, pubAccount, pubA2, subA2)) + require.Equal(t, hiddenBefore+2, queryIssue26342TableCount(t, ctx, sysDB, hiddenTable)) + + err = execIssue26342InternalSQL(ctx, cn, fmt.Sprintf( + "update mo_catalog.mo_subs set pub_name = '%s', sub_name = '%s' "+ + "where pub_account_name = '%s' and pub_name = '%s' and sub_account_id = %d", + pubB, subB, pubAccount, pubA2, tenantID, + )) + require.Error(t, err) + require.Equal(t, 1, queryIssue26342MoSubKeyCount(t, ctx, sysDB, tenantID, pubAccount, pubA2, subA2)) + require.Equal(t, 1, queryIssue26342MoSubKeyCount(t, ctx, sysDB, tenantID, pubAccount, pubB, subB)) + require.Equal(t, hiddenBefore+2, queryIssue26342TableCount(t, ctx, sysDB, hiddenTable)) + + require.NoError(t, execIssue26342InternalSQL(ctx, cn, fmt.Sprintf( + "update mo_catalog.mo_subs set sub_account_name = 'stale' "+ + "where pub_account_name = '%s' and sub_account_id = %d", + pubAccount, tenantID, + ))) + require.NoError(t, execIssue26342InternalSQL(ctx, cn, + "update mo_catalog.mo_subs t1 inner join mo_catalog.mo_account t2 "+ + "on t1.sub_account_id = t2.account_id set t1.sub_account_name = t2.account_name")) + require.Equal(t, tenantName, queryIssue26342SubAccountName(t, ctx, sysDB, tenantID, pubAccount, pubA2)) + + _, err = tenantDB.ExecContext(ctx, fmt.Sprintf( + "update mo_catalog.mo_subs set status = status + 1 "+ + "where pub_account_name = '%s' and pub_name = '%s' and sub_account_id = %d", + pubAccount, pubA2, tenantID, + )) + require.Error(t, err) + require.Equal(t, int64(1), queryIssue26342MoSubStatus(t, ctx, sysDB, tenantID, pubAccount, pubA2)) + }) +} + +func openIssue26342DB(t *testing.T, dsn string) *sql.DB { + t.Helper() + db, err := sql.Open("mysql", dsn) + require.NoError(t, err) + db.SetMaxOpenConns(1) + return db +} + +func assertIssue26342MoSubsDDL(t *testing.T, ctx context.Context, db *sql.DB) { + t.Helper() + var tableName, createSQL string + require.NoError(t, db.QueryRowContext(ctx, "show create table mo_catalog.mo_subs").Scan(&tableName, &createSQL)) + normalized := strings.ToLower(strings.Join(strings.Fields(strings.ReplaceAll(createSQL, "`", "")), " ")) + require.Contains(t, normalized, "primary key (pub_account_name,pub_name,sub_account_id)") + require.Contains(t, normalized, "unique key sub_account_id (sub_account_id,sub_name)") +} + +func queryIssue26342MoSubsUniqueIndexTable(t *testing.T, ctx context.Context, db *sql.DB) string { + t.Helper() + var tableName string + err := db.QueryRowContext(ctx, ` + select distinct i.index_table_name + from mo_catalog.mo_indexes i + join mo_catalog.mo_tables t on i.table_id = t.rel_id + where t.reldatabase = 'mo_catalog' + and t.relname = 'mo_subs' + and i.name <> 'PRIMARY' + and i.index_table_name <> '' + limit 1`).Scan(&tableName) + require.NoError(t, err) + require.NotEmpty(t, tableName) + return tableName +} + +func queryIssue26342TableCount(t *testing.T, ctx context.Context, db *sql.DB, tableName string) int { + t.Helper() + require.NotContains(t, tableName, "`") + var count int + err := db.QueryRowContext(ctx, fmt.Sprintf("select count(*) from mo_catalog.`%s`", tableName)).Scan(&count) + require.NoError(t, err) + return count +} + +func insertIssue26342MoSub( + t *testing.T, + ctx context.Context, + cn embed.ServiceOperator, + tenantID uint32, + tenantName string, + subName string, + pubAccount string, + pubName string, +) { + t.Helper() + err := execIssue26342InternalSQL(ctx, cn, fmt.Sprintf(` + insert into mo_catalog.mo_subs ( + sub_account_id, sub_account_name, sub_name, sub_time, + pub_account_id, pub_account_name, pub_name, pub_database, + pub_tables, pub_time, pub_comment, status + ) values (%d, '%s', '%s', now(), 0, '%s', '%s', 'db26342', '*', now(), 'test', 1)`, + tenantID, tenantName, subName, pubAccount, pubName)) + require.NoError(t, err) +} + +func execIssue26342InternalSQL(ctx context.Context, cn embed.ServiceOperator, sql string) error { + _, err := execIssue26342InternalSQLWithPlan(ctx, cn, sql) + return err +} + +func execIssue26342InternalSQLWithPlan( + ctx context.Context, + cn embed.ServiceOperator, + sql string, +) (*plan.Query, error) { + result, err := testutils.GetSQLExecutor(cn).Exec( + ctx, + sql, + executor.Options{}. + WithDatabase("mo_catalog"). + WithAccountID(0). + WithWaitCommittedLogApplied(), + ) + logicalPlan := result.LogicalPlan + result.Close() + return logicalPlan, err +} + +func assertIssue26342MultiUpdateContexts(t *testing.T, query *plan.Query, hiddenTable string) { + t.Helper() + var multiUpdate *plan.Node + for _, node := range query.Nodes { + if node.NodeType == plan.Node_MULTI_UPDATE { + multiUpdate = node + break + } + } + require.NotNil(t, multiUpdate) + require.Len(t, multiUpdate.UpdateCtxList, 2) + + contexts := make(map[string]*plan.UpdateCtx, len(multiUpdate.UpdateCtxList)) + for _, updateCtx := range multiUpdate.UpdateCtxList { + require.NotNil(t, updateCtx.TableDef) + require.NotEmpty(t, updateCtx.InsertCols) + require.NotEmpty(t, updateCtx.DeleteCols) + contexts[updateCtx.TableDef.Name] = updateCtx + } + baseCtx := contexts["mo_subs"] + require.NotNil(t, baseCtx) + require.NotNil(t, baseCtx.TableDef.Pkey) + require.Equal(t, + []string{"pub_account_name", "pub_name", "sub_account_id"}, + baseCtx.TableDef.Pkey.Names, + ) + require.NotNil(t, contexts[hiddenTable]) +} + +func queryIssue26342MoSubsCount( + t *testing.T, + ctx context.Context, + db *sql.DB, + tenantID uint32, + pubAccount string, +) int { + t.Helper() + var count int + err := db.QueryRowContext(ctx, + "select count(*) from mo_catalog.mo_subs where sub_account_id = ? and pub_account_name = ?", + tenantID, pubAccount).Scan(&count) + require.NoError(t, err) + return count +} + +func queryIssue26342MoSubKeyCount( + t *testing.T, + ctx context.Context, + db *sql.DB, + tenantID uint32, + pubAccount string, + pubName string, + subName string, +) int { + t.Helper() + var count int + err := db.QueryRowContext(ctx, ` + select count(*) from mo_catalog.mo_subs + where sub_account_id = ? and pub_account_name = ? and pub_name = ? and sub_name = ?`, + tenantID, pubAccount, pubName, subName).Scan(&count) + require.NoError(t, err) + return count +} + +func queryIssue26342SubAccountName( + t *testing.T, + ctx context.Context, + db *sql.DB, + tenantID uint32, + pubAccount string, + pubName string, +) string { + t.Helper() + var name string + err := db.QueryRowContext(ctx, ` + select sub_account_name from mo_catalog.mo_subs + where sub_account_id = ? and pub_account_name = ? and pub_name = ?`, + tenantID, pubAccount, pubName).Scan(&name) + require.NoError(t, err) + return name +} + +func queryIssue26342MoSubStatus( + t *testing.T, + ctx context.Context, + db *sql.DB, + tenantID uint32, + pubAccount string, + pubName string, +) int64 { + t.Helper() + var status int64 + err := db.QueryRowContext(ctx, ` + select status from mo_catalog.mo_subs + where sub_account_id = ? and pub_account_name = ? and pub_name = ?`, + tenantID, pubAccount, pubName).Scan(&status) + require.NoError(t, err) + return status +}