fix fk meta

pkg/bootstrap/service_upgrade_tenant.go

@@ -247,8 +247,12 @@ func (s *service) asyncUpgradeTenantTask(ctx context.Context) {
 						zap.String("tenant-version", createVersion),
 						zap.String("upgrade", upgrade.String()))
 
-					// createVersion >= upgrade.ToVersion already upgrade
-					if versions.Compare(createVersion, upgrade.ToVersion) >= 0 {
+					// A version-offset refresh can create an upgrade whose source and
+					// target version strings are identical. Reapply that handler so
+					// tenants at the old offset receive its idempotent metadata changes.
+					versionComparison := versions.Compare(createVersion, upgrade.ToVersion)
+					if versionComparison > 0 ||
+						(versionComparison == 0 && upgrade.FromVersion != upgrade.ToVersion) {
 						continue
 					}
 

pkg/bootstrap/service_upgrade_tenant_test.go

@@ -158,6 +158,83 @@ func Test_asyncUpgradeTenantTask_SkipsTenantAtTargetVersion(t *testing.T) {
 	)
 }
 
+func Test_asyncUpgradeTenantTask_ReappliesSameVersionOffsetUpgrade(t *testing.T) {
+	sid := ""
+	runtime.RunTest(
+		sid,
+		func(rt runtime.Runtime) {
+			var taskReady atomic.Bool
+			var finalized atomic.Bool
+			var tenantVersionUpdated atomic.Bool
+			ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*50)
+			defer cancel()
+
+			sqlExecutor := executor.NewMemExecutor(func(sql string) (executor.Result, error) {
+				switch {
+				case strings.Contains(sql, "from mo_upgrade") &&
+					strings.Contains(sql, "where state = 1") &&
+					!strings.Contains(sql, "for update"):
+					return buildUpgradeVersionResult(100, 1, "4.0.6", "4.0.6", 9, 1, 1, 1, 1, 0), nil
+				case strings.Contains(sql, "from mo_upgrade_tenant where from_account_id >= 0"):
+					if taskReady.Load() {
+						return executor.Result{}, nil
+					}
+					return buildUpgradeTenantTaskRows([]uint64{200}, []int32{10}, []int32{10}), nil
+				case strings.Contains(sql, "select account_id, create_version from mo_account where account_id >= 10 and account_id <= 10"):
+					return buildUpgradeTenantAccountRows([]int32{10}, []string{"4.0.6"}), nil
+				case strings.Contains(sql, "update mo_account set create_version = '4.0.6' where account_id = 10"):
+					tenantVersionUpdated.Store(true)
+					return executor.Result{AffectedRows: 1}, nil
+				case strings.Contains(sql, "update mo_upgrade_tenant set ready = 1") &&
+					strings.Contains(sql, "where id = 200"):
+					taskReady.Store(true)
+					return executor.Result{AffectedRows: 1}, nil
+				case strings.Contains(sql, "select 1 from mo_upgrade_tenant where upgrade_id = 100 and ready = 0"):
+					return executor.Result{}, nil
+				case strings.Contains(sql, "from mo_upgrade") &&
+					strings.Contains(sql, "where id = 100 for update"):
+					return buildUpgradeVersionResult(100, 1, "4.0.6", "4.0.6", 9, 1, 1, 1, 1, 0), nil
+				case strings.Contains(sql, "update mo_upgrade set total_tenant = 1, ready_tenant = 0"):
+					return executor.Result{AffectedRows: 1}, nil
+				case strings.Contains(sql, "update mo_upgrade set total_tenant = 1, ready_tenant = 1") &&
+					strings.Contains(sql, "state = 2"):
+					finalized.Store(true)
+					cancel()
+					return executor.Result{AffectedRows: 1}, nil
+				default:
+					return executor.Result{}, fmt.Errorf("unexpected sql: %s", sql)
+				}
+			})
+
+			h := newTestVersionHandler("4.0.6", "4.0.5", versions.Yes, versions.Yes, 9)
+			s := newServiceForTest(
+				sid,
+				&memLocker{},
+				clock.NewHLCClock(func() int64 { return 0 }, 0),
+				nil,
+				sqlExecutor,
+				func(s *service) {
+					s.handles = append(s.handles, h)
+				},
+				WithCheckUpgradeTenantDuration(time.Millisecond),
+			)
+
+			txnOperator := mock_frontend.NewMockTxnOperator(gomock.NewController(t))
+			txnOperator.EXPECT().TxnOptions().Return(txn.TxnOptions{CN: sid}).AnyTimes()
+
+			s.exec = executor.NewMemExecutor2(func(sql string) (executor.Result, error) {
+				return sqlExecutor.Exec(context.Background(), sql, executor.Options{})
+			}, txnOperator)
+
+			s.asyncUpgradeTenantTask(ctx)
+			require.True(t, taskReady.Load())
+			require.True(t, finalized.Load())
+			require.Equal(t, uint64(1), h.callHandleTenantUpgrade.Load())
+			require.True(t, tenantVersionUpdated.Load())
+		},
+	)
+}
+
 func Test_asyncUpgradeTenantTask_AutoCompletesDeletedTenantTasks(t *testing.T) {
 	sid := ""
 	runtime.RunTest(

pkg/bootstrap/versions/v4_0_6/tenant_upgrade_list.go

@@ -14,6 +14,8 @@
 package v4_0_6
 
 import (
+	"fmt"
+
 	"github.com/matrixorigin/matrixone/pkg/bootstrap/versions"
 	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"github.com/matrixorigin/matrixone/pkg/sql/mongodb"
@@ -24,9 +26,27 @@ import (
 var tenantUpgEntries = []versions.UpgradeEntry{
 	newMongoDBCatalogTable(mongodb.TableConnections, mongodb.ConnectionsDDL),
 	newMongoDBCatalogTable(mongodb.TableMappings, mongodb.MappingsDDL),
+	addForeignKeyMetadataColumn("referenced_index_name", "varchar(5000) not null default ''", "on_update"),
+	addForeignKeyMetadataColumn("on_delete_origin", "varchar(64) not null default 'ACTION_ORIGIN_LEGACY_AMBIGUOUS'", "referenced_index_name"),
+	addForeignKeyMetadataColumn("on_update_origin", "varchar(64) not null default 'ACTION_ORIGIN_LEGACY_AMBIGUOUS'", "on_delete_origin"),
+	upgradeInformationSchemaKeyColumnUsage(),
+	upgradeInformationSchemaReferentialConstraints(),
 	populateInformationSchemaCharacterSets(),
 }
 
+func addForeignKeyMetadataColumn(column, definition, after string) versions.UpgradeEntry {
+	return versions.UpgradeEntry{
+		Schema:    catalog.MO_CATALOG,
+		TableName: catalog.MOForeignKeys,
+		UpgType:   versions.ADD_COLUMN,
+		UpgSql:    fmt.Sprintf("alter table %s.%s add column %s %s after %s", catalog.MO_CATALOG, catalog.MOForeignKeys, column, definition, after),
+		CheckFunc: func(txn executor.TxnExecutor, accountID uint32) (bool, error) {
+			columnInfo, err := versions.CheckTableColumn(txn, accountID, catalog.MO_CATALOG, catalog.MOForeignKeys, column)
+			return columnInfo.IsExits, err
+		},
+	}
+}
+
 func populateInformationSchemaCharacterSets() versions.UpgradeEntry {
 	return versions.UpgradeEntry{
 		Schema:    sysview.InformationDBConst,
@@ -59,3 +79,35 @@ func newMongoDBCatalogTable(name, ddl string) versions.UpgradeEntry {
 		},
 	}
 }
+
+func upgradeInformationSchemaKeyColumnUsage() versions.UpgradeEntry {
+	return versions.UpgradeEntry{
+		Schema:    sysview.InformationDBConst,
+		TableName: "KEY_COLUMN_USAGE",
+		UpgType:   versions.CREATE_VIEW,
+		UpgSql:    sysview.InformationSchemaKeyColumnUsageDDL,
+		CheckFunc: checkViewDefinition("KEY_COLUMN_USAGE", sysview.InformationSchemaKeyColumnUsageDDL),
+		PreSql:    fmt.Sprintf("DROP TABLE IF EXISTS %s.%s;", sysview.InformationDBConst, "KEY_COLUMN_USAGE"),
+	}
+}
+
+func upgradeInformationSchemaReferentialConstraints() versions.UpgradeEntry {
+	return versions.UpgradeEntry{
+		Schema:    sysview.InformationDBConst,
+		TableName: "REFERENTIAL_CONSTRAINTS",
+		UpgType:   versions.MODIFY_VIEW,
+		UpgSql:    sysview.InformationSchemaReferentialConstraintsDDL,
+		CheckFunc: checkViewDefinition("REFERENTIAL_CONSTRAINTS", sysview.InformationSchemaReferentialConstraintsDDL),
+		PreSql:    fmt.Sprintf("DROP VIEW IF EXISTS %s.%s;", sysview.InformationDBConst, "REFERENTIAL_CONSTRAINTS"),
+	}
+}
+
+func checkViewDefinition(viewName, definition string) func(executor.TxnExecutor, uint32) (bool, error) {
+	return func(txn executor.TxnExecutor, accountID uint32) (bool, error) {
+		exists, viewDef, err := versions.CheckViewDefinition(txn, accountID, sysview.InformationDBConst, viewName)
+		if err != nil {
+			return false, err
+		}
+		return exists && viewDef == definition, nil
+	}
+}