From fea15e1bb145f15bec0ae5f592a048e929ca6b73 Mon Sep 17 00:00:00 2001 From: gouhongshen Date: Fri, 28 Aug 2026 07:15:39 +0800 Subject: [PATCH] fix(view): decode signed account IDs in metadata admission --- pkg/sql/compile/view_metadata_recovery.go | 62 +++++++++++++- .../compile/view_metadata_recovery_test.go | 80 ++++++++++++++++++- 2 files changed, 138 insertions(+), 4 deletions(-) diff --git a/pkg/sql/compile/view_metadata_recovery.go b/pkg/sql/compile/view_metadata_recovery.go index 84b1cdc5d0f03..8d38637a24ce3 100644 --- a/pkg/sql/compile/view_metadata_recovery.go +++ b/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,"+ diff --git a/pkg/sql/compile/view_metadata_recovery_test.go b/pkg/sql/compile/view_metadata_recovery_test.go index fe30afaf5e910..cceb82ca9e7c9 100644 --- a/pkg/sql/compile/view_metadata_recovery_test.go +++ b/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")