fix: refresh persisted View metadata after dependency DDL

pkg/bootstrap/service.go

@@ -186,7 +186,7 @@ func (s *service) Bootstrap(ctx context.Context) error {
 
 	if ok, err := s.checkAlreadyBootstrappedWithRetry(ctx); ok {
 		s.logger.Info("mo already bootstrapped")
-		return nil
+		return s.refreshFinalVersionReadiness(ctx)
 	} else if err != nil {
 		return err
 	}
@@ -216,7 +216,10 @@ func (s *service) Bootstrap(ctx context.Context) error {
 			s.logger.Info("waiting bootstrap completed",
 				zap.Bool("result", ok),
 				zap.Error(err))
-			return err
+			if err != nil {
+				return err
+			}
+			return s.refreshFinalVersionReadiness(ctx)
 		}
 	}
 }
@@ -427,6 +430,42 @@ func (s *service) completeBootstrap() {
 	}
 
 	s.logger.Info("successfully completed bootstrap")
+	s.upgrade.finalVersionCompleted.Store(true)
+}
+
+func (s *service) refreshFinalVersionReadiness(ctx context.Context) error {
+	final := s.getFinalVersionHandle().Metadata()
+	result, err := s.exec.Exec(ctx, fmt.Sprintf(
+		"select state from %s.%s where version = '%s' and version_offset = %d",
+		catalog.MO_CATALOG, catalog.MOVersionTable, final.Version, final.VersionOffset),
+		executor.Options{}.
+			WithMinCommittedTS(s.now()).
+			WithWaitCommittedLogApplied().
+			WithAccountID(catalog.System_Account))
+	if err != nil {
+		return err
+	}
+	defer result.Close()
+	ready := false
+	rowsSeen := 0
+	result.ReadRows(func(rows int, columns []*vector.Vector) bool {
+		for row := range rows {
+			rowsSeen++
+			if rowsSeen > 1 {
+				return false
+			}
+			ready = vector.GetFixedAtWithTypeCheck[int32](columns[0], row) == versions.StateReady
+		}
+		return rowsSeen <= 1
+	})
+	if rowsSeen > 1 {
+		return moerr.NewInternalErrorf(ctx,
+			"duplicate final catalog version %s offset %d", final.Version, final.VersionOffset)
+	}
+	if ready {
+		s.upgrade.finalVersionCompleted.Store(true)
+	}
+	return nil
 }
 
 func (s *service) now() timestamp.Timestamp {

pkg/bootstrap/service_test.go

@@ -303,6 +303,9 @@ func TestBootstrapWithWait(t *testing.T) {
 				if sql == fmt.Sprintf("show tables from %s", bootstrappedCheckerDB) {
 					return newBootstrapStringResult(allBootstrappedCheckerTables()...), nil
 				}
+				if strings.HasPrefix(sql, "select state from mo_catalog.mo_version") {
+					return newBootstrapStateResult(versions.StateReady), nil
+				}
 				return executor.Result{}, nil
 			})
 
@@ -320,6 +323,7 @@ func TestBootstrapWithWait(t *testing.T) {
 
 			require.NoError(t, b.Bootstrap(ctx))
 			assert.True(t, n.Load() > 0)
+			assert.True(t, b.IsFinalVersionReady())
 		},
 	)
 }
@@ -789,6 +793,39 @@ func newBootstrapStringResult(values ...string) executor.Result {
 	return memRes.GetResult()
 }
 
