From d32ae8663d0251b565b76b8b43cb0356e518e82c Mon Sep 17 00:00:00 2001 From: GreatRiver Date: Mon, 12 Oct 2026 14:48:18 +0800 Subject: [PATCH 1/3] fix(disttae): batch catalog ID lookups for table statistics --- pkg/vm/engine/disttae/cache/catalog.go | 55 +++++ .../disttae/cache/catalog_id_lookup_test.go | 210 ++++++++++++++++++ pkg/vm/engine/disttae/mo_table_stats.go | 82 +++++-- .../disttae/mo_table_stats_lookup_test.go | 166 ++++++++++++++ 4 files changed, 493 insertions(+), 20 deletions(-) create mode 100644 pkg/vm/engine/disttae/cache/catalog_id_lookup_test.go create mode 100644 pkg/vm/engine/disttae/mo_table_stats_lookup_test.go diff --git a/pkg/vm/engine/disttae/cache/catalog.go b/pkg/vm/engine/disttae/cache/catalog.go index 78d943519cf42..11457c2289afa 100644 --- a/pkg/vm/engine/disttae/cache/catalog.go +++ b/pkg/vm/engine/disttae/cache/catalog.go @@ -378,6 +378,61 @@ func (cc *CatalogCache) GetTableByIdAndTime(accountID uint32, databaseId, tblId return rel } +// GetTablesByIdAndTime resolves IDs in one traversal of the name-ordered +// catalog, using the same visibility rules as GetTableByIdAndTime. Results +// retain the input order and duplicates, with nil for IDs not visible at ts. +// A zero databaseID searches all databases in the account. +func (cc *CatalogCache) GetTablesByIdAndTime(accountID uint32, databaseID uint64, ids []uint64, ts timestamp.Timestamp) []*TableItem { + if len(ids) == 0 { + return nil + } + if len(ids) == 1 { + return []*TableItem{cc.GetTableByIdAndTime(accountID, databaseID, ids[0], ts)} + } + + wanted := make(map[uint64]*TableItem, len(ids)) + for _, id := range ids { + wanted[id] = nil + } + remaining := len(wanted) + prevTableName := "" + prevDatabaseID := databaseID + cc.tables.data.Ascend(&TableItem{AccountId: accountID, DatabaseId: databaseID}, func(item *TableItem) bool { + if item.AccountId != accountID { + return false + } + if item.DatabaseId != prevDatabaseID { + if databaseID != 0 { + return false + } + prevDatabaseID = item.DatabaseId + prevTableName = "" + } + if item.Ts.Greater(ts) { + return true + } + if item.Name == prevTableName { + return true + } + // A visible deletion or a replacement ID hides every older version + // of the same name, including IDs requested by this batch. + prevTableName = item.Name + if !item.deleted { + if found, requested := wanted[item.Id]; requested && found == nil { + wanted[item.Id] = item + remaining-- + } + } + return remaining != 0 + }) + + items := make([]*TableItem, len(ids)) + for i, id := range ids { + items[i] = wanted[id] + } + return items +} + // GetTableById scans retained catalog versions. A latest DROP tombstone hides // every older incarnation of that name, so historical rows cannot make a // dropped or truncated table appear current. diff --git a/pkg/vm/engine/disttae/cache/catalog_id_lookup_test.go b/pkg/vm/engine/disttae/cache/catalog_id_lookup_test.go new file mode 100644 index 0000000000000..eea32f6eba37c --- /dev/null +++ b/pkg/vm/engine/disttae/cache/catalog_id_lookup_test.go @@ -0,0 +1,210 @@ +// 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 cache + +import ( + "fmt" + "sync" + "testing" + + "github.com/matrixorigin/matrixone/pkg/container/types" + "github.com/matrixorigin/matrixone/pkg/pb/timestamp" + "github.com/stretchr/testify/require" +) + +func TestGetTablesByIdAndTimeSnapshotVisibility(t *testing.T) { + cc := NewCatalog() + cc.UpdateDuration(types.TS{}, types.MaxTs()) + item := func(aid uint32, db, id uint64, name string, ts int64, version uint32, deleted bool) *TableItem { + return &TableItem{AccountId: aid, DatabaseId: db, Id: id, Name: name, + Ts: timestamp.Timestamp{PhysicalTime: ts}, Version: version, deleted: deleted} + } + old := item(1, 10, 101, "a", 10, 1, false) + recreated := item(1, 10, 102, "a", 30, 1, false) + altered := item(1, 10, 102, "a", 40, 2, false) + truncated := item(1, 10, 103, "a", 50, 1, false) + b := item(1, 10, 201, "b", 10, 1, false) + view := item(1, 10, 202, "view", 20, 1, false) + view.Kind = "v" + otherDB := item(1, 20, 301, "a", 10, 1, false) + otherTenant := item(2, 10, 401, "a", 10, 1, false) + for _, value := range []*TableItem{ + old, item(1, 10, 101, "a", 20, 0, true), recreated, + item(1, 10, 102, "a", 40, 0, true), altered, + item(1, 10, 102, "a", 50, 0, true), truncated, + b, item(1, 10, 201, "b", 60, 0, true), view, otherDB, otherTenant, + } { + cc.setTableItem(value, false) + } + ids := []uint64{103, 101, 999, 102, 301, 201, 201, 202, 401} + for _, tc := range []struct { + name string + aid uint32 + db uint64 + ts int64 + want []*TableItem + }{ + {"before-create", 1, 10, 9, []*TableItem{nil, nil, nil, nil, nil, nil, nil, nil, nil}}, + {"created", 1, 10, 10, []*TableItem{nil, old, nil, nil, nil, b, b, nil, nil}}, + {"dropped", 1, 10, 20, []*TableItem{nil, nil, nil, nil, nil, b, b, view, nil}}, + {"recreated", 1, 10, 30, []*TableItem{nil, nil, nil, recreated, nil, b, b, view, nil}}, + {"same-timestamp-alter", 1, 10, 40, []*TableItem{nil, nil, nil, altered, nil, b, b, view, nil}}, + {"same-timestamp-truncate", 1, 10, 50, []*TableItem{truncated, nil, nil, nil, nil, b, b, view, nil}}, + {"latest-drop", 1, 10, 60, []*TableItem{truncated, nil, nil, nil, nil, nil, nil, view, nil}}, + {"wildcard-database", 1, 0, 40, []*TableItem{nil, nil, nil, altered, otherDB, b, b, view, nil}}, + {"other-database", 1, 20, 40, []*TableItem{nil, nil, nil, nil, otherDB, nil, nil, nil, nil}}, + {"other-tenant", 2, 10, 40, []*TableItem{nil, nil, nil, nil, nil, nil, nil, nil, otherTenant}}, + {"absent-scope", 3, 0, 40, []*TableItem{nil, nil, nil, nil, nil, nil, nil, nil, nil}}, + } { + t.Run(tc.name, func(t *testing.T) { + ts := timestamp.Timestamp{PhysicalTime: tc.ts} + got := cc.GetTablesByIdAndTime(tc.aid, tc.db, ids, ts) + require.Len(t, got, len(ids)) + for i := range got { + require.Same(t, tc.want[i], got[i], "id=%d", ids[i]) + require.Same(t, cc.GetTableByIdAndTime(tc.aid, tc.db, ids[i], ts), got[i]) + } + }) + } + t.Run("empty-and-singleton", func(t *testing.T) { + ts := timestamp.Timestamp{PhysicalTime: 40} + require.Nil(t, cc.GetTablesByIdAndTime(1, 10, nil, ts)) + require.Nil(t, cc.GetTablesByIdAndTime(1, 10, []uint64{}, ts)) + require.Equal(t, []*TableItem{altered}, cc.GetTablesByIdAndTime(1, 10, []uint64{102}, ts)) + require.Equal(t, []*TableItem{nil}, cc.GetTablesByIdAndTime(1, 10, []uint64{101}, ts)) + require.Equal(t, []*TableItem{altered, altered}, cc.GetTablesByIdAndTime(1, 10, []uint64{102, 102}, ts)) + }) + t.Run("GC-retained-boundary", func(t *testing.T) { + cc.GC(timestamp.Timestamp{PhysicalTime: 45}) + require.False(t, cc.CanServe(types.BuildTS(40, 0))) + require.True(t, cc.CanServe(types.BuildTS(45, 0))) + for _, ts := range []int64{45, 50, 60} { + stamp := timestamp.Timestamp{PhysicalTime: ts} + got := cc.GetTablesByIdAndTime(1, 0, []uint64{101, 102, 103, 201, 202, 301}, stamp) + want := []*TableItem{nil, altered, nil, b, view, otherDB} + if ts >= 50 { + want[1], want[2] = nil, truncated + } + if ts >= 60 { + want[3] = nil + } + for i := range got { + require.Same(t, want[i], got[i], "ts=%d index=%d", ts, i) + } + } + }) +} + +func TestGetTablesByIdAndTimeConcurrentWriterAndGC(t *testing.T) { + cc := NewCatalog() + cc.UpdateDuration(types.TS{}, types.MaxTs()) + old := &TableItem{AccountId: 1, DatabaseId: 10, Id: 101, Name: "a", Ts: timestamp.Timestamp{PhysicalTime: 10}} + visible := &TableItem{AccountId: 1, DatabaseId: 10, Id: 102, Name: "a", Ts: timestamp.Timestamp{PhysicalTime: 40}, Version: 2} + for _, item := range []*TableItem{ + old, + {AccountId: 1, DatabaseId: 10, Id: 101, Name: "a", Ts: timestamp.Timestamp{PhysicalTime: 30}, deleted: true}, + {AccountId: 1, DatabaseId: 10, Id: 102, Name: "a", Ts: timestamp.Timestamp{PhysicalTime: 30}}, + visible, + } { + cc.setTableItem(item, false) + } + ids := []uint64{101, 102, 103, 999, 102} + want := []*TableItem{nil, visible, nil, nil, visible} + stamp := timestamp.Timestamp{PhysicalTime: 50} + start := make(chan struct{}) + ready := make(chan struct{}, 3) + var wg sync.WaitGroup + wg.Add(3) + go func() { + defer wg.Done() + ready <- struct{}{} + <-start + for i := int64(60); i < 124; i++ { + cc.setTableItem(&TableItem{AccountId: 1, DatabaseId: 10, Id: 103, Name: "a", Ts: timestamp.Timestamp{PhysicalTime: i}}, false) + } + }() + go func() { + defer wg.Done() + ready <- struct{}{} + <-start + for range 16 { + cc.GC(stamp) + } + }() + go func() { + defer wg.Done() + ready <- struct{}{} + <-start + for range 256 { + got := cc.GetTablesByIdAndTime(1, 10, ids, stamp) + for i := range want { + if got[i] != want[i] { + t.Errorf("retained snapshot id=%d: got %p, want %p", ids[i], got[i], want[i]) + return + } + } + } + }() + for range 3 { + <-ready + } + close(start) + wg.Wait() + require.True(t, cc.CanServe(types.BuildTS(50, 0))) + require.Equal(t, want, cc.GetTablesByIdAndTime(1, 10, ids, stamp)) +} + +var tableIDLookupBenchmarkSink []*TableItem + +func BenchmarkGetTablesByIdAndTime(b *testing.B) { + for _, names := range []int{64, 4096} { + cc := NewCatalog() + for i := 0; i < names; i++ { + for v := int64(1); v <= 4; v++ { + cc.setTableItem(&TableItem{AccountId: 1, DatabaseId: 10, Name: fmt.Sprintf("t%08d", i), + Id: uint64(i + 1000), Ts: timestamp.Timestamp{PhysicalTime: v}}, false) + } + } + for _, asked := range []int{1, 64} { + ids := make([]uint64, asked) + for i := range ids { + ids[i] = uint64(1<<40) + uint64(i) + } + for _, mode := range []string{"scalar", "batch"} { + b.Run(fmt.Sprintf("names=%d/asked=%d/%s", names, asked, mode), func(b *testing.B) { + stamp := timestamp.Timestamp{PhysicalTime: 100} + b.ReportAllocs() + for i := 0; i < b.N; i++ { + if mode == "batch" { + tableIDLookupBenchmarkSink = cc.GetTablesByIdAndTime(1, 10, ids, stamp) + } else { + out := make([]*TableItem, len(ids)) + for j, id := range ids { + out[j] = cc.GetTableByIdAndTime(1, 10, id, stamp) + } + tableIDLookupBenchmarkSink = out + } + } + b.StopTimer() + for _, item := range tableIDLookupBenchmarkSink { + if item != nil { + b.Fatal("missing ID unexpectedly found") + } + } + }) + } + } + } +} diff --git a/pkg/vm/engine/disttae/mo_table_stats.go b/pkg/vm/engine/disttae/mo_table_stats.go index 9c89869964a12..8dbf6ac58895c 100644 --- a/pkg/vm/engine/disttae/mo_table_stats.go +++ b/pkg/vm/engine/disttae/mo_table_stats.go @@ -50,6 +50,7 @@ import ( v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2" "github.com/matrixorigin/matrixone/pkg/vm/engine" "github.com/matrixorigin/matrixone/pkg/vm/engine/cmd_util" + "github.com/matrixorigin/matrixone/pkg/vm/engine/disttae/cache" "github.com/matrixorigin/matrixone/pkg/vm/engine/disttae/logtailreplay" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/options" "github.com/matrixorigin/matrixone/pkg/vm/process" @@ -1567,6 +1568,15 @@ func buildTablePairFromCache( ) (tbl tablePair, ok bool) { item := eng.(*Engine).GetLatestCatalogCache().GetTableByIdAndTime(uint32(accId), dbId, tblId, snapshot.ToTimestamp()) + return buildTablePairFromItem(item, accId, dbId, tblId, snapshot, onlyUpdateTS) +} + +func buildTablePairFromItem( + item *cache.TableItem, + accId, dbId, tblId uint64, + snapshot types.TS, + onlyUpdateTS bool, +) (tbl tablePair, ok bool) { if item == nil || item.IsDeleted() { // account, db, tbl may delete already // the `update_time` not change anymore @@ -1591,6 +1601,56 @@ func buildTablePairFromCache( return tbl, true } +func buildTablePairsFromCache( + eng engine.Engine, + accs, dbs, tbls []uint64, + snapshot types.TS, + onlyUpdateTS bool, +) []tablePair { + if len(tbls) == 0 { + return nil + } + if len(tbls) == 1 { + if tbl, ok := buildTablePairFromCache(eng, accs[0], dbs[0], tbls[0], snapshot, onlyUpdateTS); ok { + return []tablePair{tbl} + } + return nil + } + + // ID lookup scans a name-ordered BTree. Group requests by catalog scope + // so missing and dropped IDs do not each rescan all retained versions. + groups := make(map[[2]uint64][]int) + for i := range tbls { + key := [2]uint64{accs[i], dbs[i]} + groups[key] = append(groups[key], i) + } + cc := eng.(*Engine).GetLatestCatalogCache() + ts := snapshot.ToTimestamp() + items := make([]*cache.TableItem, len(tbls)) + for key, indices := range groups { + if len(indices) == 1 { + i := indices[0] + items[i] = cc.GetTableByIdAndTime(uint32(key[0]), key[1], tbls[i], ts) + continue + } + ids := make([]uint64, len(indices)) + for i, index := range indices { + ids[i] = tbls[index] + } + for i, item := range cc.GetTablesByIdAndTime(uint32(key[0]), key[1], ids, ts) { + items[indices[i]] = item + } + } + + var pairs []tablePair + for i, item := range items { + if tbl, ok := buildTablePairFromItem(item, accs[i], dbs[i], tbls[i], snapshot, onlyUpdateTS); ok { + pairs = append(pairs, tbl) + } + } + return pairs +} + func (tp *tablePair) String() string { return fmt.Sprintf( "%d-%s(%d)-%s(%d)-valid(%v)-kind(%v)-alphaOrder(%d)-onlyTS(%v)", @@ -3080,29 +3140,11 @@ func getChangedTableList( return } - var ( - tp tablePair - ok bool - ) - - for i := range nullTSAcc { - if tp, ok = buildTablePairFromCache(de, - nullTSAcc[i], nullTSDB[i], nullTSTbl[i], *to, false); !ok { - continue - } - *pairs = append(*pairs, tp) - } + *pairs = append(*pairs, buildTablePairsFromCache(de, nullTSAcc, nullTSDB, nullTSTbl, *to, false)...) // a table in the response means there exists updates, // need calculate the size/rows... - for i := range resp.AccIds { - if tp, ok = buildTablePairFromCache(de, - resp.AccIds[i], resp.DatabaseIds[i], resp.TableIds[i], *to, false); !ok { - continue - } - - *pairs = append(*pairs, tp) - } + *pairs = append(*pairs, buildTablePairsFromCache(de, resp.AccIds, resp.DatabaseIds, resp.TableIds, *to, false)...) // for the CheckChanged type, a table not in the response // means no update yet, update it's ts only. diff --git a/pkg/vm/engine/disttae/mo_table_stats_lookup_test.go b/pkg/vm/engine/disttae/mo_table_stats_lookup_test.go new file mode 100644 index 0000000000000..bb3320b0dd083 --- /dev/null +++ b/pkg/vm/engine/disttae/mo_table_stats_lookup_test.go @@ -0,0 +1,166 @@ +// 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 disttae + +import ( + "context" + "strconv" + "testing" + + "github.com/matrixorigin/matrixone/pkg/catalog" + "github.com/matrixorigin/matrixone/pkg/common/mpool" + "github.com/matrixorigin/matrixone/pkg/common/runtime" + "github.com/matrixorigin/matrixone/pkg/container/types" + "github.com/matrixorigin/matrixone/pkg/container/vector" + "github.com/matrixorigin/matrixone/pkg/pb/timestamp" + "github.com/matrixorigin/matrixone/pkg/pb/txn" + "github.com/matrixorigin/matrixone/pkg/txn/client" + "github.com/matrixorigin/matrixone/pkg/txn/rpc" + "github.com/matrixorigin/matrixone/pkg/vm/engine/cmd_util" + "github.com/matrixorigin/matrixone/pkg/vm/engine/disttae/cache" + "github.com/stretchr/testify/require" +) + +func newStatsLookupEngine(t *testing.T) *Engine { + t.Helper() + mp := mpool.MustNewZero() + t.Cleanup(func() { require.Zero(t, mp.CurrNB()) }) + eng := &Engine{mp: mp} + eng.catalog.Store(cache.NewCatalog()) + packer := types.NewPacker() + defer packer.Close() + for _, tbl := range []catalog.Table{ + {AccountId: 1, DatabaseId: 10, TableId: 101, DatabaseName: "db1", TableName: "a", Kind: "r"}, + {AccountId: 1, DatabaseId: 10, TableId: 102, DatabaseName: "db1", TableName: "v", Kind: "v"}, + {AccountId: 1, DatabaseId: 20, TableId: 201, DatabaseName: "db2", TableName: "a", Kind: "r"}, + {AccountId: 2, DatabaseId: 10, TableId: 301, DatabaseName: "tenant2", TableName: "a", Kind: "r"}, + } { + bat, err := catalog.GenCreateTableTuple(tbl, mp, packer) + require.NoError(t, err) + _, err = fillRandomRowidAndZeroTs(bat, mp) + require.NoError(t, err) + vector.MustFixedColWithTypeCheck[types.TS](bat.GetVector(cache.MO_TIMESTAMP_IDX))[0] = types.BuildTS(1, 0) + eng.GetLatestCatalogCache().InsertTable(bat) + bat.Clean(mp) + eng.GetLatestCatalogCache().GetTableById(tbl.AccountId, tbl.DatabaseId, tbl.TableId).PrimarySeqnum = 7 + } + return eng +} + +func TestBuildTablePairsFromCachePreservesOrderAndMetadata(t *testing.T) { + eng := newStatsLookupEngine(t) + accs := []uint64{2, 1, 1, 1, 1, 1, 1} + dbs := []uint64{10, 20, 10, 10, 10, 0, 10} + ids := []uint64{301, 201, 999, 102, 101, 201, 101} + snapshot := types.BuildTS(10, 0) + want := []tablePair{ + {valid: true, acc: 2, db: 10, tbl: 301, dbName: "tenant2", tblName: "a", relKind: "r", pkSequence: 7, snapshot: snapshot, onlyUpdateTS: true}, + {valid: true, acc: 1, db: 20, tbl: 201, dbName: "db2", tblName: "a", relKind: "r", pkSequence: 7, snapshot: snapshot, onlyUpdateTS: true}, + {valid: true, acc: 1, db: 10, tbl: 101, dbName: "db1", tblName: "a", relKind: "r", pkSequence: 7, snapshot: snapshot, onlyUpdateTS: true}, + {valid: true, acc: 1, db: 0, tbl: 201, dbName: "db2", tblName: "a", relKind: "r", pkSequence: 7, snapshot: snapshot, onlyUpdateTS: true}, + {valid: true, acc: 1, db: 10, tbl: 101, dbName: "db1", tblName: "a", relKind: "r", pkSequence: 7, snapshot: snapshot, onlyUpdateTS: true}, + } + require.Equal(t, want, buildTablePairsFromCache(eng, accs, dbs, ids, snapshot, true)) + require.Nil(t, buildTablePairsFromCache(eng, nil, nil, nil, snapshot, false)) + for _, id := range []uint64{101, 102, 999} { + t.Run("singleton-"+strconv.FormatUint(id, 10), func(t *testing.T) { + got := buildTablePairsFromCache(eng, []uint64{1}, []uint64{10}, []uint64{id}, snapshot, true) + if id == 101 { + require.Equal(t, []tablePair{want[2]}, got) + } else { + require.Empty(t, got) + } + }) + } + require.Empty(t, buildTablePairsFromCache(eng, []uint64{1, 1}, []uint64{10, 10}, []uint64{102, 999}, snapshot, false)) +} + +func TestGetChangedTableListBatchLookupConsumer(t *testing.T) { + eng := newStatsLookupEngine(t) + waiter := client.NewTimestampWaiter(runtime.DefaultRuntime().Logger()) + defer waiter.Close() + waiter.NotifyLatestCommitTS(timestamp.Timestamp{PhysicalTime: 100}) + client.RunTxnTests(func(cli client.TxnClient, _ rpc.TxnSender) { + eng.cli = cli + for _, typ := range []cmd_util.ChangedListType{cmd_util.CheckChanged, cmd_util.CollectChanged} { + t.Run(typ.String(), func(t *testing.T) { + var pairs []tablePair + var to, oldest types.TS + calls := 0 + from := timestamp.Timestamp{PhysicalTime: 10} + latest := timestamp.Timestamp{PhysicalTime: 20} + stamps := []timestamp.Timestamp{{}, from, {}, latest} + if typ == cmd_util.CollectChanged { + stamps = []timestamp.Timestamp{from, latest} + } + err := GetChangedTableList(t.Context(), "", eng, + []uint64{1, 1, 1, 1}, []uint64{10, 10, 10, 10}, []uint64{101, 999, 102, 101}, stamps, + &to, &oldest, typ, + func(aid, db, id int64, name, dbName, kind string, sequence int, snapshot types.TS) { + pairs = append(pairs, tablePair{acc: aid, db: db, tbl: id, tblName: name, + dbName: dbName, relKind: kind, pkSequence: sequence, snapshot: snapshot}) + }, + func(_ context.Context, _ txn.TxnMeta, req *cmd_util.GetChangedTableListReq, resp *cmd_util.GetChangedTableListResp) (func(), error) { + calls++ + require.Equal(t, typ, req.Type) + if typ == cmd_util.CheckChanged { + require.Equal(t, []uint64{999, 101}, req.TableIds) + require.Equal(t, []uint64{1, 1}, req.AccIds) + require.Equal(t, []uint64{10, 10}, req.DatabaseIds) + } else { + require.Empty(t, req.TableIds) + } + require.Equal(t, []*timestamp.Timestamp{&from, &latest}, req.TS) + resp.AccIds = []uint64{1, 1, 1, 1, 2, 1} + resp.DatabaseIds = []uint64{20, 10, 10, 10, 10, 10} + resp.TableIds = []uint64{201, 101, 999, 101, 301, 102} + resp.Oldest = &from + return nil, nil + }) + require.NoError(t, err) + require.Equal(t, 1, calls) + require.Equal(t, types.TimestampToTS(from), oldest) + require.Equal(t, types.TimestampToTS(timestamp.Timestamp{PhysicalTime: 100}.Next()), to) + wantIDs := []int64{201, 101, 101, 301} + if typ == cmd_util.CheckChanged { + wantIDs = append([]int64{101}, wantIDs...) + } + var gotIDs []int64 + for _, pair := range pairs { + gotIDs = append(gotIDs, pair.tbl) + require.Equal(t, to, pair.snapshot) + require.Equal(t, "r", pair.relKind) + require.Equal(t, 7, pair.pkSequence) + require.Equal(t, "a", pair.tblName) + switch pair.tbl { + case 101: + require.Equal(t, int64(1), pair.acc) + require.Equal(t, int64(10), pair.db) + require.Equal(t, "db1", pair.dbName) + case 201: + require.Equal(t, int64(1), pair.acc) + require.Equal(t, int64(20), pair.db) + require.Equal(t, "db2", pair.dbName) + case 301: + require.Equal(t, int64(2), pair.acc) + require.Equal(t, int64(10), pair.db) + require.Equal(t, "tenant2", pair.dbName) + } + } + require.Equal(t, wantIDs, gotIDs) + }) + } + }, client.WithTimestampWaiter(waiter), client.WithEnableSacrificingFreshness()) +} From 54d8f0080425d5a0aaf45a668d307b68950dbce5 Mon Sep 17 00:00:00 2001 From: GreatRiver Date: Mon, 12 Oct 2026 15:43:25 +0800 Subject: [PATCH 2/3] test(disttae): preallocate changed-table result IDs --- pkg/vm/engine/disttae/mo_table_stats_lookup_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/vm/engine/disttae/mo_table_stats_lookup_test.go b/pkg/vm/engine/disttae/mo_table_stats_lookup_test.go index bb3320b0dd083..6516ae96aa5e3 100644 --- a/pkg/vm/engine/disttae/mo_table_stats_lookup_test.go +++ b/pkg/vm/engine/disttae/mo_table_stats_lookup_test.go @@ -137,7 +137,7 @@ func TestGetChangedTableListBatchLookupConsumer(t *testing.T) { if typ == cmd_util.CheckChanged { wantIDs = append([]int64{101}, wantIDs...) } - var gotIDs []int64 + gotIDs := make([]int64, 0, len(pairs)) for _, pair := range pairs { gotIDs = append(gotIDs, pair.tbl) require.Equal(t, to, pair.snapshot) From 0da8dbfd612321826dfa4080f82db52c70a91d53 Mon Sep 17 00:00:00 2001 From: GreatRiver Date: Mon, 12 Oct 2026 18:19:02 +0800 Subject: [PATCH 3/3] test(disttae): unregister statistics lookup fixture pools --- pkg/vm/engine/disttae/mo_table_stats_lookup_test.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pkg/vm/engine/disttae/mo_table_stats_lookup_test.go b/pkg/vm/engine/disttae/mo_table_stats_lookup_test.go index 6516ae96aa5e3..afd0625f58d54 100644 --- a/pkg/vm/engine/disttae/mo_table_stats_lookup_test.go +++ b/pkg/vm/engine/disttae/mo_table_stats_lookup_test.go @@ -36,7 +36,10 @@ import ( func newStatsLookupEngine(t *testing.T) *Engine { t.Helper() mp := mpool.MustNewZero() - t.Cleanup(func() { require.Zero(t, mp.CurrNB()) }) + t.Cleanup(func() { + defer mpool.DeleteMPool(mp) + require.Zero(t, mp.CurrNB()) + }) eng := &Engine{mp: mp} eng.catalog.Store(cache.NewCatalog()) packer := types.NewPacker()