fix(view): decode signed account IDs in metadata admission

pkg/sql/compile/view_metadata_recovery.go

@@ -26,6 +26,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/pubsub"
 	moruntime "github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/common/sqlquote"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/defines"
 	"github.com/matrixorigin/matrixone/pkg/pb/api"
@@ -119,6 +120,41 @@ func RequireViewMetadataRevalidation(ctx context.Context, sqlExecutor executor.S
 
 const viewMetadataRevalidationSeedComplete = uint32(^uint32(0))
 
+func decodeViewMetadataAccountIDs(column *vector.Vector, rows int) ([]uint32, error) {
+	if column == nil {
+		return nil, moerr.NewInternalErrorNoCtx("view metadata account_id vector is nil")
+	}
+	if rows < 0 {
+		return nil, moerr.NewInternalErrorNoCtxf(
+			"view metadata account_id vector has invalid row count %d", rows)
+	}
+	if column.GetType().Oid != types.T_int32 {
+		return nil, moerr.NewInternalErrorNoCtxf(
+			"view metadata account_id vector has type %s, expected %s",
+			column.GetType().String(), types.T_int32.String())
+	}
+	if !column.IsConst() && rows > column.Length() {
+		return nil, moerr.NewInternalErrorNoCtxf(
+			"view metadata account_id vector has %d rows, length %d",
+			rows, column.Length())
+	}
+
+	accounts := make([]uint32, rows)
+	for i := range accounts {
+		if column.IsNull(uint64(i)) {
+			return nil, moerr.NewInternalErrorNoCtxf(
+				"view metadata account_id vector contains NULL at row %d", i)
+		}
+		accountID := vector.GetFixedAtWithTypeCheck[int32](column, i)
+		if accountID < 0 {
+			return nil, moerr.NewInternalErrorNoCtxf(
+				"view metadata account_id is negative: %d", accountID)
+		}
+		accounts[i] = uint32(accountID)
+	}
+	return accounts, nil
+}
+
 func seedViewMetadataRevalidationPage(txn executor.TxnExecutor) (complete bool, active bool, err error) {
 	cursorResult, err := txn.Exec(fmt.Sprintf(
 		"select source_account_id,source_relation_kind,dependency_generation from %s.%s "+
@@ -153,11 +189,35 @@ func seedViewMetadataRevalidationPage(txn executor.TxnExecutor) (complete bool,
 		return false, true, err
 	}
 	accounts := make([]uint32, 0, viewMetadataRecoveryPageSize)
+	var pageErr error
 	page.ReadRows(func(rows int, columns []*vector.Vector) bool {
-		accounts = append(accounts, vector.MustFixedColNoTypeCheck[uint32](columns[0])[:rows]...)
+		if rows == 0 {
+			return true
+		}
+		if rows < 0 || len(accounts) > viewMetadataRecoveryPageSize ||
+			rows > viewMetadataRecoveryPageSize-len(accounts) {
+			pageErr = moerr.NewInternalErrorNoCtxf(
+				"view metadata account page has too many rows: batch=%d, read=%d, maximum=%d",
+				rows, len(accounts), viewMetadataRecoveryPageSize)
+			return false
+		}
+		if len(columns) != 1 {
+			pageErr = moerr.NewInternalErrorNoCtxf(
+				"view metadata account page returned %d columns, expected 1", len(columns))
+			return false
+		}
+		var pageAccounts []uint32
+		pageAccounts, pageErr = decodeViewMetadataAccountIDs(columns[0], rows)
+		if pageErr != nil {
+			return false
+		}
+		accounts = append(accounts, pageAccounts...)
 		return true
 	})
 	page.Close()
+	if pageErr != nil {
+		return false, true, pageErr
+	}
 	for _, accountID := range accounts {
 		result, execErr := txn.Exec(fmt.Sprintf(
 			"replace into %s.%s (%s) values (%d,0,0,0,'%s','%s',0,0,0,0,0,"+

pkg/sql/compile/view_metadata_recovery_test.go

@@ -1730,11 +1730,11 @@ func TestSeedViewMetadataRevalidationPageIsBounded(t *testing.T) {
 	require.NoError(t, executor.AppendStringRows(marker, 1,
 		[]string{catalog.ViewRefreshStatusRevalidateRequired}))
 	require.NoError(t, executor.AppendFixedRows(marker, 2, []uint64{9}))
-	accounts := executor.NewMemResult([]types.Type{types.T_uint32.ToType()}, proc.Mp())
+	accounts := executor.NewMemResult([]types.Type{types.T_int32.ToType()}, proc.Mp())
 	accounts.NewBatchWithRowCount(viewMetadataRecoveryPageSize)
-	ids := make([]uint32, viewMetadataRecoveryPageSize)
+	ids := make([]int32, viewMetadataRecoveryPageSize)
 	for i := range ids {
-		ids[i] = uint32(i + 1)
+		ids[i] = int32(i + 1)
 	}
 	require.NoError(t, executor.AppendFixedRows(accounts, 0, ids))
 	results := []executor.Result{marker.GetResult(), accounts.GetResult()}
@@ -1757,6 +1757,80 @@ func TestSeedViewMetadataRevalidationPageIsBounded(t *testing.T) {
 		fmt.Sprintf("set source_account_id=%d", viewMetadataRecoveryPageSize))
 }
 
+func TestSeedViewMetadataRevalidationPageRejectsInvalidAccountPage(t *testing.T) {
+	tests := []struct {
+		name          string
+		accountType   types.T
+		rowCount      int
+		accountIDs    []int32
+		unsignedIDs   []uint32
+		expectedError string
+	}{
+		{
+			name:          "unexpected account vector type",
+			accountType:   types.T_uint32,
+			rowCount:      1,
+			unsignedIDs:   []uint32{1},
+			expectedError: "expected INT",
+		},
+		{
+			name:          "negative account id",
+			accountType:   types.T_int32,
+			rowCount:      1,
+			accountIDs:    []int32{-1},
+			expectedError: "account_id is negative",
+		},
+		{
+			name:          "batch row count exceeds vector length",
+			accountType:   types.T_int32,
+			rowCount:      2,
+			accountIDs:    []int32{1},
+			expectedError: "rows, length",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			proc := testutil.NewProcess(t)
+			marker := executor.NewMemResult([]types.Type{
+				types.T_uint32.ToType(), types.T_varchar.ToType(), types.T_uint64.ToType(),
+			}, proc.Mp())
+			marker.NewBatchWithRowCount(1)
+			require.NoError(t, executor.AppendFixedRows(marker, 0, []uint32{0}))
+			require.NoError(t, executor.AppendStringRows(marker, 1,
+				[]string{catalog.ViewRefreshStatusRevalidateRequired}))
+			require.NoError(t, executor.AppendFixedRows(marker, 2, []uint64{9}))
+
+			accounts := executor.NewMemResult([]types.Type{tc.accountType.ToType()}, proc.Mp())
+			accounts.NewBatchWithRowCount(tc.rowCount)
+			if tc.accountType == types.T_int32 {
+				require.NoError(t, executor.AppendFixedRows(accounts, 0, tc.accountIDs))
+			} else {
+				require.NoError(t, executor.AppendFixedRows(accounts, 0, tc.unsignedIDs))
+			}
+			exec := &viewMetadataCleanupRecordingExecutor{
+				results: []executor.Result{marker.GetResult(), accounts.GetResult()},
+			}
+
+			var complete bool
+			var active bool
+			var seedErr error
+			require.NotPanics(t, func() {
+				seedErr = exec.ExecTxn(context.Background(), func(txn executor.TxnExecutor) error {
+					var err error
+					complete, active, err = seedViewMetadataRevalidationPage(txn)
+					return err
+				}, executor.Options{})
+			})
+			require.ErrorContains(t, seedErr, tc.expectedError)
+			require.False(t, complete)
+			require.True(t, active)
+			// The page is decoded before any metadata rows are written.
+			require.Len(t, exec.sqls, 2)
+		})
+	}
+}
+
 func TestViewMetadataRevalidationActivationPropagatesCatalogErrors(t *testing.T) {
 	testErr := moerr.NewInternalErrorNoCtx("catalog unavailable")