+func newBootstrapStateResult(states ...int32) executor.Result {
+	memRes := executor.NewMemResult(
+		[]types.Type{types.T_int32.ToType()}, mpool.MustNewZero())
+	memRes.NewBatchWithRowCount(len(states))
+	executor.AppendFixedRows(memRes, 0, states)
+	return memRes.GetResult()
+}
+
+func TestFinalVersionReadinessRequiresExactReadyCatalogRow(t *testing.T) {
+	tests := []struct {
+		name   string
+		states []int32
+		ready  bool
+	}{
+		{name: "missing"},
+		{name: "created", states: []int32{versions.StateCreated}},
+		{name: "upgrading tenants", states: []int32{versions.StateUpgradingTenant}},
+		{name: "ready", states: []int32{versions.StateReady}, ready: true},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			exec := executor.NewMemExecutor(func(sql string) (executor.Result, error) {
+				require.Contains(t, sql, "where version = '4.0.6'")
+				return newBootstrapStateResult(tc.states...), nil
+			})
+			svc := NewService("", &memLocker{},
+				clock.NewHLCClock(func() int64 { return 0 }, 0), nil, exec)
+			require.NoError(t, svc.(*service).refreshFinalVersionReadiness(context.Background()))
+			require.Equal(t, tc.ready, svc.IsFinalVersionReady())
+		})
+	}
+}
+
 func newBootstrapTestContext(timeout time.Duration) (context.Context, context.CancelFunc) {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	frontendParameters := &config.FrontendParameters{}

pkg/bootstrap/types.go

@@ -56,6 +56,9 @@ type Service interface {
 	GetFinalVersion() string
 	// GetFinalVersionOffset Get mo final version offset, which is based on the current code
 	GetFinalVersionOffset() int32
+	// IsFinalVersionReady reports whether this CN has observed the catalog at
+	// its exact final version and offset in the READY state.
+	IsFinalVersionReady() bool
 	// Close close bootstrap service
 	Close() error
 }

pkg/bootstrap/upgrade.go

@@ -82,6 +82,10 @@ func (s *service) GetFinalVersionOffset() int32 {
 	return int32(s.handles[len(s.handles)-1].Metadata().VersionOffset)
 }
 
+func (s *service) IsFinalVersionReady() bool {
+	return s.upgrade.finalVersionCompleted.Load()
+}
+
 func (s *service) getVersionHandle(version string) VersionHandle {
 	for _, h := range s.handles {
 		if h.Metadata().Version == version {

pkg/bootstrap/versions/v4_0_6/cluster_upgrade_list.go

@@ -29,6 +29,26 @@ const retiredKafkaSinkTaskCode = 4
 
 var clusterUpgEntries = []versions.UpgradeEntry{
 	retireKafkaSinkDaemonTasks,
+	createMoViewDependencies,
+	createMoViewRefresh,
+}
+
+var createMoViewDependencies = newViewMetadataCatalogTable(
+	catalog.MO_VIEW_DEPENDENCIES, catalog.MoViewDependenciesDDL)
+
+var createMoViewRefresh = newViewMetadataCatalogTable(
+	catalog.MO_VIEW_REFRESH, catalog.MoViewRefreshDDL)
+
+func newViewMetadataCatalogTable(name, ddl string) versions.UpgradeEntry {
+	return versions.UpgradeEntry{
+		Schema:    catalog.MO_CATALOG,
+		TableName: name,
+		UpgType:   versions.CREATE_NEW_TABLE,
+		UpgSql:    ddl,
+		CheckFunc: func(txn executor.TxnExecutor, accountID uint32) (bool, error) {
+			return versions.CheckTableDefinition(txn, accountID, catalog.MO_CATALOG, name)
+		},
+	}
 }
 
 var retireKafkaSinkDaemonTasks = versions.UpgradeEntry{

pkg/bootstrap/versions/v4_0_6/upgrade_test.go

@@ -37,8 +37,17 @@ import (
 
 func TestUpgradeEntries(t *testing.T) {
 	require.Len(t, tenantUpgEntries, 9)
-	require.Len(t, clusterUpgEntries, 1)
+	require.Len(t, clusterUpgEntries, 3)
 	require.Equal(t, retireKafkaSinkDaemonTasks.UpgSql, clusterUpgEntries[0].UpgSql)
+	require.Equal(t, catalog.MO_VIEW_DEPENDENCIES, clusterUpgEntries[1].TableName)
+	require.Equal(t, catalog.MO_VIEW_REFRESH, clusterUpgEntries[2].TableName)
+	for _, entry := range clusterUpgEntries[1:] {
+		require.Equal(t, versions.CREATE_NEW_TABLE, entry.UpgType)
+		require.Contains(t, strings.ToLower(entry.UpgSql), "create cluster table mo_catalog.mo_view_")
+		require.NotContains(t, strings.ToLower(entry.UpgSql), "\n\t\taccount_id int")
+	}
+	require.Contains(t, catalog.MoViewDependenciesDDL,
+		"primary key(account_id, target_relation_id, dependency_ordinal)")
 	require.Equal(t, mongodb.TableConnections, tenantUpgEntries[0].TableName)
 	require.Equal(t, mongodb.TableMappings, tenantUpgEntries[1].TableName)
 	for _, entry := range tenantUpgEntries[:2] {

pkg/catalog/types.go

@@ -160,6 +160,11 @@ const (
 	// MO_SUBS subscriptions meta table
 	MO_SUBS = "mo_subs"
 
+	// MO_VIEW_DEPENDENCIES stores exact reverse bindings for persisted Views.
+	MO_VIEW_DEPENDENCIES = "mo_view_dependencies"
+	// MO_VIEW_REFRESH stores monotonic refresh state and worker leases.
+	MO_VIEW_REFRESH = "mo_view_refresh"
+
 	// MO_SNAPSHOTS
 	MO_SNAPSHOTS = "mo_snapshots"
 

pkg/catalog/view_metadata.go

@@ -0,0 +1,90 @@
+// 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 catalog
+
+const ViewRefreshStatusCurrent = "CURRENT"
+const ViewRefreshStatusPending = "PENDING"
+const ViewRefreshStatusDiscovering = "DISCOVERING"
+const ViewRefreshStatusRunning = "RUNNING"
+const ViewRefreshStatusInvalid = "INVALID"
+const ViewRefreshStatusLegacyScan = "LEGACY_SCAN"
+
+const LegacyViewScanCursorDatabase = "__mo_legacy_view_scan__"
+const LegacyViewScanCursorRelation = "__mo_legacy_view_scan_cursor__"
+
+const ViewMetadataLifecycleGateSQL = "select rel_id from mo_catalog.mo_tables " +
+	"where account_id=0 and reldatabase='mo_catalog' and relname='mo_view_refresh' for update"
+
+const MoViewDependenciesColumns = "account_id,target_database_id,target_relation_id," +
+	"target_logical_id,target_database_name,target_relation_name,dependency_ordinal,source_account_id," +
+	"source_database_id,source_relation_id,source_logical_id,source_database_name," +
+	"source_relation_name,source_database_name_key,source_relation_name_key," +
+	"source_relation_kind,subscription_name,publisher_account_id,snapshot_data," +
+	"lower_case_table_names,dependency_generation"
+
+const MoViewRefreshColumns = "account_id,target_database_id,target_relation_id," +
+	"target_logical_id,target_database_name,target_relation_name,target_generation," +
+	"completed_generation,status,failure_code,next_retry_at,lease_owner,lease_epoch," +
+	"lease_expires_at,attempts"
+
+const MoViewDependenciesDDL = `create cluster table mo_catalog.mo_view_dependencies (
+		target_database_id bigint unsigned not null,
+		target_relation_id bigint unsigned not null,
+		target_logical_id bigint unsigned not null,
+		target_database_name varchar(5000) not null,
+		target_relation_name varchar(5000) not null,
+		dependency_ordinal int unsigned not null,
+		source_account_id int unsigned not null,
+		source_database_id bigint unsigned not null,
+		source_relation_id bigint unsigned not null,
+		source_logical_id bigint unsigned not null,
+		source_database_name varchar(5000) not null,
+		source_relation_name varchar(5000) not null,
+		source_database_name_key varchar(64) not null,
+		source_relation_name_key varchar(64) not null,
+		source_relation_kind varchar(32) not null,
+		subscription_name varchar(5000) not null default '',
+		publisher_account_id int unsigned not null default 0,
+		snapshot_data text,
+		lower_case_table_names bigint not null,
+		dependency_generation bigint unsigned not null,
+		primary key(account_id, target_relation_id, dependency_ordinal),
+		index idx_view_dependency_source_id(source_account_id, source_database_id,
+			source_relation_id),
+		index idx_view_dependency_source_logical(source_account_id, source_database_id,
+			source_logical_id),
+		index idx_view_dependency_source_name(source_account_id, source_database_name_key,
+			source_relation_name_key)
+)`
+
+const MoViewRefreshDDL = `create cluster table mo_catalog.mo_view_refresh (
+		target_database_id bigint unsigned not null,
+		target_relation_id bigint unsigned not null,
+		target_logical_id bigint unsigned not null,
+		target_database_name varchar(5000) not null,
+		target_relation_name varchar(5000) not null,
+		target_generation bigint unsigned not null,
+		completed_generation bigint unsigned not null,
+		status varchar(32) not null,
+		failure_code int unsigned not null default 0,
+		next_retry_at timestamp null,
+		lease_owner varchar(128) not null default '',
+		lease_epoch bigint unsigned not null default 0,
+		lease_expires_at timestamp null,
+		attempts int unsigned not null default 0,
+		primary key(account_id, target_relation_id),
+		index idx_view_refresh_pending(status, next_retry_at, account_id,
+			target_relation_id)
+)`

pkg/clusterservice/cluster.go

@@ -585,15 +585,16 @@ func (c *cluster) copyServices() *services {
 
 func newCNService(cn logpb.CNStore) metadata.CNService {
 	return metadata.CNService{
-		ServiceID:              cn.UUID,
-		PipelineServiceAddress: cn.ServiceAddress,
-		SQLAddress:             cn.SQLAddress,
-		LockServiceAddress:     cn.LockServiceAddress,
-		ShardServiceAddress:    cn.ShardServiceAddress,
-		WorkState:              cn.WorkState,
-		Labels:                 cn.Labels,
-		QueryAddress:           cn.QueryAddress,
-		CommitID:               cn.CommitID,
+		ServiceID:                    cn.UUID,
+		PipelineServiceAddress:       cn.ServiceAddress,
+		SQLAddress:                   cn.SQLAddress,
+		LockServiceAddress:           cn.LockServiceAddress,
+		ShardServiceAddress:          cn.ShardServiceAddress,
+		WorkState:                    cn.WorkState,
+		Labels:                       cn.Labels,
+		QueryAddress:                 cn.QueryAddress,
+		CommitID:                     cn.CommitID,
+		ViewMetadataRefreshSupported: cn.ViewMetadataRefreshSupported,
 		// why set this cfg, cc https://github.com/matrixorigin/matrixone/issues/16537
 		// should be used in getCNList
 		CPUTotal: cn.Resource.CPUTotal,

pkg/clusterservice/view_metadata_capability.go

@@ -0,0 +1,39 @@
+// 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 clusterservice
+
+import "github.com/matrixorigin/matrixone/pkg/pb/metadata"
+
+// AllKnownCNsSupportViewMetadataRefresh is the rolling-upgrade activation
+// barrier. The heartbeat bit means both binary support and that the CN has
+// observed its exact final catalog version/offset in READY state. An absent
+// snapshot and any old, upgrading, or draining-unready CN keep legacy mode.
+func AllKnownCNsSupportViewMetadataRefresh(serviceID string) bool {
+	cluster, ready, err := lookupMOCluster(serviceID)
+	return err == nil && ready && allKnownCNsSupportViewMetadataRefresh(cluster)
+}
+
+func allKnownCNsSupportViewMetadataRefresh(cluster MOCluster) bool {
+	found, supported := false, true
+	cluster.GetCNService(NewSelectAll(), func(service metadata.CNService) bool {
+		found = true
+		if !service.ViewMetadataRefreshSupported {
+			supported = false
+			return false
+		}
+		return true
+	})
+	return found && supported
+}