From 1280c6725d52caeb4af00ef0a5758bb26af8b739 Mon Sep 17 00:00:00 2001 From: iamlinjunhong <1030420200@qq.com> Date: Fri, 31 Jul 2026 15:31:09 +0800 Subject: [PATCH 1/6] fix: resolve MongoDB external scan follow-ups Harden catalog typing, temporal conversion, and predicate pushdown. Bound scan and max_by memory work, share client dials, retire clients asynchronously, and align restore plus local E2E lifecycle behavior. Fixes #26485 --- etc/launch-mongodb-local/compose.yaml | 2 +- optools/mongodb_ci.bash | 41 ++-- pkg/cnservice/server.go | 24 +- pkg/frontend/mongodb.go | 10 +- pkg/frontend/server.go | 6 +- pkg/frontend/snapshot.go | 8 + pkg/frontend/snapshot_test.go | 10 + pkg/sql/colexec/aggexec/maxby.go | 115 ++++++++-- pkg/sql/colexec/aggexec/maxby_test.go | 101 +++++++++ pkg/sql/colexec/mongoscan/mongoscan.go | 9 +- pkg/sql/colexec/mongoscan/mongoscan_test.go | 37 +++ pkg/sql/compile/ddl.go | 17 +- pkg/sql/features/table_feature.go | 11 +- pkg/sql/mongodb/converter.go | 93 +++++++- pkg/sql/mongodb/envelope.go | 35 ++- pkg/sql/mongodb/mongodb_test.go | 236 ++++++++++++++++++++ pkg/sql/mongodb/plan_predicate.go | 13 +- pkg/sql/mongodb/pool.go | 82 +++++-- pkg/sql/mongodb/retirement.go | 85 +++++++ pkg/sql/mongodb/retirement_test.go | 77 +++++++ pkg/sql/plan/build_ddl.go | 4 + pkg/sql/plan/build_ddl_test.go | 7 +- pkg/sql/plan/build_show_util.go | 10 +- pkg/sql/plan/build_show_util_test.go | 6 +- pkg/sql/plan/deepcopy_mongodb_test.go | 34 +++ pkg/sql/plan/mongodb_util.go | 17 +- pkg/sql/plan/query_builder.go | 21 +- pkg/sql/plan/query_builder_test.go | 4 +- test/mongodb/mongodb_e2e_local.go | 84 ++++++- 29 files changed, 1090 insertions(+), 109 deletions(-) diff --git a/etc/launch-mongodb-local/compose.yaml b/etc/launch-mongodb-local/compose.yaml index 402cde072ebe1..d686b8f2f9a7e 100644 --- a/etc/launch-mongodb-local/compose.yaml +++ b/etc/launch-mongodb-local/compose.yaml @@ -9,7 +9,7 @@ services: entrypoint: ["bash", "-c", "cp /run/key-source /tmp/mongodb-keyfile && chown mongodb:mongodb /tmp/mongodb-keyfile && chmod 400 /tmp/mongodb-keyfile && exec /usr/local/bin/docker-entrypoint.sh \"$$@\"", "--"] command: ["mongod", "--replSet", "rs0", "--bind_ip_all", "--auth", "--keyFile", "/tmp/mongodb-keyfile"] ports: - - "127.0.0.1:${MONGODB_PORT}:27017" + - "127.0.0.1:${MONGODB_PORT:-}:27017" environment: MONGO_INITDB_ROOT_USERNAME: ${MONGODB_ROOT_USER} MONGO_INITDB_ROOT_PASSWORD: ${MONGODB_ROOT_PASSWORD} diff --git a/optools/mongodb_ci.bash b/optools/mongodb_ci.bash index 6c6b94eccb607..b8a9549100174 100755 --- a/optools/mongodb_ci.bash +++ b/optools/mongodb_ci.bash @@ -71,16 +71,6 @@ cleanup() { return "$status" } -free_port() { - python3 - <<'PY' -import socket -s = socket.socket() -s.bind(("127.0.0.1", 0)) -print(s.getsockname()[1]) -s.close() -PY -} - wait_mongo() { local deadline=$((SECONDS + 120)) until docker compose -p "$COMPOSE_PROJECT_NAME" -f "$ROOT_DIR/etc/launch-mongodb-local/compose.yaml" exec -T mongo \ @@ -91,6 +81,22 @@ wait_mongo() { done } +wait_mo_port() { + local deadline=$((SECONDS + 120)) port="" + while (( SECONDS < deadline )); do + if [[ -n "$MO_PID" ]] && ! kill -0 "$MO_PID" >/dev/null 2>&1; then + die "MatrixOne exited before publishing its frontend listener" + fi + port="$(sed -nE 's/.*Server Listening on : [^ ]*:([0-9]+).*/\1/p' "$TMP_DIR/mo-service.log" | tail -1)" + if [[ "$port" =~ ^[1-9][0-9]*$ ]]; then + printf '%s\n' "$port" + return + fi + sleep 0.25 + done + die "MatrixOne did not publish its frontend listener" +} + wait_primary() { local deadline=$((SECONDS + 120)) until docker compose -p "$COMPOSE_PROJECT_NAME" -f "$ROOT_DIR/etc/launch-mongodb-local/compose.yaml" exec -T mongo \ @@ -138,7 +144,10 @@ run_e2e() { TMP_DIR="$(mktemp -d "${TMPDIR:-/tmp}/mo-mongodb-e2e.XXXXXX")" trap cleanup EXIT export COMPOSE_PROJECT_NAME="mo-mongodb-$(basename "$TMP_DIR" | tr '[:upper:].' '[:lower:]-')" - export MONGODB_PORT="$(free_port)" MO_PORT="$(free_port)" + # Let Docker and MatrixOne bind port 0 themselves. The selected listeners + # stay owned from allocation through use, eliminating the bind-close-rebind + # window that let adjacent CI jobs steal either port. + export MONGODB_PORT="" MO_PORT="0" export MONGODB_ROOT_USER="root_$(openssl rand -hex 6)" export MONGODB_ROOT_PASSWORD="$(openssl rand -hex 24)" export MONGODB_READER_PASSWORD="$(openssl rand -hex 24)" @@ -147,7 +156,12 @@ run_e2e() { openssl rand -base64 756 >"$MONGODB_KEYFILE" chmod 600 "$MONGODB_KEYFILE" + (cd "$ROOT_DIR" && make build) + generate_mo_config docker compose -p "$COMPOSE_PROJECT_NAME" -f "$ROOT_DIR/etc/launch-mongodb-local/compose.yaml" up -d + MONGODB_PORT="$(docker compose -p "$COMPOSE_PROJECT_NAME" -f "$ROOT_DIR/etc/launch-mongodb-local/compose.yaml" port mongo 27017 | sed -nE 's/.*:([0-9]+)$/\1/p' | tail -1)" + [[ "$MONGODB_PORT" =~ ^[1-9][0-9]*$ ]] || die "Docker did not publish the MongoDB listener" + export MONGODB_PORT wait_mongo docker compose -p "$COMPOSE_PROJECT_NAME" -f "$ROOT_DIR/etc/launch-mongodb-local/compose.yaml" exec -T mongo \ mongosh --quiet -u "$MONGODB_ROOT_USER" -p "$MONGODB_ROOT_PASSWORD" --authenticationDatabase admin \ @@ -158,9 +172,6 @@ run_e2e() { -e MONGODB_READER_NEXT_PASSWORD="$MONGODB_READER_NEXT_PASSWORD" mongo \ mongosh --quiet -u "$MONGODB_ROOT_USER" -p "$MONGODB_ROOT_PASSWORD" --authenticationDatabase admin \ <"$ROOT_DIR/etc/launch-mongodb-local/init_and_seed.js" >/dev/null - - (cd "$ROOT_DIR" && make build) - generate_mo_config export MO_MONGODB_E2E_CREDENTIAL="{\"Username\":\"mo_reader\",\"Password\":\"$MONGODB_READER_PASSWORD\"}" export MO_MONGODB_E2E_CREDENTIAL_NEXT="{\"Username\":\"mo_reader_next\",\"Password\":\"$MONGODB_READER_NEXT_PASSWORD\"}" if [[ "$(uname -s)" == Darwin ]]; then @@ -170,6 +181,8 @@ run_e2e() { fi "$ROOT_DIR/mo-service" -launch "$TMP_DIR/mo-config/launch.toml" >"$TMP_DIR/mo-service.log" 2>&1 & MO_PID=$! + MO_PORT="$(wait_mo_port)" + export MO_PORT (cd "$ROOT_DIR" && go run ./test/mongodb/mongodb_e2e_local.go \ --dsn "root:111@tcp(127.0.0.1:$MO_PORT)/?timeout=5s&readTimeout=30s&writeTimeout=30s" \ --mongo-host "127.0.0.1:$MONGODB_PORT" --report-dir "$REPORT_DIR") diff --git a/pkg/cnservice/server.go b/pkg/cnservice/server.go index 70825aa34d004..43f9ecce0ff22 100644 --- a/pkg/cnservice/server.go +++ b/pkg/cnservice/server.go @@ -1168,17 +1168,23 @@ func (s *service) initMongoDBRuntime() { MaxConversionErrors: parameters.MaxConversionErrors, MaxConversionErrorRate: parameters.MaxConversionErrorRate, MaxSourceConcurrency: parameters.MaxSourceConcurrency, } + pool := sqlmongodb.NewValidatedClientPool( + sqlmongodb.OfficialClientFactory{}, + sqlmongodb.CatalogConnectionResolver{Executor: s.sqlExecutor}, + config.MaxCachedClients, + ) + retirements := sqlmongodb.NewClientRetirementQueue( + pool, + sqlmongodb.ClusterRemoteClientRetirer{Cluster: s.moCluster, QueryClient: s.queryClient}, + sqlmongodb.DefaultClientRetirementQueueCapacity, + ) dependencies := &sqlmongodb.RuntimeDependencies{ Config: config, Connections: sqlmongodb.CatalogConnectionResolver{Executor: s.sqlExecutor}, Mappings: sqlmongodb.CatalogMappingResolver{Executor: s.sqlExecutor}, Secrets: sqlmongodb.EnvSecretResolver{}, - Pool: sqlmongodb.NewValidatedClientPool( - sqlmongodb.OfficialClientFactory{}, - sqlmongodb.CatalogConnectionResolver{Executor: s.sqlExecutor}, - config.MaxCachedClients, - ), - Limiter: sqlmongodb.NewSourceLimiter(config.MaxSourceConcurrency), + Pool: pool, Limiter: sqlmongodb.NewSourceLimiter(config.MaxSourceConcurrency), + Retirements: retirements, } runtime.ServiceRuntime(s.cfg.UUID).SetGlobalVariables(sqlmongodb.RuntimeDependenciesKey, dependencies) } @@ -1195,7 +1201,11 @@ func (s *service) closeMongoDBRuntime() error { } ctx, cancel := context.WithTimeoutCause(context.Background(), 10*time.Second, moerr.CauseShutdown) defer cancel() - return dependencies.Pool.Close(ctx) + var err error + if dependencies.Retirements != nil { + err = dependencies.Retirements.Close(ctx) + } + return errors.Join(err, dependencies.Pool.Close(ctx)) } func (s *service) initIncrService() { diff --git a/pkg/frontend/mongodb.go b/pkg/frontend/mongodb.go index 8ccf6bc7f97de..7f63c96aaa533 100644 --- a/pkg/frontend/mongodb.go +++ b/pkg/frontend/mongodb.go @@ -177,8 +177,14 @@ func handleDropMongoDBConnection(ctx context.Context, ses *Session, stmt *tree.D } func retireMongoDBClients(ctx context.Context, service string, retirement mongodb.ClientRetirement) { - if dependencies := mongoDBRuntimeDependencies(service); dependencies != nil && dependencies.Pool != nil { - _ = retirement.Apply(dependencies.Pool) + if dependencies := mongoDBRuntimeDependencies(service); dependencies != nil { + if dependencies.Retirements != nil { + dependencies.Retirements.Submit(retirement) + return + } + if dependencies.Pool != nil { + _ = retirement.Apply(dependencies.Pool) + } } pu := getPuIfPresent(service) if pu == nil || pu.QueryClient == nil { diff --git a/pkg/frontend/server.go b/pkg/frontend/server.go index 2d0df9f16ddb8..6309da5191f8d 100644 --- a/pkg/frontend/server.go +++ b/pkg/frontend/server.go @@ -112,7 +112,11 @@ func (mo *MOServer) GetRoutineManager() *RoutineManager { } func (mo *MOServer) Start() error { - logutil.Infof("Server Listening on : %s ", mo.addr) + address := mo.addr + if len(mo.listeners) > 0 && mo.listeners[0] != nil { + address = mo.listeners[0].Addr().String() + } + logutil.Infof("Server Listening on : %s ", address) mo.running = true mo.startTempTableGC(24 * time.Hour) mo.startConnectionLivenessMonitor() diff --git a/pkg/frontend/snapshot.go b/pkg/frontend/snapshot.go index a4f7f9a0d80a2..b38068d1588b7 100644 --- a/pkg/frontend/snapshot.go +++ b/pkg/frontend/snapshot.go @@ -34,6 +34,7 @@ import ( indexplugin "github.com/matrixorigin/matrixone/pkg/indexplugin" pbplan "github.com/matrixorigin/matrixone/pkg/pb/plan" "github.com/matrixorigin/matrixone/pkg/pb/timestamp" + sqlmongodb "github.com/matrixorigin/matrixone/pkg/sql/mongodb" "github.com/matrixorigin/matrixone/pkg/sql/parsers" "github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect" "github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect/mysql" @@ -132,6 +133,13 @@ var ( catalog.MOPartitionMetadata: 1, catalog.MOPartitionTables: 1, + + // MongoDB external tables are deliberately skipped by bulk restore. + // Their table-ID keyed mappings must follow the same policy; cloning a + // historical row without its external table creates an orphan that can + // permanently block DROP MONGODB CONNECTION. Snapshot and PITR share this + // system-table policy. + sqlmongodb.TableMappings: 1, } ) diff --git a/pkg/frontend/snapshot_test.go b/pkg/frontend/snapshot_test.go index c17ef16c33d05..a2c6d40034ff4 100644 --- a/pkg/frontend/snapshot_test.go +++ b/pkg/frontend/snapshot_test.go @@ -36,6 +36,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/defines" mock_frontend "github.com/matrixorigin/matrixone/pkg/frontend/test" "github.com/matrixorigin/matrixone/pkg/pb/txn" + sqlmongodb "github.com/matrixorigin/matrixone/pkg/sql/mongodb" "github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect" "github.com/matrixorigin/matrixone/pkg/sql/parsers/tree" "github.com/matrixorigin/matrixone/pkg/sql/plan" @@ -83,6 +84,15 @@ func TestGetFkDepsFromTableInfos(t *testing.T) { require.NotContains(t, deps, genKey("d", "v")) } +func TestMongoDBMappingsFollowExternalTableRestoreSkipPolicy(t *testing.T) { + info := &tableInfo{dbName: moCatalog, tblName: sqlmongodb.TableMappings, typ: "BASE TABLE"} + for _, accountID := range []uint32{sysAccountID, 7} { + require.True(t, needSkipTable(accountID, moCatalog, sqlmongodb.TableMappings)) + require.True(t, needSkipSystemTable(accountID, info)) + } + require.Equal(t, int8(1), needSkipTablesInMocatalog[sqlmongodb.TableMappings]) +} + func TestMergeFkDepsDeduplicatesSources(t *testing.T) { child := genKey("d", "child") parent := genKey("d", "parent") diff --git a/pkg/sql/colexec/aggexec/maxby.go b/pkg/sql/colexec/aggexec/maxby.go index 0044e57cc8420..203da3e60b83d 100644 --- a/pkg/sql/colexec/aggexec/maxby.go +++ b/pkg/sql/colexec/aggexec/maxby.go @@ -16,6 +16,7 @@ package aggexec import ( "bytes" + "io" "math" "slices" @@ -34,6 +35,12 @@ const maxByVarlenaCompactionSlack = 1 << 20 type maxByExec struct { aggExec nonNullValue bool + varlenaUsage [][]maxByVarlenaUsage +} + +type maxByVarlenaUsage struct { + liveBytes int + staleBytes int } func makeMaxByExec(mp *mpool.MPool, id int64, nonNullValue bool, params []types.Type) AggFuncExec { @@ -84,7 +91,7 @@ func (exec *maxByExec) BatchFill(offset int, groups []uint64, vectors []*vector. x, y := exec.getXY(group - 1) state := &exec.state[x] if state.vecs[1].IsNull(uint64(y)) || candidateWins(vectors, rows, state.vecs, int(y), exec.argTypes) { - if err := exec.copyWinner(state.vecs, int(y), vectors, rows); err != nil { + if err := exec.copyWinner(x, state.vecs, int(y), vectors, rows); err != nil { return err } } @@ -114,7 +121,7 @@ func (exec *maxByExec) BatchMerge(next AggFuncExec, offset int, groups []uint64) current := exec.state[x1].vecs rows := [3]int{int(y2), int(y2), int(y2)} if current[1].IsNull(uint64(y1)) || candidateWins(candidate, rows, current, int(y1), exec.argTypes) { - if err := exec.copyWinner(current, int(y1), candidate, rows); err != nil { + if err := exec.copyWinner(x1, current, int(y1), candidate, rows); err != nil { return err } } @@ -228,18 +235,34 @@ func compareFloat64(a, b float64) int { return types.GenericAscCompare(a, b) } -func (exec *maxByExec) copyWinner(dst []*vector.Vector, dstRow int, src []*vector.Vector, srcRows [3]int) error { +func (exec *maxByExec) copyWinner( + chunk int, + dst []*vector.Vector, + dstRow int, + src []*vector.Vector, + srcRows [3]int, +) error { + usage := exec.ensureVarlenaUsage(chunk) + oldLive := make([]int, len(dst)) + newLive := make([]int, len(dst)) // Reserve every fallible varlen allocation before mutating any of the three // correlated state vectors. Without this preflight, an OOM after copying the // value but before copying order/tie would publish a mixed winner. Growing // capacity is harmless if a later reservation fails; the logical state stays // byte-for-byte unchanged and remains safe to serialize or free. for i := range dst { - if src[i].IsNull(uint64(srcRows[i])) || !dst[i].GetType().IsVarlen() { + if !dst[i].GetType().IsVarlen() { + continue + } + if !dst[i].IsNull(uint64(dstRow)) { + oldLive[i] = maxByAreaBytes(dst[i].GetRawBytesAt(dstRow)) + } + if src[i].IsNull(uint64(srcRows[i])) { continue } valueBytes := len(src[i].GetRawBytesAt(srcRows[i])) - if valueBytes <= types.VarlenaInlineSize { + newLive[i] = maxByAreaBytes(src[i].GetRawBytesAt(srcRows[i])) + if newLive[i] == 0 { continue } if err := dst[i].PreExtendWithArea(0, valueBytes, exec.mp); err != nil { @@ -256,35 +279,57 @@ func (exec *maxByExec) copyWinner(dst []*vector.Vector, dstRow int, src []*vecto } dst[i].UnsetNull(uint64(dstRow)) } - for _, vec := range dst { + for i, vec := range dst { + if vec.GetType().IsVarlen() { + usage[i].liveBytes += newLive[i] - oldLive[i] + usage[i].staleBytes += oldLive[i] + } // Compaction is an optional bound on stale varlen area. A failed compact // clone leaves the valid original untouched, so memory pressure must not // turn a fully copied winner into an aggregate error with ambiguous state. - _ = compactMaxByStateVector(vec, exec.mp) + _ = compactMaxByStateVector(vec, &usage[i], exec.mp) } return nil } -func compactMaxByStateVector(vec *vector.Vector, mp *mpool.MPool) error { - if vec == nil || !vec.GetType().IsVarlen() { - return nil +func maxByAreaBytes(value []byte) int { + if len(value) <= types.VarlenaInlineSize { + return 0 } - fixedCapacity := vec.Capacity() * vec.GetType().TypeSize() - areaCapacity := vec.Allocated() - fixedCapacity - if areaCapacity <= maxByVarlenaCompactionSlack { - return nil + return len(value) +} + +func (exec *maxByExec) ensureVarlenaUsage(chunk int) []maxByVarlenaUsage { + for len(exec.varlenaUsage) < len(exec.state) { + exec.varlenaUsage = append(exec.varlenaUsage, nil) + } + if exec.varlenaUsage[chunk] != nil { + return exec.varlenaUsage[chunk] } - liveBytes := 0 - for row := 0; row < vec.Length(); row++ { - if vec.IsNull(uint64(row)) { + usage := make([]maxByVarlenaUsage, len(exec.state[chunk].vecs)) + for i, vec := range exec.state[chunk].vecs { + if vec == nil || !vec.GetType().IsVarlen() { continue } - valueBytes := len(vec.GetRawBytesAt(row)) - if valueBytes > types.VarlenaInlineSize { - liveBytes += valueBytes + for row := 0; row < vec.Length(); row++ { + if !vec.IsNull(uint64(row)) { + usage[i].liveBytes += maxByAreaBytes(vec.GetRawBytesAt(row)) + } } + usage[i].staleBytes = max(0, len(vec.GetArea())-usage[i].liveBytes) } - if areaCapacity <= 2*liveBytes+maxByVarlenaCompactionSlack { + exec.varlenaUsage[chunk] = usage + return usage +} + +func compactMaxByStateVector(vec *vector.Vector, usage *maxByVarlenaUsage, mp *mpool.MPool) error { + if vec == nil || !vec.GetType().IsVarlen() { + return nil + } + fixedCapacity := vec.Capacity() * vec.GetType().TypeSize() + areaCapacity := vec.Allocated() - fixedCapacity + if areaCapacity <= maxByVarlenaCompactionSlack || + usage.staleBytes <= usage.liveBytes+maxByVarlenaCompactionSlack { return nil } compact, err := vec.CloneToFlatCompact(mp) @@ -293,9 +338,36 @@ func compactMaxByStateVector(vec *vector.Vector, mp *mpool.MPool) error { } vec.Free(mp) *vec = *compact + usage.staleBytes = 0 + return nil +} + +func (exec *maxByExec) GroupGrow(more int) error { + oldChunks := len(exec.state) + if err := exec.aggExec.GroupGrow(more); err != nil { + return err + } + if exec.chunkSize == 1 { + // The single-group fast path replaces state[0] instead of appending a + // chunk, so any accounting derived from the prior vector is invalid. + exec.varlenaUsage = nil + oldChunks = 0 + } + for len(exec.varlenaUsage) < len(exec.state) { + exec.varlenaUsage = append(exec.varlenaUsage, nil) + } + for chunk := oldChunks; chunk < len(exec.state); chunk++ { + exec.varlenaUsage[chunk] = make([]maxByVarlenaUsage, len(exec.state[chunk].vecs)) + } return nil } +func (exec *maxByExec) UnmarshalFromReader(reader io.Reader, mp *mpool.MPool) error { + err := exec.aggExec.UnmarshalFromReader(reader, mp) + exec.varlenaUsage = nil + return err +} + func (exec *maxByExec) SetExtraInformation(any, int) error { return nil } func (exec *maxByExec) Flush() ([]*vector.Vector, error) { @@ -310,5 +382,6 @@ func (exec *maxByExec) Flush() ([]*vector.Vector, error) { exec.state[i].length = 0 exec.state[i].capacity = 0 } + exec.varlenaUsage = nil return result, nil } diff --git a/pkg/sql/colexec/aggexec/maxby_test.go b/pkg/sql/colexec/aggexec/maxby_test.go index 78a2533e22b08..a3d8adb92e30e 100644 --- a/pkg/sql/colexec/aggexec/maxby_test.go +++ b/pkg/sql/colexec/aggexec/maxby_test.go @@ -80,6 +80,107 @@ func TestMaxByCompactsReplacedVarlenaState(t *testing.T) { require.Less(t, state.Allocated(), 2<<20, "winner state must be bounded by live groups, not by replaced input rows") } +func TestMaxByTracksManyGroupVarlenaUsageIncrementally(t *testing.T) { + mp := mpool.MustNewZero() + params := []types.Type{types.T_varchar.ToType(), types.T_int64.ToType(), types.T_int64.ToType()} + exec := makeMaxByExec(mp, 7012, false, params).(*maxByExec) + require.NoError(t, exec.GroupGrow(AggBatchSize)) + + valueVec := vector.NewVec(types.T_varchar.ToType()) + orderVec := vector.NewVec(types.T_int64.ToType()) + tieVec := vector.NewVec(types.T_int64.ToType()) + groups := make([]uint64, AggBatchSize) + initialValue := []byte(strings.Repeat("i", 128)) + for i := range groups { + groups[i] = uint64(i + 1) + require.NoError(t, vector.AppendBytes(valueVec, initialValue, false, mp)) + require.NoError(t, vector.AppendFixed(orderVec, int64(0), false, mp)) + require.NoError(t, vector.AppendFixed(tieVec, int64(i), false, mp)) + } + require.NoError(t, exec.BatchFill(0, groups, []*vector.Vector{valueVec, orderVec, tieVec})) + require.Equal(t, AggBatchSize*len(initialValue), exec.varlenaUsage[0][0].liveBytes) + + candidateValue := vector.NewVec(types.T_varchar.ToType()) + candidateOrder := vector.NewVec(types.T_int64.ToType()) + candidateTie := vector.NewVec(types.T_int64.ToType()) + require.NoError(t, vector.AppendBytes(candidateValue, []byte(strings.Repeat("w", 64<<10)), false, mp)) + require.NoError(t, vector.AppendFixed(candidateOrder, int64(1), false, mp)) + require.NoError(t, vector.AppendFixed(candidateTie, int64(0), false, mp)) + for winner := int64(1); winner <= 64; winner++ { + require.NoError(t, vector.SetFixedAtNoTypeCheck(candidateOrder, 0, winner)) + require.NoError(t, exec.Fill(0, 0, []*vector.Vector{candidateValue, candidateOrder, candidateTie})) + } + + usage := exec.varlenaUsage[0][0] + require.Equal(t, (AggBatchSize-1)*len(initialValue)+(64<<10), usage.liveBytes) + require.LessOrEqual(t, usage.staleBytes, usage.liveBytes+maxByVarlenaCompactionSlack, + "compaction should reset stale accounting without rescanning all groups per winner") + require.Less(t, exec.state[0].vecs[0].Allocated(), 5<<20) + + for _, vec := range []*vector.Vector{valueVec, orderVec, tieVec, candidateValue, candidateOrder, candidateTie} { + vec.Free(mp) + } + exec.Free() + require.Zero(t, mp.CurrNB()) +} + +func BenchmarkMaxByManyGroupsRepeatedWinners(b *testing.B) { + mp := mpool.MustNewZero() + params := []types.Type{types.T_varchar.ToType(), types.T_int64.ToType(), types.T_int64.ToType()} + exec := makeMaxByExec(mp, 7013, false, params).(*maxByExec) + if err := exec.GroupGrow(AggBatchSize); err != nil { + b.Fatal(err) + } + valueVec := vector.NewVec(types.T_varchar.ToType()) + orderVec := vector.NewVec(types.T_int64.ToType()) + tieVec := vector.NewVec(types.T_int64.ToType()) + groups := make([]uint64, AggBatchSize) + initialValue := []byte(strings.Repeat("i", 128)) + for i := range groups { + groups[i] = uint64(i + 1) + if err := vector.AppendBytes(valueVec, initialValue, false, mp); err != nil { + b.Fatal(err) + } + if err := vector.AppendFixed(orderVec, int64(0), false, mp); err != nil { + b.Fatal(err) + } + if err := vector.AppendFixed(tieVec, int64(i), false, mp); err != nil { + b.Fatal(err) + } + } + if err := exec.BatchFill(0, groups, []*vector.Vector{valueVec, orderVec, tieVec}); err != nil { + b.Fatal(err) + } + candidateValue := vector.NewVec(types.T_varchar.ToType()) + candidateOrder := vector.NewVec(types.T_int64.ToType()) + candidateTie := vector.NewVec(types.T_int64.ToType()) + if err := vector.AppendBytes(candidateValue, []byte(strings.Repeat("w", 64<<10)), false, mp); err != nil { + b.Fatal(err) + } + if err := vector.AppendFixed(candidateOrder, int64(1), false, mp); err != nil { + b.Fatal(err) + } + if err := vector.AppendFixed(candidateTie, int64(0), false, mp); err != nil { + b.Fatal(err) + } + candidate := []*vector.Vector{candidateValue, candidateOrder, candidateTie} + defer func() { + for _, vec := range append([]*vector.Vector{valueVec, orderVec, tieVec}, candidate...) { + vec.Free(mp) + } + exec.Free() + }() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if err := vector.SetFixedAtNoTypeCheck(candidate[1], 0, int64(i+1)); err != nil { + b.Fatal(err) + } + if err := exec.Fill(0, 0, candidate); err != nil { + b.Fatal(err) + } + } +} + func TestMaxByNullContractAndDeterministicMerge(t *testing.T) { mp := mpool.MustNewZero() params := []types.Type{types.T_varchar.ToType(), types.T_int64.ToType(), types.T_varchar.ToType()} diff --git a/pkg/sql/colexec/mongoscan/mongoscan.go b/pkg/sql/colexec/mongoscan/mongoscan.go index 5574c8051e222..69e6ae23985ba 100644 --- a/pkg/sql/colexec/mongoscan/mongoscan.go +++ b/pkg/sql/colexec/mongoscan/mongoscan.go @@ -221,8 +221,15 @@ func (scan *MongoScan) Call(proc *process.Process) (vm.CallResult, error) { break } decodeStarted := time.Now() - if err = scan.ctr.converter.AppendDocument(proc.Ctx, bat, raw, proc.Mp()); err != nil { + if err = scan.ctr.converter.AppendDocumentWithBudget(proc.Ctx, bat, raw, proc.Mp(), maxBatchBytes); err != nil { metric.MongoDBPhaseDurationHistogram.WithLabelValues("decode_append").Observe(time.Since(decodeStarted).Seconds()) + if mongodb.IsDecodedBatchBudgetExceeded(err) && bat.RowCount() > 0 { + // The current document did not fit the decoded/vector budget. Keep + // it for the next Call just like a raw-byte boundary; the converter + // rolled every vector back to the last committed row. + scan.ctr.pendingRaw = append(scan.ctr.pendingRaw[:0], raw...) + break + } bat.Clean(proc.Mp()) scan.ctr.done = true scan.closeResources(proc.Ctx) diff --git a/pkg/sql/colexec/mongoscan/mongoscan_test.go b/pkg/sql/colexec/mongoscan/mongoscan_test.go index b6f7790845c14..d81fe016cbd35 100644 --- a/pkg/sql/colexec/mongoscan/mongoscan_test.go +++ b/pkg/sql/colexec/mongoscan/mongoscan_test.go @@ -16,6 +16,7 @@ package mongoscan import ( "context" "errors" + "fmt" "sync" "testing" @@ -374,6 +375,42 @@ func TestMongoScanBatchAndStatementLimits(t *testing.T) { require.Zero(t, proc.Mp().CurrNB()) }) + t.Run("decoded duplicated projection exceeds batch", func(t *testing.T) { + payload := make([]byte, 256<<10) + doc, err := bson.Marshal(bson.D{{Key: "payload", Value: bson.Binary{Data: payload}}}) + require.NoError(t, err) + cursor := &testCursor{docs: [][]byte{doc}} + deps, _ := testScanDependencies(cursor) + deps.Config.BatchRows = 10 + deps.Config.MaxBatchBytes = 1 << 20 + mapping := deps.Mappings.(testMappingResolver) + mapping.mapping.Columns = nil + spec := testScanPlan() + spec.Columns = nil + for i := range 8 { + name := fmt.Sprintf("payload_%d", i) + mapping.mapping.Columns = append(mapping.mapping.Columns, mongodb.ColumnMapping{ + Name: name, Path: "payload", TypeID: int32(types.T_blob), Conversion: mongodb.ConversionStrict, + }) + spec.Columns = append(spec.Columns, &plan.MongoColumnMapping{ + Name: name, Path: "payload", MoType: plan.Type{Id: int32(types.T_blob)}, ConversionMode: mongodb.ConversionStrict, + }) + } + deps.Mappings = mapping + proc := testutil.NewProcessWithMPool(t, "", mpool.MustNewZero()) + proc.Ctx = defines.AttachAccountId(proc.Ctx, 7) + scan := NewArgument().WithScan(spec) + scan.Dependencies = deps + require.NoError(t, scan.Prepare(proc)) + _, err = scan.Call(proc) + require.True(t, mongodb.IsDecodedBatchBudgetExceeded(err)) + require.Equal(t, 1, cursor.closed) + scan.Free(proc, true, err) + require.NoError(t, deps.Pool.Close(t.Context())) + proc.Free() + require.Zero(t, proc.Mp().CurrNB()) + }) + t.Run("statement row limit", func(t *testing.T) { cursor := &testCursor{docs: [][]byte{doc1, doc2}} deps, _ := testScanDependencies(cursor) diff --git a/pkg/sql/compile/ddl.go b/pkg/sql/compile/ddl.go index 4542f45475d21..7dd501391b170 100644 --- a/pkg/sql/compile/ddl.go +++ b/pkg/sql/compile/ddl.go @@ -2161,11 +2161,17 @@ func icebergCreateSQLFromPlanTableDef(tableDef *plan.TableDef) string { } func (c *Compile) maybeInsertMongoDBTableMapping(dbSource engine.Database, rel engine.Relation, qry *plan.CreateTable) error { + if qry == nil || qry.GetTableDef() == nil || !features.IsMongoDBExternal(qry.GetTableDef().FeatureFlag) { + return nil + } createSQL := icebergCreateSQLFromPlanTableDef(qry.GetTableDef()) env, found, err := sqlmongodb.ParseCreateSQLEnvelope(c.proc.Ctx, createSQL) - if err != nil || !found { + if err != nil { return err } + if !found { + return moerr.NewInternalError(c.proc.Ctx, "typed MongoDB table plan is missing its catalog envelope") + } accountID, err := defines.GetAccountId(c.proc.Ctx) if err != nil { return err @@ -2226,11 +2232,18 @@ func (c *Compile) lookupMongoDBConnectionID(accountID uint32, name string) (uint } func (c *Compile) maybeDeleteMongoDBTableMapping(dbSource engine.Database, rel engine.Relation, tableDef *plan.TableDef) error { + isMongoDB, err := plan2.IsMongoDBTableDef(c.proc.Ctx, tableDef) + if err != nil || !isMongoDB { + return err + } createSQL := icebergCreateSQLFromPlanTableDef(tableDef) _, found, err := sqlmongodb.ParseCreateSQLEnvelope(c.proc.Ctx, createSQL) - if err != nil || !found { + if err != nil { return err } + if !found { + return moerr.NewInternalError(c.proc.Ctx, "MongoDB external table is missing its catalog envelope") + } accountID, err := defines.GetAccountId(c.proc.Ctx) if err != nil { return err diff --git a/pkg/sql/features/table_feature.go b/pkg/sql/features/table_feature.go index 11bbf7212d9e2..e57449214a87d 100644 --- a/pkg/sql/features/table_feature.go +++ b/pkg/sql/features/table_feature.go @@ -15,9 +15,10 @@ package features const ( - Partitioned = 1 << iota - IndexTable = 1 << iota - Partition = 1 << iota + Partitioned = 1 << iota + IndexTable = 1 << iota + Partition = 1 << iota + MongoDBExternal = 1 << iota ) func IsPartitioned(f uint64) bool { @@ -31,3 +32,7 @@ func IsIndexTable(f uint64) bool { func IsPartition(f uint64) bool { return f&Partition != 0 } + +func IsMongoDBExternal(f uint64) bool { + return f&MongoDBExternal != 0 +} diff --git a/pkg/sql/mongodb/converter.go b/pkg/sql/mongodb/converter.go index 2178454106ec7..aa8ece6edb6a5 100644 --- a/pkg/sql/mongodb/converter.go +++ b/pkg/sql/mongodb/converter.go @@ -35,6 +35,8 @@ import ( const conversionErrorRateMinAttempts = 100 +var errDecodedBatchBudget = errors.New("MongoDB decoded batch byte limit exceeded") + type Converter struct { columns []ColumnMapping maxValueBytes int64 @@ -117,6 +119,19 @@ func (c *Converter) NewBatch() *batch.Batch { } func (c *Converter) AppendDocument(ctx context.Context, bat *batch.Batch, raw []byte, mp *mpool.MPool) error { + return c.AppendDocumentWithBudget(ctx, bat, raw, mp, 0) +} + +// AppendDocumentWithBudget appends one BSON document while admitting every +// vector slot and varlen payload before it is copied. maxBatchBytes <= 0 keeps +// the legacy unbounded behavior for direct converter users. +func (c *Converter) AppendDocumentWithBudget( + ctx context.Context, + bat *batch.Batch, + raw []byte, + mp *mpool.MPool, + maxBatchBytes int64, +) error { if int64(len(raw)) > c.maxValueBytes { return moerr.NewInvalidInput(ctx, "MongoDB BSON document exceeds max-value-bytes") } @@ -125,9 +140,18 @@ func (c *Converter) AppendDocument(ctx context.Context, bat *batch.Batch, raw [] return moerr.NewInvalidInput(ctx, "MongoDB returned invalid BSON") } startRows := bat.RowCount() + startAttempts, startErrors := c.conversionAttempts, c.conversionErrors + budget := decodedBatchBudget{remaining: maxBatchBytes - int64(bat.Size()), enabled: maxBatchBytes > 0} + if budget.enabled && budget.remaining < 0 { + return errDecodedBatchBudget + } committed := false defer func() { if !committed { + // A budget miss can defer this same raw document to the next batch. + // Keep statement-level try_null accounting transactional with the row + // so the retry does not count its conversions twice. + c.conversionAttempts, c.conversionErrors = startAttempts, startErrors for _, vec := range bat.Vecs { vec.SetLength(startRows) } @@ -136,6 +160,9 @@ func (c *Converter) AppendDocument(ctx context.Context, bat *batch.Batch, raw [] }() for i, column := range c.columns { value, found, lookupErr := lookupScalarPath(doc, column.Path) + if err := budget.reserve(int64(bat.Vecs[i].GetType().TypeSize())); err != nil { + return err + } if !found || lookupErr == nil && (value.Type == bson.TypeNull || value.Type == bson.TypeUndefined) { if column.NotNullable { return mongoDBNotNullError(ctx, column) @@ -150,7 +177,7 @@ func (c *Converter) AppendDocument(ctx context.Context, bat *batch.Batch, raw [] } appendErr := lookupErr if appendErr == nil { - appendErr = c.appendValue(bat.Vecs[i], value, column, mp) + appendErr = c.appendValue(bat.Vecs[i], value, column, mp, &budget) } if err := appendErr; err != nil { if errors.Is(err, errConversion) && column.Conversion == ConversionTryNull { @@ -180,6 +207,26 @@ func (c *Converter) AppendDocument(ctx context.Context, bat *batch.Batch, raw [] return nil } +type decodedBatchBudget struct { + remaining int64 + enabled bool +} + +func (b *decodedBatchBudget) reserve(bytes int64) error { + if !b.enabled || bytes <= 0 { + return nil + } + if bytes > b.remaining { + return errDecodedBatchBudget + } + b.remaining -= bytes + return nil +} + +func IsDecodedBatchBudgetExceeded(err error) bool { + return errors.Is(err, errDecodedBatchBudget) +} + func mongoDBNotNullError(ctx context.Context, column ColumnMapping) error { return moerr.NewInvalidInputf(ctx, "MongoDB NOT NULL column %s at path %s produced NULL", column.Name, column.Path) } @@ -209,7 +256,13 @@ func lookupScalarPath(doc bson.Raw, path string) (bson.RawValue, bool, error) { return bson.RawValue{}, false, nil } -func (c *Converter) appendValue(vec *vector.Vector, value bson.RawValue, column ColumnMapping, mp *mpool.MPool) error { +func (c *Converter) appendValue( + vec *vector.Vector, + value bson.RawValue, + column ColumnMapping, + mp *mpool.MPool, + budget *decodedBatchBudget, +) error { target := types.T(column.TypeID) switch target { case types.T_bool: @@ -309,14 +362,18 @@ func (c *Converter) appendValue(vec *vector.Vector, value bson.RawValue, column if !ok { return errConversion } - dt := types.DatetimeFromUnixWithNsec(time.UTC, millis/1000, (millis%1000)*int64(time.Millisecond)) + instant := time.UnixMilli(millis).UTC() + if instant.Year() < types.MinDatetimeYear || instant.Year() > types.MaxDatetimeYear { + return errConversion + } + dt := types.DatetimeFromUnixWithNsec(time.UTC, instant.Unix(), int64(instant.Nanosecond())) switch target { case types.T_date: return vector.AppendFixed(vec, dt.ToDate(), false, mp) case types.T_datetime: return vector.AppendFixed(vec, dt.TruncateToScale(column.Scale), false, mp) default: - return vector.AppendFixed(vec, dt.ToTimestamp(time.UTC), false, mp) + return vector.AppendFixed(vec, dt.ToTimestamp(time.UTC).TruncateToScale(column.Scale), false, mp) } case types.T_char, types.T_varchar, types.T_text: var data []byte @@ -327,7 +384,7 @@ func (c *Converter) appendValue(vec *vector.Vector, value bson.RawValue, column } else { return errConversion } - return c.appendString(vec, data, column.Width, mp) + return c.appendString(vec, data, column.Width, mp, budget) case types.T_binary, types.T_varbinary, types.T_blob: var data []byte if _, bytes, ok := value.BinaryOK(); ok { @@ -337,7 +394,7 @@ func (c *Converter) appendValue(vec *vector.Vector, value bson.RawValue, column } else { return errConversion } - return c.appendBytes(vec, data, column.Width, mp) + return c.appendBytes(vec, data, column.Width, mp, budget) case types.T_json: var decoded any if err := value.Unmarshal(&decoded); err != nil { @@ -358,26 +415,44 @@ func (c *Converter) appendValue(vec *vector.Vector, value bson.RawValue, column if err != nil { return errConversion } - return c.appendBytes(vec, data, column.Width, mp) + return c.appendBytes(vec, data, column.Width, mp, budget) default: return errConversion } } -func (c *Converter) appendBytes(vec *vector.Vector, value []byte, width int32, mp *mpool.MPool) error { +func (c *Converter) appendBytes( + vec *vector.Vector, + value []byte, + width int32, + mp *mpool.MPool, + budget *decodedBatchBudget, +) error { if int64(len(value)) > c.maxValueBytes || width > 0 && int32(len(value)) > width { return errConversion } + if err := budget.reserve(int64(len(value))); err != nil { + return err + } return vector.AppendBytes(vec, value, false, mp) } -func (c *Converter) appendString(vec *vector.Vector, value []byte, width int32, mp *mpool.MPool) error { +func (c *Converter) appendString( + vec *vector.Vector, + value []byte, + width int32, + mp *mpool.MPool, + budget *decodedBatchBudget, +) error { // MO CHAR/VARCHAR width is measured in Unicode code points, while the // memory-protection limit remains a byte limit. This matches the existing // external reader and avoids rejecting valid multi-byte strings early. if int64(len(value)) > c.maxValueBytes || width > 0 && utf8.RuneCount(value) > int(width) { return errConversion } + if err := budget.reserve(int64(len(value))); err != nil { + return err + } return vector.AppendBytes(vec, value, false, mp) } diff --git a/pkg/sql/mongodb/envelope.go b/pkg/sql/mongodb/envelope.go index 4c08dd0e0dcce..6531abaa2eb6d 100644 --- a/pkg/sql/mongodb/envelope.go +++ b/pkg/sql/mongodb/envelope.go @@ -25,10 +25,14 @@ import ( "github.com/matrixorigin/matrixone/pkg/common/moerr" ) -const CreateSQLEnvelopePrefix = "MO_MONGODB:" +const ( + CreateSQLEnvelopePrefix = "MO_MONGODB:" + CreateSQLKindMongoDB = "mongodb_table" +) type CreateSQLEnvelope struct { Version int + Kind string Connection string Database string Collection string @@ -42,8 +46,9 @@ type CreateSQLEnvelope struct { func BuildCreateSQLEnvelope(mapping TableMapping) string { columns, _ := json.Marshal(mapping.Columns) return fmt.Sprintf( - "/* %s version=1; connection=%s; database=%s; collection=%s; schema_mode=%s; conversion_mode=%s; split_key=%s; max_parallelism=%d; columns=%s */", + "/* %s version=2; kind=%s; connection=%s; database=%s; collection=%s; schema_mode=%s; conversion_mode=%s; split_key=%s; max_parallelism=%d; columns=%s */", CreateSQLEnvelopePrefix, + CreateSQLKindMongoDB, url.QueryEscape(mapping.Connection), url.QueryEscape(mapping.Database), url.QueryEscape(mapping.Collection), @@ -56,11 +61,16 @@ func BuildCreateSQLEnvelope(mapping TableMapping) string { } func ParseCreateSQLEnvelope(ctx context.Context, createSQL string) (CreateSQLEnvelope, bool, error) { - idx := strings.Index(createSQL, CreateSQLEnvelopePrefix) - if idx < 0 { + // rel_createsql for a generic external table is user-controlled JSON. Only + // recognize the planner-owned leading comment envelope; searching the whole + // string would let a filepath inject this marker and cross the MongoDB + // account-admin boundary. + createSQL = strings.TrimSpace(createSQL) + prefix := "/* " + CreateSQLEnvelopePrefix + if !strings.HasPrefix(createSQL, prefix) { return CreateSQLEnvelope{}, false, nil } - start := idx + len(CreateSQLEnvelopePrefix) + start := len(prefix) end := strings.Index(createSQL[start:], "*/") if end < 0 { return CreateSQLEnvelope{}, true, moerr.NewInvalidInput(ctx, "MongoDB rel_createsql envelope is not closed") @@ -82,8 +92,18 @@ func ParseCreateSQLEnvelope(ctx context.Context, createSQL string) (CreateSQLEnv fields[strings.ToLower(strings.TrimSpace(key))] = decoded } version, err := strconv.Atoi(fields["version"]) - if err != nil || version != 1 { - return CreateSQLEnvelope{}, true, moerr.NewInvalidInput(ctx, "MongoDB rel_createsql envelope version must be 1") + if err != nil || version < 1 || version > 2 { + return CreateSQLEnvelope{}, true, moerr.NewInvalidInput(ctx, "MongoDB rel_createsql envelope version must be 1 or 2") + } + kind := fields["kind"] + if version == 1 && kind == "" { + // Version 1 predates the explicit kind field. Its anchored envelope is + // still unambiguous and remains readable for tables created before the + // discriminator was added. + kind = CreateSQLKindMongoDB + } + if kind != CreateSQLKindMongoDB { + return CreateSQLEnvelope{}, true, moerr.NewInvalidInput(ctx, "MongoDB rel_createsql envelope kind must be mongodb_table") } parallelism, err := strconv.ParseInt(fields["max_parallelism"], 10, 32) if err != nil || parallelism != 1 { @@ -95,6 +115,7 @@ func ParseCreateSQLEnvelope(ctx context.Context, createSQL string) (CreateSQLEnv } env := CreateSQLEnvelope{ Version: version, + Kind: kind, Connection: fields["connection"], Database: fields["database"], Collection: fields["collection"], diff --git a/pkg/sql/mongodb/mongodb_test.go b/pkg/sql/mongodb/mongodb_test.go index d1aeb7764da1c..157977291f280 100644 --- a/pkg/sql/mongodb/mongodb_test.go +++ b/pkg/sql/mongodb/mongodb_test.go @@ -18,8 +18,10 @@ import ( "context" "crypto/tls" "errors" + "fmt" "math" "net" + "runtime" "strings" "sync" "testing" @@ -76,6 +78,8 @@ func TestCreateSQLEnvelopeRoundTripAndRejectsParallelMVP(t *testing.T) { env, found, err := ParseCreateSQLEnvelope(ctx, raw) require.NoError(t, err) require.True(t, found) + require.Equal(t, 2, env.Version) + require.Equal(t, CreateSQLKindMongoDB, env.Kind) require.Equal(t, mapping.Connection, env.Connection) require.Equal(t, int32(1), env.MaxParallelism) require.Equal(t, mapping.Columns, env.Columns) @@ -84,6 +88,28 @@ func TestCreateSQLEnvelopeRoundTripAndRejectsParallelMVP(t *testing.T) { require.NoError(t, err) require.False(t, found) + for _, injected := range []string{ + `{"filepath":"` + raw + `"}`, + "create external table x (a int) infile " + raw, + `{"filepath":"MO_MONGODB: version=1; connection=admin */"}`, + } { + _, found, err = ParseCreateSQLEnvelope(ctx, injected) + require.NoError(t, err) + require.False(t, found, injected) + } + + legacy := strings.Replace(raw, "version=2; kind=mongodb_table;", "version=1;", 1) + env, found, err = ParseCreateSQLEnvelope(ctx, legacy) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, 1, env.Version) + require.Equal(t, CreateSQLKindMongoDB, env.Kind) + + wrongKind := strings.Replace(raw, "kind=mongodb_table", "kind=generic_external", 1) + _, found, err = ParseCreateSQLEnvelope(ctx, wrongKind) + require.True(t, found) + require.ErrorContains(t, err, "kind") + bad := BuildCreateSQLEnvelope(TableMapping{ Connection: "c", Database: "d", Collection: "x", MaxParallelism: 2, Columns: mapping.Columns, @@ -426,6 +452,37 @@ func TestPlanPredicatePushesTryNullBSONDateTimeWithSafeRounding(t *testing.T) { require.Nil(t, pushed, "strict temporal conversion must not hide malformed values") } +func TestPlanPredicateKeepsSubMillisecondScaleTemporalMappingsResidual(t *testing.T) { + columnExpr := &planpb.Expr{Expr: &planpb.Expr_Col{Col: &planpb.ColRef{ColPos: 0}}} + literalDatetime := types.DatetimeFromUnixWithNsec(time.UTC, 10, 0) + literalExpr := &planpb.Expr{ + Typ: planpb.Type{Id: int32(types.T_datetime)}, + Expr: &planpb.Expr_Lit{Lit: &planpb.Literal{Value: &planpb.Literal_I64Val{I64Val: int64(literalDatetime)}}}, + } + comparison := &planpb.Expr{Expr: &planpb.Expr_F{F: &planpb.Function{ + Func: &planpb.ObjectRef{ObjName: "="}, Args: []*planpb.Expr{columnExpr, literalExpr}, + }}} + in := &planpb.Expr{Expr: &planpb.Expr_F{F: &planpb.Function{ + Func: &planpb.ObjectRef{ObjName: "in"}, Args: []*planpb.Expr{columnExpr, { + Expr: &planpb.Expr_List{List: &planpb.ExprList{List: []*planpb.Expr{literalExpr}}}, + }}, + }}} + + for _, target := range []types.T{types.T_datetime, types.T_timestamp} { + for scale := int32(0); scale < 3; scale++ { + columns := []*planpb.MongoColumnMapping{{ + Path: "ts", ConversionMode: ConversionTryNull, + MoType: planpb.Type{Id: int32(target), Scale: scale}, + }} + for _, filter := range []*planpb.Expr{comparison, in} { + pushed, residual := PushdownPlanFilters(t.Context(), []*planpb.Expr{filter}, columns) + require.Nil(t, pushed, "%s(%d)", target, scale) + require.NotEmpty(t, residual) + } + } + } +} + func TestTemporalCandidateRoundingBeforeUnixEpoch(t *testing.T) { require.Equal(t, int64(-2), floorDiv(-1001, 1000)) require.Equal(t, int64(-1), ceilDiv(-1001, 1000)) @@ -710,6 +767,79 @@ func TestConverterRejectsOversizeAndUnsupportedType(t *testing.T) { require.Zero(t, mp.CurrNB()) } +func TestConverterTemporalRangeScaleAndTryNull(t *testing.T) { + mp := mpool.MustNewZero() + invalidValues := []bson.DateTime{ + bson.DateTime(math.MinInt64), + bson.DateTime(time.Date(10000, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli()), + bson.DateTime(math.MaxInt64), + } + for _, conversion := range []string{ConversionStrict, ConversionTryNull} { + converter, err := NewConverter(t.Context(), []ColumnMapping{{ + Name: "ts", TypeID: int32(types.T_timestamp), Scale: 0, Conversion: conversion, + }}, 1024) + require.NoError(t, err) + bat := converter.NewBatch() + for _, invalid := range invalidValues { + raw, marshalErr := bson.Marshal(bson.D{{Key: "ts", Value: invalid}}) + require.NoError(t, marshalErr) + err = converter.AppendDocument(t.Context(), bat, raw, mp) + if conversion == ConversionStrict { + require.ErrorContains(t, err, "cannot be converted") + require.Zero(t, bat.RowCount()) + } else { + require.NoError(t, err) + require.True(t, bat.Vecs[0].IsNull(uint64(bat.RowCount()-1))) + } + } + bat.Clean(mp) + } + + converter, err := NewConverter(t.Context(), []ColumnMapping{{ + Name: "ts", TypeID: int32(types.T_timestamp), Scale: 0, Conversion: ConversionStrict, + }}, 1024) + require.NoError(t, err) + bat := converter.NewBatch() + instant := time.Date(2026, 7, 29, 10, 11, 12, 100*int(time.Millisecond), time.UTC) + raw, err := bson.Marshal(bson.D{{Key: "ts", Value: instant}}) + require.NoError(t, err) + require.NoError(t, converter.AppendDocument(t.Context(), bat, raw, mp)) + want, err := types.ParseTimestamp(time.UTC, "2026-07-29 10:11:12", 0) + require.NoError(t, err) + require.Equal(t, want, vector.GetFixedAtNoTypeCheck[types.Timestamp](bat.Vecs[0], 0)) + bat.Clean(mp) + require.Zero(t, mp.CurrNB()) +} + +func TestConverterEnforcesDecodedVectorBudgetIncrementally(t *testing.T) { + const ( + columnCount = 64 + valueBytes = 256 << 10 + budget = 1 << 20 + ) + columns := make([]ColumnMapping, columnCount) + for i := range columns { + columns[i] = ColumnMapping{ + Name: fmt.Sprintf("copy_%d", i), Path: "payload", + TypeID: int32(types.T_blob), Conversion: ConversionTryNull, + } + } + converter, err := NewConverter(t.Context(), columns, valueBytes+1024) + require.NoError(t, err) + raw, err := bson.Marshal(bson.D{{Key: "payload", Value: bson.Binary{Data: make([]byte, valueBytes)}}}) + require.NoError(t, err) + mp := mpool.MustNewZero() + bat := converter.NewBatch() + err = converter.AppendDocumentWithBudget(t.Context(), bat, raw, mp, budget) + require.True(t, IsDecodedBatchBudgetExceeded(err), "unexpected conversion result: %v", err) + require.Zero(t, bat.RowCount()) + require.Zero(t, converter.conversionAttempts, "a deferred row must not be counted before it commits") + require.LessOrEqual(t, bat.Size(), budget) + require.Less(t, bat.Allocated(), 2*budget, "conversion must stop before duplicating the value into every mapped column") + bat.Clean(mp) + require.Zero(t, mp.CurrNB()) +} + func TestConverterVarcharWidthCountsUnicodeCharacters(t *testing.T) { converter, err := NewConverter(t.Context(), []ColumnMapping{ {Name: "value", Path: "value", TypeID: int32(types.T_varchar), Width: 2}, @@ -1333,6 +1463,112 @@ func TestClientPoolTenantIsolationRotationAndIdempotentRelease(t *testing.T) { require.NoError(t, pool.Close(ctx)) } +type singleflightFactory struct { + mu sync.Mutex + connects int + started chan struct{} + release chan struct{} + err error +} + +func (f *singleflightFactory) Connect(context.Context, Connection, Credentials, RuntimeConfig) (Client, error) { + f.mu.Lock() + f.connects++ + f.mu.Unlock() + f.started <- struct{}{} + <-f.release + if f.err != nil { + return nil, f.err + } + return &fakeClient{}, nil +} + +func TestClientPoolSingleflightsColdAcquisitionPerExactKey(t *testing.T) { + const callers = 16 + factory := &singleflightFactory{ + started: make(chan struct{}, callers), + release: make(chan struct{}), + } + pool := NewClientPool(factory) + connection := Connection{AccountID: 1, ConnectionID: 9, Version: 1} + type result struct { + lease *ClientLease + err error + } + results := make(chan result, callers) + for range callers { + go func() { + lease, err := pool.Acquire(t.Context(), connection, Credentials{}, RuntimeConfig{}) + results <- result{lease: lease, err: err} + }() + } + <-factory.started + for { + pool.mu.Lock() + flight := pool.flights[poolKey{accountID: 1, connectionID: 9, version: 1, identity: credentialIdentity(Credentials{})}] + waiting := flight != nil && flight.waiters == callers-1 + pool.mu.Unlock() + if waiting { + break + } + runtime.Gosched() + } + close(factory.release) + + var first Client + for range callers { + acquired := <-results + require.NoError(t, acquired.err) + if first == nil { + first = acquired.lease.Client() + } else { + require.Same(t, first, acquired.lease.Client()) + } + require.NoError(t, acquired.lease.Release(t.Context())) + } + factory.mu.Lock() + require.Equal(t, 1, factory.connects) + factory.mu.Unlock() + require.NoError(t, pool.Close(t.Context())) +} + +func TestClientPoolSingleflightSharesColdAcquisitionFailure(t *testing.T) { + const callers = 16 + connectErr := errors.New("injected connect failure") + factory := &singleflightFactory{ + started: make(chan struct{}, callers), + release: make(chan struct{}), + err: connectErr, + } + pool := NewClientPool(factory) + results := make(chan error, callers) + for range callers { + go func() { + _, err := pool.Acquire(t.Context(), Connection{AccountID: 1, ConnectionID: 9, Version: 1}, Credentials{}, RuntimeConfig{}) + results <- err + }() + } + <-factory.started + for { + pool.mu.Lock() + flight := pool.flights[poolKey{accountID: 1, connectionID: 9, version: 1, identity: credentialIdentity(Credentials{})}] + waiting := flight != nil && flight.waiters == callers-1 + pool.mu.Unlock() + if waiting { + break + } + runtime.Gosched() + } + close(factory.release) + for range callers { + require.ErrorIs(t, <-results, connectErr) + } + factory.mu.Lock() + require.Equal(t, 1, factory.connects) + factory.mu.Unlock() + require.NoError(t, pool.Close(t.Context())) +} + func TestClientPoolDetectsInPlaceSecretRotation(t *testing.T) { factory := &fakeFactory{} pool := NewClientPool(factory) diff --git a/pkg/sql/mongodb/plan_predicate.go b/pkg/sql/mongodb/plan_predicate.go index 487d253b809d0..5106313c5626f 100644 --- a/pkg/sql/mongodb/plan_predicate.go +++ b/pkg/sql/mongodb/plan_predicate.go @@ -137,8 +137,17 @@ func eligibleComparisonColumn(column *plan.MongoColumnMapping) bool { switch types.T(column.MoType.Id) { case types.T_bool, types.T_int8, types.T_int16, types.T_int32, types.T_int64, - types.T_uint8, types.T_uint16, types.T_uint32, types.T_uint64, - types.T_datetime, types.T_timestamp: + types.T_uint8, types.T_uint16, types.T_uint32, types.T_uint64: + return true + case types.T_datetime, types.T_timestamp: + // BSON DateTime has millisecond precision. DATETIME/TIMESTAMP(0..2) + // normalize several distinct BSON instants to the same MO value, so a + // raw equality/comparison/IN predicate can be narrower than the residual. + // Keep those mappings residual-only until a preimage-range translation is + // implemented. + if column.MoType.Scale < 3 { + return false + } return true default: // BSON int64/double values can collapse to the same FLOAT after diff --git a/pkg/sql/mongodb/pool.go b/pkg/sql/mongodb/pool.go index 68b09e04e6ed4..7902fa47decdb 100644 --- a/pkg/sql/mongodb/pool.go +++ b/pkg/sql/mongodb/pool.go @@ -50,6 +50,12 @@ type retirementState struct { dropped bool } +type connectFlight struct { + done chan struct{} + err error + waiters int +} + // ClientPool keeps authentication state tenant- and generation-local. A // rotated connection creates a new key and drains prior generations only // after their last cursor lease is released. @@ -67,6 +73,7 @@ type ClientPool struct { // a late client starts draining instead of republishing an old idle pool. retirements map[connectionKey]retirementState connecting map[connectionKey]int + flights map[poolKey]*connectFlight closed bool clock uint64 maxIdle int @@ -91,6 +98,7 @@ func newClientPool(factory ClientFactory, validator ConnectionResolver, maxCache entries: make(map[poolKey]*poolEntry), retirements: make(map[connectionKey]retirementState), connecting: make(map[connectionKey]int), + flights: make(map[poolKey]*connectFlight), maxIdle: maxIdle, } } @@ -117,20 +125,43 @@ func (p *ClientPool) Acquire(ctx context.Context, connection Connection, credent version: connection.Version, identity: credentialIdentity(credentials), } - p.mu.Lock() - if p.closed { - p.mu.Unlock() - return nil, moerr.NewInternalError(ctx, "MongoDB client pool is closed") - } - if entry := p.entries[key]; entry != nil { - // Exact-key reuse is safe even while the generation is draining: only - // statements whose catalog plan already carries this old version can - // ask for it. New statements resolve the newer version and cannot land - // here. - entry.refs++ - client := entry.client - p.mu.Unlock() - return &ClientLease{pool: p, key: key, client: client}, nil + var flight *connectFlight + for { + p.mu.Lock() + if p.closed { + p.mu.Unlock() + return nil, moerr.NewInternalError(ctx, "MongoDB client pool is closed") + } + if entry := p.entries[key]; entry != nil { + // Exact-key reuse is safe even while the generation is draining: only + // statements whose catalog plan already carries this old version can + // ask for it. New statements resolve the newer version and cannot land + // here. + entry.refs++ + client := entry.client + p.mu.Unlock() + return &ClientLease{pool: p, key: key, client: client}, nil + } + if existing := p.flights[key]; existing != nil { + existing.waiters++ + done := existing.done + p.mu.Unlock() + select { + case <-done: + if err := context.Cause(ctx); err != nil { + return nil, err + } + if existing.err != nil { + return nil, existing.err + } + continue + case <-ctx.Done(): + return nil, context.Cause(ctx) + } + } + flight = &connectFlight{done: make(chan struct{})} + p.flights[key] = flight + break } sourceKey := connectionKey{accountID: key.accountID, connectionID: key.connectionID} p.connecting[sourceKey]++ @@ -142,7 +173,7 @@ func (p *ClientPool) Acquire(ctx context.Context, connection Connection, credent ctx, connection.AccountID, connection.ConnectionID, connection.Version) if validateErr != nil { p.mu.Lock() - p.finishConnectLocked(sourceKey) + p.finishConnectLocked(sourceKey, key, flight, validateErr) p.pruneRetirementLocked(sourceKey) p.mu.Unlock() return nil, validateErr @@ -156,11 +187,12 @@ func (p *ClientPool) Acquire(ctx context.Context, connection Connection, credent validated.Version != connection.Version || validated.CredentialSecretRef != connection.CredentialSecretRef || validated.TLSCASecretRef != connection.TLSCASecretRef { + changedErr := moerr.NewInvalidInput(ctx, "MongoDB connection changed during client acquisition") p.mu.Lock() - p.finishConnectLocked(sourceKey) + p.finishConnectLocked(sourceKey, key, flight, changedErr) p.pruneRetirementLocked(sourceKey) p.mu.Unlock() - return nil, moerr.NewInvalidInput(ctx, "MongoDB connection changed during client acquisition") + return nil, changedErr } connection = validated } @@ -168,14 +200,14 @@ func (p *ClientPool) Acquire(ctx context.Context, connection Connection, credent client, err := p.factory.Connect(ctx, connection, credentials, cfg) if err != nil { p.mu.Lock() - p.finishConnectLocked(sourceKey) + p.finishConnectLocked(sourceKey, key, flight, err) p.pruneRetirementLocked(sourceKey) p.mu.Unlock() return nil, err } p.mu.Lock() - p.finishConnectLocked(sourceKey) + p.finishConnectLocked(sourceKey, key, flight, nil) if p.closed { p.mu.Unlock() _ = disconnectClients([]Client{client}) @@ -355,12 +387,17 @@ func (p *ClientPool) isRetiredLocked(key poolKey) bool { return retirement.dropped || key.version < retirement.versionFloor } -func (p *ClientPool) finishConnectLocked(connection connectionKey) { +func (p *ClientPool) finishConnectLocked(connection connectionKey, key poolKey, flight *connectFlight, err error) { if p.connecting[connection] <= 1 { delete(p.connecting, connection) } else { p.connecting[connection]-- } + if p.flights[key] == flight { + flight.err = err + delete(p.flights, key) + close(flight.done) + } } // A retirement tombstone is needed only while a pre-DDL Connect can still @@ -420,6 +457,10 @@ func (p *ClientPool) Close(_ context.Context) error { p.entries = make(map[poolKey]*poolEntry) p.retirements = make(map[connectionKey]retirementState) p.connecting = make(map[connectionKey]int) + for key, flight := range p.flights { + delete(p.flights, key) + close(flight.done) + } p.mu.Unlock() return disconnectClients(clients) } @@ -464,6 +505,7 @@ type RuntimeDependencies struct { Secrets SecretResolver Pool *ClientPool Limiter *SourceLimiter + Retirements *ClientRetirementQueue } const ( diff --git a/pkg/sql/mongodb/retirement.go b/pkg/sql/mongodb/retirement.go index 81c08c7a06528..19f66fad2148d 100644 --- a/pkg/sql/mongodb/retirement.go +++ b/pkg/sql/mongodb/retirement.go @@ -64,6 +64,91 @@ type ClusterRemoteClientRetirer struct { Timeout time.Duration } +type RemoteClientRetirer interface { + Retire(context.Context, ClientRetirement) +} + +const DefaultClientRetirementQueueCapacity = 256 + +// ClientRetirementQueue moves best-effort local disconnects and cluster fanout +// off the post-commit path. Its bounded channel prevents a slow/unavailable CN +// from turning restore or DDL churn into unbounded goroutines or memory. +type ClientRetirementQueue struct { + pool *ClientPool + remote RemoteClientRetirer + jobs chan ClientRetirement + ctx context.Context + cancel context.CancelFunc + done chan struct{} + once sync.Once +} + +func NewClientRetirementQueue(pool *ClientPool, remote RemoteClientRetirer, capacity int) *ClientRetirementQueue { + if capacity <= 0 { + capacity = DefaultClientRetirementQueueCapacity + } + ctx, cancel := context.WithCancel(context.Background()) + queue := &ClientRetirementQueue{ + pool: pool, remote: remote, jobs: make(chan ClientRetirement, capacity), + ctx: ctx, cancel: cancel, done: make(chan struct{}), + } + go queue.run() + return queue +} + +// Submit never waits for remote I/O or client Disconnect. False means the +// best-effort queue is stopping or saturated; catalog generation validation +// remains the correctness authority in either case. +func (q *ClientRetirementQueue) Submit(retirement ClientRetirement) bool { + if q == nil { + return false + } + select { + case <-q.ctx.Done(): + return false + default: + } + select { + case q.jobs <- retirement: + return true + case <-q.ctx.Done(): + return false + default: + return false + } +} + +func (q *ClientRetirementQueue) run() { + defer close(q.done) + for { + select { + case <-q.ctx.Done(): + return + case retirement := <-q.jobs: + _ = retirement.Apply(q.pool) + if q.remote != nil { + q.remote.Retire(q.ctx, retirement) + } + } + } +} + +func (q *ClientRetirementQueue) Close(ctx context.Context) error { + if q == nil { + return nil + } + q.once.Do(q.cancel) + if ctx == nil { + ctx = context.Background() + } + select { + case <-q.done: + return nil + case <-ctx.Done(): + return context.Cause(ctx) + } +} + func (r ClusterRemoteClientRetirer) Retire(ctx context.Context, retirement ClientRetirement) { if r.Cluster == nil || r.QueryClient == nil { return diff --git a/pkg/sql/mongodb/retirement_test.go b/pkg/sql/mongodb/retirement_test.go index 9ad843bf37ff4..6ed62f005bcaf 100644 --- a/pkg/sql/mongodb/retirement_test.go +++ b/pkg/sql/mongodb/retirement_test.go @@ -94,6 +94,83 @@ func TestClusterRemoteClientRetirerCoversEveryCNLifecycleScope(t *testing.T) { } } +type blockingRemoteRetirer struct { + started chan ClientRetirement + release chan struct{} +} + +func (r *blockingRemoteRetirer) Retire(ctx context.Context, retirement ClientRetirement) { + r.started <- retirement + select { + case <-r.release: + case <-ctx.Done(): + } +} + +func TestClientRetirementQueueIsAsynchronousAndBounded(t *testing.T) { + remote := &blockingRemoteRetirer{ + started: make(chan ClientRetirement, 1), + release: make(chan struct{}), + } + queue := NewClientRetirementQueue(nil, remote, 1) + first := ClientRetirement{AccountID: 1, ConnectionID: 1} + require.True(t, queue.Submit(first)) + require.Equal(t, first, <-remote.started) + require.True(t, queue.Submit(ClientRetirement{AccountID: 2}), "one job should fit in the bounded backlog") + require.False(t, queue.Submit(ClientRetirement{AccountID: 3}), "a saturated queue must not block the post-commit caller") + close(remote.release) + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + require.NoError(t, queue.Close(ctx)) + require.False(t, queue.Submit(ClientRetirement{AccountID: 4})) +} + +type blockingDisconnectClient struct { + fakeClient + started chan struct{} + release chan struct{} + once sync.Once +} + +func (c *blockingDisconnectClient) Disconnect(ctx context.Context) error { + c.once.Do(func() { close(c.started) }) + select { + case <-c.release: + return nil + case <-ctx.Done(): + return context.Cause(ctx) + } +} + +type fixedRetirementClientFactory struct{ client Client } + +func (f fixedRetirementClientFactory) Connect(context.Context, Connection, Credentials, RuntimeConfig) (Client, error) { + return f.client, nil +} + +func TestClientRetirementQueueMovesLocalDisconnectOffSubmitter(t *testing.T) { + client := &blockingDisconnectClient{started: make(chan struct{}), release: make(chan struct{})} + t.Cleanup(func() { + select { + case <-client.release: + default: + close(client.release) + } + }) + pool := NewClientPool(fixedRetirementClientFactory{client: client}) + lease, err := pool.Acquire(t.Context(), Connection{AccountID: 1, ConnectionID: 9, Version: 1}, Credentials{}, RuntimeConfig{}) + require.NoError(t, err) + require.NoError(t, lease.Release(t.Context())) + + queue := NewClientRetirementQueue(pool, nil, 1) + require.True(t, queue.Submit(ClientRetirement{AccountID: 1, ConnectionID: 9}), + "post-commit submission must finish before Disconnect") + <-client.started + close(client.release) + require.NoError(t, queue.Close(t.Context())) + require.NoError(t, pool.Close(t.Context())) +} + func seedIdleClient(t *testing.T, pool *ClientPool, connection Connection) *fakeClient { t.Helper() lease, err := pool.Acquire(t.Context(), connection, Credentials{}, RuntimeConfig{}) diff --git a/pkg/sql/plan/build_ddl.go b/pkg/sql/plan/build_ddl.go index 34bf0d676d5fc..73eb5a3c01e14 100644 --- a/pkg/sql/plan/build_ddl.go +++ b/pkg/sql/plan/build_ddl.go @@ -1117,6 +1117,10 @@ func buildCreateTable( if err != nil { return nil, err } + // FeatureFlag is durable, planner-owned catalog metadata. Unlike the + // user-controlled rel_createsql payload of a generic external table, it + // is a typed discriminator that cannot be injected through filepath JSON. + createTable.TableDef.FeatureFlag |= features.MongoDBExternal properties := []*plan.Property{ {Key: catalog.SystemRelAttr_Kind, Value: catalog.SystemExternalRel}, {Key: catalog.SystemRelAttr_CreateSQL, Value: sqlmongodb.BuildCreateSQLEnvelope(spec.Mapping)}, diff --git a/pkg/sql/plan/build_ddl_test.go b/pkg/sql/plan/build_ddl_test.go index ea0612715e986..f6ef0100af136 100644 --- a/pkg/sql/plan/build_ddl_test.go +++ b/pkg/sql/plan/build_ddl_test.go @@ -32,6 +32,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/container/types" "github.com/matrixorigin/matrixone/pkg/defines" "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/sql/features" sqlmongodb "github.com/matrixorigin/matrixone/pkg/sql/mongodb" "github.com/matrixorigin/matrixone/pkg/sql/parsers" "github.com/matrixorigin/matrixone/pkg/sql/parsers/dialect" @@ -961,8 +962,9 @@ func TestBuildAlterTableRejectsMongoDBExternalTable(t *testing.T) { ctx := mock.CurrentContext().(*MockCompilerContext) ctx.objects["mongo_ext"] = &plan.ObjectRef{SchemaName: "tpch", ObjName: "mongo_ext"} ctx.tables["mongo_ext"] = &plan.TableDef{ - Name: "mongo_ext", - TableType: catalog.SystemExternalRel, + Name: "mongo_ext", + TableType: catalog.SystemExternalRel, + FeatureFlag: features.MongoDBExternal, Cols: []*plan.ColDef{ {Name: "device_id", Typ: plan.Type{Id: int32(types.T_varchar), Width: 64}}, {Name: "measurement", Typ: plan.Type{Id: int32(types.T_float64)}}, @@ -1030,6 +1032,7 @@ func TestBuildMongoDBExternalTablePreservesNotNullMapping(t *testing.T) { require.NoError(t, err) tableDef := logicPlan.GetDdl().GetCreateTable().GetTableDef() require.NotEmpty(t, tableDef.Cols) + require.True(t, features.IsMongoDBExternal(tableDef.FeatureFlag)) require.Equal(t, "v", tableDef.Cols[0].Name) require.False(t, tableDef.Cols[0].Default.NullAbility) diff --git a/pkg/sql/plan/build_show_util.go b/pkg/sql/plan/build_show_util.go index 8061e7e369d95..7e85e60155724 100644 --- a/pkg/sql/plan/build_show_util.go +++ b/pkg/sql/plan/build_show_util.go @@ -66,12 +66,16 @@ func constructCreateTableSQL( var mongoEnvelope sqlmongodb.CreateSQLEnvelope mongoColumns := make(map[string]sqlmongodb.ColumnMapping) if tableDef.TableType == catalog.SystemExternalRel { - var found bool - mongoEnvelope, found, err = sqlmongodb.ParseCreateSQLEnvelope(ctx.GetContext(), tableDef.Createsql) + var isMongoDB bool + isMongoDB, err = IsMongoDBTableDef(ctx.GetContext(), tableDef) if err != nil { return "", nil, err } - if found { + if isMongoDB { + mongoEnvelope, _, err = sqlmongodb.ParseCreateSQLEnvelope(ctx.GetContext(), tableDef.Createsql) + if err != nil { + return "", nil, err + } for _, column := range mongoEnvelope.Columns { mongoColumns[strings.ToLower(column.Name)] = column } diff --git a/pkg/sql/plan/build_show_util_test.go b/pkg/sql/plan/build_show_util_test.go index c2da281b15d11..122dc2d8370c4 100644 --- a/pkg/sql/plan/build_show_util_test.go +++ b/pkg/sql/plan/build_show_util_test.go @@ -24,6 +24,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/container/types" "github.com/matrixorigin/matrixone/pkg/iceberg/model" "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/sql/features" sqliceberg "github.com/matrixorigin/matrixone/pkg/sql/iceberg" sqlmongodb "github.com/matrixorigin/matrixone/pkg/sql/mongodb" "github.com/matrixorigin/matrixone/pkg/sql/parsers" @@ -664,8 +665,9 @@ func TestShowCreateIcebergExternalTable(t *testing.T) { func TestShowCreateMongoDBExternalTable(t *testing.T) { mock := NewMockOptimizer(false) tableDef := &plan.TableDef{ - Name: "events", - TableType: catalog.SystemExternalRel, + Name: "events", + TableType: catalog.SystemExternalRel, + FeatureFlag: features.MongoDBExternal, Createsql: sqlmongodb.BuildCreateSQLEnvelope(sqlmongodb.TableMapping{ Connection: "telemetry_source", Database: "telemetry", diff --git a/pkg/sql/plan/deepcopy_mongodb_test.go b/pkg/sql/plan/deepcopy_mongodb_test.go index 8d5b1f0b92b46..eeb49bad9e9e4 100644 --- a/pkg/sql/plan/deepcopy_mongodb_test.go +++ b/pkg/sql/plan/deepcopy_mongodb_test.go @@ -17,10 +17,15 @@ package plan import ( "bytes" "context" + "strings" "testing" "github.com/gogo/protobuf/proto" + "github.com/matrixorigin/matrixone/pkg/catalog" + "github.com/matrixorigin/matrixone/pkg/container/types" pb "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/sql/features" + sqlmongodb "github.com/matrixorigin/matrixone/pkg/sql/mongodb" "github.com/stretchr/testify/require" ) @@ -28,6 +33,35 @@ func TestMongoDBTableSurfaceFailsClosedWithoutRuntimeConfig(t *testing.T) { require.Error(t, ensureMongoDBTableSurfaceEnabled(context.Background())) } +func TestMongoDBTableDefinitionRequiresTypedDiscriminator(t *testing.T) { + mapping := sqlmongodb.TableMapping{ + Connection: "source", Database: "db", Collection: "events", + Columns: []sqlmongodb.ColumnMapping{{ + Name: "value", Path: "value", TypeID: int32(types.T_int64), Conversion: sqlmongodb.ConversionStrict, + }}, + } + tableDef := &pb.TableDef{ + TableType: catalog.SystemExternalRel, + Createsql: sqlmongodb.BuildCreateSQLEnvelope(mapping), + } + found, err := IsMongoDBTableDef(t.Context(), tableDef) + require.NoError(t, err) + require.False(t, found, "a v2 text marker alone is not trusted catalog metadata") + + tableDef.FeatureFlag = features.MongoDBExternal + found, err = IsMongoDBTableDef(t.Context(), tableDef) + require.NoError(t, err) + require.True(t, found) + + // Existing v1 tables predate FeatureFlag and remain readable, while their + // envelope is still required to occupy the complete leading catalog value. + tableDef.FeatureFlag = 0 + tableDef.Createsql = strings.Replace(tableDef.Createsql, "version=2; kind=mongodb_table;", "version=1;", 1) + found, err = IsMongoDBTableDef(t.Context(), tableDef) + require.NoError(t, err) + require.True(t, found) +} + func TestMongoScanDeepCopyAndCredentialFreeProto(t *testing.T) { original := &pb.Node{ExternScan: &pb.ExternScan{ Type: int32(pb.ExternType_MONGODB_TB), diff --git a/pkg/sql/plan/mongodb_util.go b/pkg/sql/plan/mongodb_util.go index 43c3902d65912..1a4b3c0fc3b5f 100644 --- a/pkg/sql/plan/mongodb_util.go +++ b/pkg/sql/plan/mongodb_util.go @@ -21,6 +21,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/common/moerr" "github.com/matrixorigin/matrixone/pkg/config" "github.com/matrixorigin/matrixone/pkg/defines" + "github.com/matrixorigin/matrixone/pkg/sql/features" sqlmongodb "github.com/matrixorigin/matrixone/pkg/sql/mongodb" ) @@ -56,9 +57,21 @@ func IsMongoDBTableDef(ctx context.Context, tableDef *TableDef) (bool, error) { if tableDef == nil || tableDef.TableType != catalog.SystemExternalRel { return false, nil } - _, found, err := sqlmongodb.ParseCreateSQLEnvelope(ctx, tableDef.Createsql) + env, found, err := sqlmongodb.ParseCreateSQLEnvelope(ctx, tableDef.Createsql) if err != nil { return false, err } - return found, nil + if !found { + if features.IsMongoDBExternal(tableDef.FeatureFlag) { + return false, moerr.NewInvalidInput(ctx, "MongoDB external table is missing its catalog envelope") + } + return false, nil + } + if env.Version >= 2 { + // Version 2 tables must carry the durable typed feature bit. Version 1 is + // accepted only for backward compatibility with tables created before the + // bit existed; its envelope is safe because it must be the leading value. + return features.IsMongoDBExternal(tableDef.FeatureFlag), nil + } + return true, nil } diff --git a/pkg/sql/plan/query_builder.go b/pkg/sql/plan/query_builder.go index ff488d3954f4e..d1e3d66afc061 100644 --- a/pkg/sql/plan/query_builder.go +++ b/pkg/sql/plan/query_builder.go @@ -8969,13 +8969,20 @@ func (builder *QueryBuilder) buildTable(stmt tree.TableExpr, ctx *BindContext, t } else if found { icebergEnv = env externType = plan.ExternType_ICEBERG_TB - } else if env, found, err := sqlmongodb.ParseCreateSQLEnvelope(builder.GetContext(), tableDef.Createsql); err != nil { - return 0, err - } else if found { - mongoEnv = env - externType = plan.ExternType_MONGODB_TB - if builder.isPrepareStatement { - return 0, moerr.NewNotSupported(builder.GetContext(), "prepared MongoDB external scans") + } else { + isMongoDB, err := IsMongoDBTableDef(builder.GetContext(), tableDef) + if err != nil { + return 0, err + } + if isMongoDB { + mongoEnv, _, err = sqlmongodb.ParseCreateSQLEnvelope(builder.GetContext(), tableDef.Createsql) + if err != nil { + return 0, err + } + externType = plan.ExternType_MONGODB_TB + if builder.isPrepareStatement { + return 0, moerr.NewNotSupported(builder.GetContext(), "prepared MongoDB external scans") + } } } externScan = &plan.ExternScan{ diff --git a/pkg/sql/plan/query_builder_test.go b/pkg/sql/plan/query_builder_test.go index 249738545c022..dedfcdf8dd4cf 100644 --- a/pkg/sql/plan/query_builder_test.go +++ b/pkg/sql/plan/query_builder_test.go @@ -29,6 +29,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/container/types" "github.com/matrixorigin/matrixone/pkg/iceberg/model" "github.com/matrixorigin/matrixone/pkg/pb/plan" + "github.com/matrixorigin/matrixone/pkg/sql/features" sqliceberg "github.com/matrixorigin/matrixone/pkg/sql/iceberg" sqlmongodb "github.com/matrixorigin/matrixone/pkg/sql/mongodb" "github.com/matrixorigin/matrixone/pkg/sql/parsers" @@ -171,7 +172,8 @@ func TestMongoDBExternalScanPruningKeepsResidualColumnsAndPlansPushdown(t *testi } mock.ctxt.tables["events_external"] = &plan.TableDef{ Name: "events_external", TableType: catalog.SystemExternalRel, - Createsql: sqlmongodb.BuildCreateSQLEnvelope(mapping), + FeatureFlag: features.MongoDBExternal, + Createsql: sqlmongodb.BuildCreateSQLEnvelope(mapping), Cols: []*plan.ColDef{ {Name: "device_id", Typ: plan.Type{Id: int32(types.T_varchar), Width: 20}}, {Name: "ts", Typ: plan.Type{Id: int32(types.T_datetime), Scale: 3}}, diff --git a/test/mongodb/mongodb_e2e_local.go b/test/mongodb/mongodb_e2e_local.go index 9226538e35138..784b5a2e23a01 100644 --- a/test/mongodb/mongodb_e2e_local.go +++ b/test/mongodb/mongodb_e2e_local.go @@ -17,7 +17,9 @@ import ( "strings" "time" - _ "github.com/go-sql-driver/mysql" + mysqldriver "github.com/go-sql-driver/mysql" + "github.com/matrixorigin/matrixone/pkg/container/types" + sqlmongodb "github.com/matrixorigin/matrixone/pkg/sql/mongodb" ) type report struct { @@ -46,7 +48,7 @@ func main() { err = waitForMO(ctx, db) } if err == nil { - err = run(ctx, db, host, &r) + err = runWithDSN(ctx, db, dsn, host, &r) } if err == nil { r.Status = "passed" @@ -78,6 +80,10 @@ func waitForMO(ctx context.Context, db *sql.DB) error { } func run(ctx context.Context, db *sql.DB, host string, r *report) error { + return runWithDSN(ctx, db, "", host, r) +} + +func runWithDSN(ctx context.Context, db *sql.DB, dsn, host string, r *report) error { manifest, err := loadFixtureManifest("test/mongodb/fixture_manifest.json") if err != nil { return err @@ -98,6 +104,12 @@ func run(ctx context.Context, db *sql.DB, host string, r *report) error { return err } r.Cases = append(r.Cases, "show-create-redaction-roundtrip") + if dsn != "" { + if err := verifyAuthorizationBoundary(ctx, db, dsn); err != nil { + return err + } + r.Cases = append(r.Cases, "non-admin-marker-injection-boundary") + } if err := expectScalar(ctx, db, "select count(*) from mongodb_ci.events", "5"); err != nil { return err @@ -238,6 +250,74 @@ func run(ctx context.Context, db *sql.DB, host string, r *report) error { return nil } +func verifyAuthorizationBoundary(ctx context.Context, adminDB *sql.DB, dsn string) error { + const ( + roleName = "mongodb_ci_creator" + userName = "mongodb_ci_user" + password = "mongodb_ci_password" + ) + for _, statement := range []string{ + "drop user if exists " + userName, + "drop role if exists " + roleName, + "create role " + roleName, + "create user " + userName + " identified by '" + password + "' default role " + roleName, + "grant connect on account * to " + roleName, + "grant create table on database mongodb_ci to " + roleName, + } { + if _, err := adminDB.ExecContext(ctx, statement); err != nil { + return fmt.Errorf("authorization boundary setup %s: %w", statement, err) + } + } + defer func() { + cleanupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, _ = adminDB.ExecContext(cleanupCtx, "drop user if exists "+userName) + _, _ = adminDB.ExecContext(cleanupCtx, "drop role if exists "+roleName) + }() + + config, err := mysqldriver.ParseDSN(dsn) + if err != nil { + return fmt.Errorf("parse MatrixOne DSN: %w", err) + } + config.User = userName + config.Passwd = password + config.DBName = "mongodb_ci" + userDB, err := sql.Open("mysql", config.FormatDSN()) + if err != nil { + return fmt.Errorf("open non-admin MatrixOne session: %w", err) + } + defer userDB.Close() + if err := userDB.PingContext(ctx); err != nil { + return fmt.Errorf("connect non-admin MatrixOne session: %w", err) + } + + if _, err := userDB.ExecContext(ctx, + "create external table mongodb_ci.denied_mongodb(value bigint) engine=mongodb with ('connection'='mongodb_ci','database'='mongodb_source','collection'='events','schema_mode'='explicit','conversion_mode'='strict','max_parallelism'='1')"); err == nil { + return fmt.Errorf("non-admin MongoDB table creation unexpectedly succeeded") + } + + marker := sqlmongodb.BuildCreateSQLEnvelope(sqlmongodb.TableMapping{ + Connection: "mongodb_ci", Database: "mongodb_source", Collection: "events", + Columns: []sqlmongodb.ColumnMapping{{ + Name: "value", Path: "measurement", TypeID: int32(types.T_int64), Conversion: sqlmongodb.ConversionStrict, + }}, + }) + marker = strings.Replace(marker, "version=2; kind=mongodb_table;", "version=1;", 1) + // Before the parser boundary was anchored, a generic external-table filepath + // containing this valid marker was mistaken for planner-owned MongoDB DDL. + injectionSQL := "create external table mongodb_ci.marker_injection(value bigint) infile{\"filepath\"='" + + strings.ReplaceAll(marker, "'", "''") + "'} fields terminated by ',' lines terminated by '\\n'" + if _, err := userDB.ExecContext(ctx, injectionSQL); err != nil { + return fmt.Errorf("generic marker-injection control table must remain creatable: %w", err) + } + if err := expectScalar(ctx, adminDB, + "select count(*) from mo_catalog.mo_mongodb_tables m join mo_catalog.mo_tables t on m.account_id=t.account_id and m.table_id=t.rel_id where t.account_id=0 and t.reldatabase='mongodb_ci' and t.relname='marker_injection'", + "0"); err != nil { + return fmt.Errorf("generic marker injection created a MongoDB mapping: %w", err) + } + return nil +} + func loadFixtureManifest(path string) (fixtureManifest, error) { data, err := os.ReadFile(path) if err != nil { From 2592c82693fcd9e2baf74247e040e0112ff6ef4f Mon Sep 17 00:00:00 2001 From: iamlinjunhong <1030420200@qq.com> Date: Fri, 31 Jul 2026 16:05:58 +0800 Subject: [PATCH 2/6] fix: repair MongoDB CI validation --- pkg/common/moerr/cause.go | 3 +++ pkg/common/moerr/cause_test.go | 3 +++ pkg/sql/mongodb/converter.go | 2 +- pkg/sql/mongodb/pool.go | 3 ++- pkg/sql/mongodb/retirement.go | 4 +++- test/mongodb/mongodb_e2e_local.go | 18 +++++++++--------- test/mongodb/mongodb_e2e_local_test.go | 20 ++++++++++++++++++++ 7 files changed, 41 insertions(+), 12 deletions(-) diff --git a/pkg/common/moerr/cause.go b/pkg/common/moerr/cause.go index 44528609635b2..d178630f938a2 100644 --- a/pkg/common/moerr/cause.go +++ b/pkg/common/moerr/cause.go @@ -166,6 +166,9 @@ var ( CauseBuildInsertIndexMetaBatch2 = NewInternalError(context.Background(), "buildInsertIndexMetaBatch 2") //pkg/sql/colexec/dispatch CauseWaitRemoteRegsReady = NewInternalError(context.Background(), "waitRemoteRegsReady") + //pkg/sql/mongodb + CauseMongoDBClientCleanup = NewInternalError(context.Background(), "mongodb client cleanup") + CauseMongoDBClientRetirement = NewInternalError(context.Background(), "mongodb client retirement") //pkg/sql/compile CauseIsAvailable = NewInternalError(context.Background(), "isAvailable") CauseNewMessageSenderOnClient = NewInternalError(context.Background(), "newMessageSenderOnClient") diff --git a/pkg/common/moerr/cause_test.go b/pkg/common/moerr/cause_test.go index c8426e9ff4803..77b28f69e3ca9 100644 --- a/pkg/common/moerr/cause_test.go +++ b/pkg/common/moerr/cause_test.go @@ -157,6 +157,9 @@ var causeArray = []error{ CauseWaitRemoteRegsReady, + CauseMongoDBClientCleanup, + CauseMongoDBClientRetirement, + CauseIsAvailable, CauseNewMessageSenderOnClient, CauseWaitingTheStopResponse, diff --git a/pkg/sql/mongodb/converter.go b/pkg/sql/mongodb/converter.go index aa8ece6edb6a5..f544ac793fab3 100644 --- a/pkg/sql/mongodb/converter.go +++ b/pkg/sql/mongodb/converter.go @@ -35,7 +35,7 @@ import ( const conversionErrorRateMinAttempts = 100 -var errDecodedBatchBudget = errors.New("MongoDB decoded batch byte limit exceeded") +var errDecodedBatchBudget = moerr.NewInternalErrorNoCtx("MongoDB decoded batch byte limit exceeded") type Converter struct { columns []ColumnMapping diff --git a/pkg/sql/mongodb/pool.go b/pkg/sql/mongodb/pool.go index 7902fa47decdb..18d8f2eb22a2c 100644 --- a/pkg/sql/mongodb/pool.go +++ b/pkg/sql/mongodb/pool.go @@ -469,7 +469,8 @@ func disconnectClients(clients []Client) error { if len(clients) == 0 { return nil } - ctx, cancel := context.WithTimeout(context.Background(), clientCleanupTimeout) + ctx, cancel := context.WithTimeoutCause( + context.Background(), clientCleanupTimeout, moerr.CauseMongoDBClientCleanup) defer cancel() var first error for _, client := range clients { diff --git a/pkg/sql/mongodb/retirement.go b/pkg/sql/mongodb/retirement.go index 19f66fad2148d..2f4089c5c5abc 100644 --- a/pkg/sql/mongodb/retirement.go +++ b/pkg/sql/mongodb/retirement.go @@ -21,6 +21,7 @@ import ( "time" "github.com/matrixorigin/matrixone/pkg/clusterservice" + "github.com/matrixorigin/matrixone/pkg/common/moerr" "github.com/matrixorigin/matrixone/pkg/pb/metadata" "github.com/matrixorigin/matrixone/pkg/pb/query" ) @@ -169,7 +170,8 @@ func (r ClusterRemoteClientRetirer) Retire(ctx context.Context, retirement Clien return true }) - sendCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout) + sendCtx, cancel := context.WithTimeoutCause( + context.WithoutCancel(ctx), timeout, moerr.CauseMongoDBClientRetirement) defer cancel() var wg sync.WaitGroup for _, address := range targets { diff --git a/test/mongodb/mongodb_e2e_local.go b/test/mongodb/mongodb_e2e_local.go index 784b5a2e23a01..8dd3730ad58ae 100644 --- a/test/mongodb/mongodb_e2e_local.go +++ b/test/mongodb/mongodb_e2e_local.go @@ -11,6 +11,7 @@ import ( "encoding/json" "flag" "fmt" + "net/url" "os" "path/filepath" "reflect" @@ -18,8 +19,6 @@ import ( "time" mysqldriver "github.com/go-sql-driver/mysql" - "github.com/matrixorigin/matrixone/pkg/container/types" - sqlmongodb "github.com/matrixorigin/matrixone/pkg/sql/mongodb" ) type report struct { @@ -296,13 +295,14 @@ func verifyAuthorizationBoundary(ctx context.Context, adminDB *sql.DB, dsn strin return fmt.Errorf("non-admin MongoDB table creation unexpectedly succeeded") } - marker := sqlmongodb.BuildCreateSQLEnvelope(sqlmongodb.TableMapping{ - Connection: "mongodb_ci", Database: "mongodb_source", Collection: "events", - Columns: []sqlmongodb.ColumnMapping{{ - Name: "value", Path: "measurement", TypeID: int32(types.T_int64), Conversion: sqlmongodb.ConversionStrict, - }}, - }) - marker = strings.Replace(marker, "version=2; kind=mongodb_table;", "version=1;", 1) + // Keep the E2E runner independent from MatrixOne's kernel packages: importing + // the production envelope builder here pulls the kernel CGo dependency graph + // into a runtime `go run`. This valid legacy envelope is deliberately local + // test data; type_id 23 is BIGINT in the version-1 catalog encoding. + const columnsJSON = `[{"name":"value","path":"measurement","type_id":23,"conversion":"strict"}]` + marker := "/* MO_MONGODB: version=1; connection=mongodb_ci; database=mongodb_source; collection=events; " + + "schema_mode=explicit; conversion_mode=strict; split_key=; max_parallelism=1; columns=" + + url.QueryEscape(columnsJSON) + " */" // Before the parser boundary was anchored, a generic external-table filepath // containing this valid marker was mistaken for planner-owned MongoDB DDL. injectionSQL := "create external table mongodb_ci.marker_injection(value bigint) infile{\"filepath\"='" + diff --git a/test/mongodb/mongodb_e2e_local_test.go b/test/mongodb/mongodb_e2e_local_test.go index bbd000ee6c085..ad2dad3134ba8 100644 --- a/test/mongodb/mongodb_e2e_local_test.go +++ b/test/mongodb/mongodb_e2e_local_test.go @@ -18,9 +18,12 @@ import ( "context" "database/sql" "errors" + "go/parser" + "go/token" "os" "path/filepath" "runtime" + "strconv" "strings" "testing" "time" @@ -29,6 +32,23 @@ import ( "github.com/stretchr/testify/require" ) +func TestMongoDBLocalE2ERunnerDoesNotImportKernelPackages(t *testing.T) { + repoRoot := mongoDBTestRepoRoot(t) + file, err := parser.ParseFile( + token.NewFileSet(), + filepath.Join(repoRoot, "test/mongodb/mongodb_e2e_local.go"), + nil, + parser.ImportsOnly, + ) + require.NoError(t, err) + for _, imported := range file.Imports { + path, err := strconv.Unquote(imported.Path.Value) + require.NoError(t, err) + require.Falsef(t, strings.HasPrefix(path, "github.com/matrixorigin/matrixone/"), + "standalone E2E runner must not import kernel package %s", path) + } +} + func TestMongoDBLocalE2ERunContract(t *testing.T) { repoRoot := mongoDBTestRepoRoot(t) previous, err := os.Getwd() From e642ce5cf5dbd621bcbb328290e86e027b8263e1 Mon Sep 17 00:00:00 2001 From: iamlinjunhong <1030420200@qq.com> Date: Fri, 31 Jul 2026 17:32:46 +0800 Subject: [PATCH 3/6] test: close MongoDB retirement CI gaps --- pkg/sql/mongodb/retirement.go | 8 +++++ pkg/sql/mongodb/retirement_test.go | 51 ++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+) diff --git a/pkg/sql/mongodb/retirement.go b/pkg/sql/mongodb/retirement.go index 2f4089c5c5abc..f2951c40338cb 100644 --- a/pkg/sql/mongodb/retirement.go +++ b/pkg/sql/mongodb/retirement.go @@ -122,6 +122,14 @@ func (q *ClientRetirementQueue) Submit(retirement ClientRetirement) bool { func (q *ClientRetirementQueue) run() { defer close(q.done) for { + // Once shutdown is visible, do not let a ready backlog win another + // randomized select. At most the retirement already in progress may + // finish before done is closed and the pool becomes eligible to close. + select { + case <-q.ctx.Done(): + return + default: + } select { case <-q.ctx.Done(): return diff --git a/pkg/sql/mongodb/retirement_test.go b/pkg/sql/mongodb/retirement_test.go index 6ed62f005bcaf..cbf84fa4fc19f 100644 --- a/pkg/sql/mongodb/retirement_test.go +++ b/pkg/sql/mongodb/retirement_test.go @@ -107,6 +107,57 @@ func (r *blockingRemoteRetirer) Retire(ctx context.Context, retirement ClientRet } } +func TestClientRetirementQueueLifecycleBoundaries(t *testing.T) { + var nilQueue *ClientRetirementQueue + require.False(t, nilQueue.Submit(ClientRetirement{})) + require.NoError(t, nilQueue.Close(nil)) + + queue := NewClientRetirementQueue(nil, nil, 0) + require.Equal(t, DefaultClientRetirementQueueCapacity, cap(queue.jobs)) + require.NoError(t, queue.Close(nil)) +} + +type stubbornRemoteRetirer struct { + started chan struct{} + calls chan ClientRetirement + release chan struct{} + once sync.Once +} + +func (r *stubbornRemoteRetirer) Retire(_ context.Context, retirement ClientRetirement) { + r.calls <- retirement + r.once.Do(func() { close(r.started) }) + <-r.release +} + +func TestClientRetirementQueueCloseRespectsCallerContext(t *testing.T) { + remote := &stubbornRemoteRetirer{ + started: make(chan struct{}), + calls: make(chan ClientRetirement, 2), + release: make(chan struct{}), + } + t.Cleanup(func() { + select { + case <-remote.release: + default: + close(remote.release) + } + }) + + queue := NewClientRetirementQueue(nil, remote, 2) + require.True(t, queue.Submit(ClientRetirement{})) + <-remote.started + require.True(t, queue.Submit(ClientRetirement{AccountID: 2})) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + require.ErrorIs(t, queue.Close(ctx), context.Canceled) + + close(remote.release) + require.NoError(t, queue.Close(t.Context())) + require.Equal(t, 1, len(remote.calls), "shutdown must not drain the queued backlog") +} + func TestClientRetirementQueueIsAsynchronousAndBounded(t *testing.T) { remote := &blockingRemoteRetirer{ started: make(chan ClientRetirement, 1), From dedb925c5e427005ee408a2651a3017df42d32a7 Mon Sep 17 00:00:00 2001 From: iamlinjunhong <1030420200@qq.com> Date: Mon, 3 Aug 2026 11:07:28 +0800 Subject: [PATCH 4/6] test: cover MongoDB follow-up regressions end to end --- etc/launch-mongodb-local/init_and_seed.js | 13 +++++++++++++ optools/mongodb_ci.bash | 2 +- test/mongodb/mongodb_e2e_local.go | 20 ++++++++++++++++++++ test/mongodb/mongodb_e2e_local_test.go | 6 +++++- 4 files changed, 39 insertions(+), 2 deletions(-) diff --git a/etc/launch-mongodb-local/init_and_seed.js b/etc/launch-mongodb-local/init_and_seed.js index 5614d08ce9061..ba73504f2f1eb 100644 --- a/etc/launch-mongodb-local/init_and_seed.js +++ b/etc/launch-mongodb-local/init_and_seed.js @@ -25,3 +25,16 @@ source.events.insertMany([ {_id: ObjectId("64b000000000000000000005"), device_id: "device-002", site_id: "site-east", ts: ISODate("2026-07-27T10:01:00Z"), measurement: "malformed"} ]); source.events.createIndex({ts: 1, _id: 1}); + +source.temporal_edges.drop(); +source.temporal_edges.insertOne({ + _id: ObjectId("64b000000000000000000101"), + ts: ISODate("2026-07-27T10:00:05.100Z") +}); +source.temporal_edges.createIndex({ts: 1}); + +source.decoded_budget.drop(); +source.decoded_budget.insertOne({ + _id: ObjectId("64b000000000000000000201"), + payload: "x".repeat(192 * 1024) +}); diff --git a/optools/mongodb_ci.bash b/optools/mongodb_ci.bash index b8a9549100174..0e41d89e7f78f 100755 --- a/optools/mongodb_ci.bash +++ b/optools/mongodb_ci.bash @@ -130,7 +130,7 @@ generate_mo_config() { printf '\n[cn.frontend.mongodb]\n' printf 'enable = true\nallow-loopback = true\n' printf 'connect-timeout = "10s"\nserver-selection-timeout = "10s"\nsocket-timeout = "30s"\n' - printf 'batch-rows = 2\nmax-source-concurrency = 2\n' + printf 'batch-rows = 2\nmax-batch-bytes = 1048576\nmax-value-bytes = 524288\nmax-source-concurrency = 2\n' } >>"$generated_dir/cn.toml" sed -e "s#\./etc/launch/log.toml#$generated_dir/log.toml#" \ -e "s#\./etc/launch/tn.toml#$generated_dir/tn.toml#" \ diff --git a/test/mongodb/mongodb_e2e_local.go b/test/mongodb/mongodb_e2e_local.go index 8dd3730ad58ae..ff542ec59033d 100644 --- a/test/mongodb/mongodb_e2e_local.go +++ b/test/mongodb/mongodb_e2e_local.go @@ -92,6 +92,8 @@ func runWithDSN(ctx context.Context, db *sql.DB, dsn, host string, r *report) er "create database mongodb_ci", "create mongodb connection if not exists mongodb_ci with ('hosts'='" + host + "','replica_set'='rs0','auth_source'='mongodb_source','auth_mechanism'='SCRAM-SHA-256','credential_secret_ref'='secret://env/MO_MONGODB_E2E_CREDENTIAL','tls_mode'='disabled','read_preference'='primary','read_concern'='majority','options_json'='{\"direct\":true}')", "create external table mongodb_ci.events(mongo_id char(24) mongodb_path '_id', device_id varchar(20), site_id varchar(10), ts datetime(3) mongodb_convert 'try_null', measurement double mongodb_convert 'try_null', source_batch varchar(50)) engine=mongodb with ('connection'='mongodb_ci','database'='mongodb_source','collection'='events','schema_mode'='explicit','conversion_mode'='strict','max_parallelism'='1')", + "create external table mongodb_ci.temporal_edges(ts datetime(0) mongodb_convert 'try_null') engine=mongodb with ('connection'='mongodb_ci','database'='mongodb_source','collection'='temporal_edges','schema_mode'='explicit','conversion_mode'='strict','max_parallelism'='1')", + "create external table mongodb_ci.decoded_budget(payload_1 text mongodb_path 'payload', payload_2 text mongodb_path 'payload', payload_3 text mongodb_path 'payload', payload_4 text mongodb_path 'payload', payload_5 text mongodb_path 'payload', payload_6 text mongodb_path 'payload', payload_7 text mongodb_path 'payload', payload_8 text mongodb_path 'payload') engine=mongodb with ('connection'='mongodb_ci','database'='mongodb_source','collection'='decoded_budget','schema_mode'='explicit','conversion_mode'='strict','max_parallelism'='1')", } for _, statement := range statements { if _, err := db.ExecContext(ctx, statement); err != nil { @@ -126,6 +128,24 @@ func runWithDSN(ctx context.Context, db *sql.DB, dsn, host string, r *report) er } r.Cases = append(r.Cases, "scan-projection-pushdown-null-conversion") + // BSON DateTime preserves milliseconds, while DATETIME(0) truncates them. + // The source predicate must therefore remain residual-only: an exact MongoDB + // equality on 10:00:05.000 would incorrectly exclude this .100 source row. + if err := expectScalar(ctx, db, "select count(*) from mongodb_ci.temporal_edges where ts = '2026-07-27 10:00:05'", "1"); err != nil { + return err + } + r.Cases = append(r.Cases, "low-precision-temporal-residual") + + // One BSON string is below max-value-bytes and the raw document is below + // max-batch-bytes, but projecting it into eight vectors exceeds the decoded + // batch budget. This guards the allocation amplification fixed by #26485. + if err := expectQueryFailure(ctx, db, + "select payload_1,payload_2,payload_3,payload_4,payload_5,payload_6,payload_7,payload_8 from mongodb_ci.decoded_budget", + "decoded batch byte limit exceeded"); err != nil { + return err + } + r.Cases = append(r.Cases, "decoded-vector-budget-enforced") + cancelCtx, cancel := context.WithCancel(ctx) cancel() if err := db.QueryRowContext(cancelCtx, "select count(*) from mongodb_ci.events").Scan(new(string)); err == nil { diff --git a/test/mongodb/mongodb_e2e_local_test.go b/test/mongodb/mongodb_e2e_local_test.go index ad2dad3134ba8..c0a6d6d2291ce 100644 --- a/test/mongodb/mongodb_e2e_local_test.go +++ b/test/mongodb/mongodb_e2e_local_test.go @@ -60,7 +60,7 @@ func TestMongoDBLocalE2ERunContract(t *testing.T) { require.NoError(t, err) db, mock := newMongoDBE2ESQLMock(t) - for range 4 { + for range 6 { mock.ExpectExec(".*").WillReturnResult(sqlmock.NewResult(0, 1)) } mock.ExpectQuery("show create table").WillReturnRows(sqlmock.NewRows([]string{"table", "ddl"}).AddRow( @@ -74,6 +74,8 @@ func TestMongoDBLocalE2ERunContract(t *testing.T) { mock.ExpectQuery("select mongo_id").WillReturnRows(fixtureRows) expectMongoDBE2EScalar(mock, "3") expectMongoDBE2EScalar(mock, "3") + expectMongoDBE2EScalar(mock, "1") + mock.ExpectQuery("select payload_1").WillReturnError(errors.New("MongoDB decoded batch byte limit exceeded")) // A pre-canceled context is rejected by database/sql before it reaches the // driver, so no sqlmock expectation is consumed here. expectMongoDBE2EScalar(mock, "5") @@ -112,6 +114,8 @@ func TestMongoDBLocalE2ERunContract(t *testing.T) { "secret-backed-ddl", "show-create-redaction-roundtrip", "scan-projection-pushdown-null-conversion", + "low-precision-temporal-residual", + "decoded-vector-budget-enforced", "multi-batch-cancel-recovery", "mongoscan-timewin-gapfill", "atomic-aggregate-watermark", From 2aefcf288f857f00b036a091e7dbac3c8ebfbadc Mon Sep 17 00:00:00 2001 From: iamlinjunhong <1030420200@qq.com> Date: Mon, 3 Aug 2026 14:32:55 +0800 Subject: [PATCH 5/6] test: cover MongoDB CI validation paths --- pkg/sql/colexec/mongoscan/mongoscan_test.go | 8 +- pkg/sql/compile/ddl.go | 8 -- pkg/sql/compile/ddl_test.go | 126 ++++++++++++++++++++ pkg/sql/plan/query_builder_test.go | 41 +++++++ 4 files changed, 174 insertions(+), 9 deletions(-) diff --git a/pkg/sql/colexec/mongoscan/mongoscan_test.go b/pkg/sql/colexec/mongoscan/mongoscan_test.go index d81fe016cbd35..e2588ad0532af 100644 --- a/pkg/sql/colexec/mongoscan/mongoscan_test.go +++ b/pkg/sql/colexec/mongoscan/mongoscan_test.go @@ -376,10 +376,12 @@ func TestMongoScanBatchAndStatementLimits(t *testing.T) { }) t.Run("decoded duplicated projection exceeds batch", func(t *testing.T) { + smallDoc, err := bson.Marshal(bson.D{{Key: "payload", Value: bson.Binary{Data: []byte("fits")}}}) + require.NoError(t, err) payload := make([]byte, 256<<10) doc, err := bson.Marshal(bson.D{{Key: "payload", Value: bson.Binary{Data: payload}}}) require.NoError(t, err) - cursor := &testCursor{docs: [][]byte{doc}} + cursor := &testCursor{docs: [][]byte{smallDoc, doc}} deps, _ := testScanDependencies(cursor) deps.Config.BatchRows = 10 deps.Config.MaxBatchBytes = 1 << 20 @@ -402,6 +404,10 @@ func TestMongoScanBatchAndStatementLimits(t *testing.T) { scan := NewArgument().WithScan(spec) scan.Dependencies = deps require.NoError(t, scan.Prepare(proc)) + first, err := scan.Call(proc) + require.NoError(t, err) + require.Equal(t, 1, first.Batch.RowCount()) + require.NotEmpty(t, scan.ctr.pendingRaw) _, err = scan.Call(proc) require.True(t, mongodb.IsDecodedBatchBudgetExceeded(err)) require.Equal(t, 1, cursor.closed) diff --git a/pkg/sql/compile/ddl.go b/pkg/sql/compile/ddl.go index 7dd501391b170..f7dc15ca85726 100644 --- a/pkg/sql/compile/ddl.go +++ b/pkg/sql/compile/ddl.go @@ -2236,14 +2236,6 @@ func (c *Compile) maybeDeleteMongoDBTableMapping(dbSource engine.Database, rel e if err != nil || !isMongoDB { return err } - createSQL := icebergCreateSQLFromPlanTableDef(tableDef) - _, found, err := sqlmongodb.ParseCreateSQLEnvelope(c.proc.Ctx, createSQL) - if err != nil { - return err - } - if !found { - return moerr.NewInternalError(c.proc.Ctx, "MongoDB external table is missing its catalog envelope") - } accountID, err := defines.GetAccountId(c.proc.Ctx) if err != nil { return err diff --git a/pkg/sql/compile/ddl_test.go b/pkg/sql/compile/ddl_test.go index 0f29d52171adf..50ead8026fb51 100644 --- a/pkg/sql/compile/ddl_test.go +++ b/pkg/sql/compile/ddl_test.go @@ -38,19 +38,145 @@ import ( "github.com/matrixorigin/matrixone/pkg/common/buffer" "github.com/matrixorigin/matrixone/pkg/common/moerr" "github.com/matrixorigin/matrixone/pkg/common/mpool" + moruntime "github.com/matrixorigin/matrixone/pkg/common/runtime" "github.com/matrixorigin/matrixone/pkg/defines" mock_frontend "github.com/matrixorigin/matrixone/pkg/frontend/test" plan2 "github.com/matrixorigin/matrixone/pkg/pb/plan" "github.com/matrixorigin/matrixone/pkg/pb/timestamp" "github.com/matrixorigin/matrixone/pkg/pb/txn" + "github.com/matrixorigin/matrixone/pkg/sql/features" + sqlmongodb "github.com/matrixorigin/matrixone/pkg/sql/mongodb" "github.com/matrixorigin/matrixone/pkg/sql/parsers/tree" "github.com/matrixorigin/matrixone/pkg/sql/plan" "github.com/matrixorigin/matrixone/pkg/testutil" + "github.com/matrixorigin/matrixone/pkg/util/executor" hnswruntime "github.com/matrixorigin/matrixone/pkg/vectorindex/hnsw/plugin/runtime" "github.com/matrixorigin/matrixone/pkg/vm/engine" "github.com/matrixorigin/matrixone/pkg/vm/process" ) +type mongoDBMappingTestExecutor struct { + results map[string]executor.Result + sqls []string +} + +func (e *mongoDBMappingTestExecutor) Exec( + _ context.Context, + sql string, + _ executor.Options, +) (executor.Result, error) { + e.sqls = append(e.sqls, sql) + return e.results[sql], nil +} + +func (*mongoDBMappingTestExecutor) ExecTxn( + context.Context, + func(executor.TxnExecutor) error, + executor.Options, +) error { + return nil +} + +func newMongoDBMappingTestCompile( + t *testing.T, + ctrl *gomock.Controller, + exec executor.SQLExecutor, +) (*Compile, *mock_frontend.MockDatabase, *mock_frontend.MockRelation) { + t.Helper() + proc := testutil.NewProcess(t) + ctx := defines.AttachAccountId(context.Background(), 7) + proc.Ctx = ctx + proc.ReplaceTopCtx(ctx) + moruntime.ServiceRuntime(proc.GetService()).SetGlobalVariables(moruntime.InternalSQLExecutor, exec) + return &Compile{proc: proc, pn: &plan2.Plan{}}, + mock_frontend.NewMockDatabase(ctrl), mock_frontend.NewMockRelation(ctrl) +} + +func mongoDBConnectionResult(t *testing.T, proc *process.Process, connectionID, disabled uint64) executor.Result { + t.Helper() + columnTypes := make([]types.Type, 18) + for i := range columnTypes { + columnTypes[i] = types.T_uint64.ToType() + } + result := executor.NewMemResult(columnTypes, proc.Mp()) + result.NewBatchWithRowCount(1) + require.NoError(t, executor.AppendFixedRows(result, 1, []uint64{connectionID})) + require.NoError(t, executor.AppendFixedRows(result, 17, []uint64{disabled})) + return result.GetResult() +} + +func TestMongoDBTableMappingDDLValidationAndPersistence(t *testing.T) { + mapping := sqlmongodb.TableMapping{ + Connection: "source", Database: "telemetry", Collection: "events", + SchemaMode: sqlmongodb.SchemaExplicit, Conversion: sqlmongodb.ConversionStrict, + Columns: []sqlmongodb.ColumnMapping{{ + Name: "value", Path: "value", TypeID: int32(types.T_int64), Conversion: sqlmongodb.ConversionStrict, + }}, + } + + t.Run("insert mapping", func(t *testing.T) { + ctrl := gomock.NewController(t) + exec := &mongoDBMappingTestExecutor{results: make(map[string]executor.Result)} + c, db, rel := newMongoDBMappingTestCompile(t, ctrl, exec) + db.EXPECT().GetDatabaseId(gomock.Any()).Return("8") + rel.EXPECT().GetTableID(gomock.Any()).Return(uint64(9)) + lookupSQL := sqlmongodb.GetConnectionByNameForUpdateSQL(7, mapping.Connection) + exec.results[lookupSQL] = mongoDBConnectionResult(t, c.proc, 42, 0) + qry := &plan2.CreateTable{TableDef: &plan2.TableDef{ + FeatureFlag: features.MongoDBExternal, + Createsql: sqlmongodb.BuildCreateSQLEnvelope(mapping), + }} + + require.NoError(t, c.maybeInsertMongoDBTableMapping(db, rel, qry)) + require.Len(t, exec.sqls, 2) + require.Equal(t, lookupSQL, exec.sqls[0]) + require.Contains(t, exec.sqls[1], "insert into mo_catalog."+sqlmongodb.TableMappings) + require.Contains(t, exec.sqls[1], "values (7,8,9,42") + require.Zero(t, c.proc.Mp().CurrNB()) + }) + + t.Run("typed insert requires envelope", func(t *testing.T) { + ctrl := gomock.NewController(t) + exec := &mongoDBMappingTestExecutor{} + c, _, _ := newMongoDBMappingTestCompile(t, ctrl, exec) + qry := &plan2.CreateTable{TableDef: &plan2.TableDef{FeatureFlag: features.MongoDBExternal}} + + err := c.maybeInsertMongoDBTableMapping(nil, nil, qry) + require.ErrorContains(t, err, "missing its catalog envelope") + require.Empty(t, exec.sqls) + }) + + t.Run("typed insert rejects malformed envelope", func(t *testing.T) { + ctrl := gomock.NewController(t) + exec := &mongoDBMappingTestExecutor{} + c, _, _ := newMongoDBMappingTestCompile(t, ctrl, exec) + qry := &plan2.CreateTable{TableDef: &plan2.TableDef{ + FeatureFlag: features.MongoDBExternal, + Createsql: "/* MO_MONGODB: version=2", + }} + + err := c.maybeInsertMongoDBTableMapping(nil, nil, qry) + require.ErrorContains(t, err, "envelope is not closed") + require.Empty(t, exec.sqls) + }) + + t.Run("delete mapping", func(t *testing.T) { + ctrl := gomock.NewController(t) + exec := &mongoDBMappingTestExecutor{} + c, db, rel := newMongoDBMappingTestCompile(t, ctrl, exec) + db.EXPECT().GetDatabaseId(gomock.Any()).Return("8") + rel.EXPECT().GetTableID(gomock.Any()).Return(uint64(9)) + tableDef := &plan2.TableDef{ + TableType: catalog.SystemExternalRel, + FeatureFlag: features.MongoDBExternal, + Createsql: sqlmongodb.BuildCreateSQLEnvelope(mapping), + } + + require.NoError(t, c.maybeDeleteMongoDBTableMapping(db, rel, tableDef)) + require.Equal(t, []string{sqlmongodb.DeleteTableMappingSQL(7, 8, 9)}, exec.sqls) + }) +} + func TestConvertDBEOBToNoSuchTable(t *testing.T) { err := convertDBEOBToNoSuchTable(context.Background(), moerr.GetOkExpectedEOB(), "db1", "t2") require.True(t, moerr.IsMoErrCode(err, moerr.ErrNoSuchTable)) diff --git a/pkg/sql/plan/query_builder_test.go b/pkg/sql/plan/query_builder_test.go index dedfcdf8dd4cf..c430981770dec 100644 --- a/pkg/sql/plan/query_builder_test.go +++ b/pkg/sql/plan/query_builder_test.go @@ -199,6 +199,47 @@ func TestMongoDBExternalScanPruningKeepsResidualColumnsAndPlansPushdown(t *testi require.Equal(t, "mo-residual:ff", scanNode.ExternScan.MongodbScan.ResidualFilterDigest) } +func TestMongoDBExternalScanRejectsInvalidCatalogState(t *testing.T) { + newMock := func(createSQL string) *MockOptimizer { + mock := NewMockOptimizer(false) + mock.ctxt.dbs["telemetry_source"] = true + mock.ctxt.objects["events_external"] = &plan.ObjectRef{ + DbName: "telemetry_source", ObjName: "events_external", Obj: 42, + } + mock.ctxt.tables["events_external"] = &plan.TableDef{ + Name: "events_external", TableType: catalog.SystemExternalRel, + FeatureFlag: features.MongoDBExternal, + Createsql: createSQL, + Cols: []*plan.ColDef{{ + Name: "value", Typ: plan.Type{Id: int32(types.T_int64)}, + }}, + } + return mock + } + mapping := sqlmongodb.TableMapping{ + Connection: "telemetry_source", Database: "telemetry", Collection: "events", + SchemaMode: sqlmongodb.SchemaExplicit, Conversion: sqlmongodb.ConversionStrict, + Columns: []sqlmongodb.ColumnMapping{{ + Name: "value", Path: "value", TypeID: int32(types.T_int64), Conversion: sqlmongodb.ConversionStrict, + }}, + } + + t.Run("malformed envelope", func(t *testing.T) { + mock := newMock("/* MO_MONGODB: version=2") + _, err := runOneStmt(mock, t, "select value from telemetry_source.events_external") + require.ErrorContains(t, err, "envelope is not closed") + }) + + t.Run("prepared scan", func(t *testing.T) { + mock := newMock(sqlmongodb.BuildCreateSQLEnvelope(mapping)) + stmts, err := parsers.Parse(mock.ctxt.GetContext(), dialect.MYSQL, + "select value from telemetry_source.events_external", 1) + require.NoError(t, err) + _, err = BuildPlan(&mock.ctxt, stmts[0], true) + require.ErrorContains(t, err, "prepared MongoDB external scans") + }) +} + func TestCanPruneSampleExprs(t *testing.T) { makeCol := func(tag, pos int32, notNullable bool) *plan.Expr { return &plan.Expr{ From ae43f4a90c3468e6b3385a9575df28ba1e6c3b0c Mon Sep 17 00:00:00 2001 From: iamlinjunhong <1030420200@qq.com> Date: Mon, 3 Aug 2026 15:22:31 +0800 Subject: [PATCH 6/6] test: isolate fuzzy runtime filter protocol state --- pkg/sql/plan/runtime_filter_test.go | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/pkg/sql/plan/runtime_filter_test.go b/pkg/sql/plan/runtime_filter_test.go index c35383022f6fd..7d24f1ed836d2 100644 --- a/pkg/sql/plan/runtime_filter_test.go +++ b/pkg/sql/plan/runtime_filter_test.go @@ -784,6 +784,24 @@ func TestSerializedExactRuntimeFilterPairContract(t *testing.T) { } func TestFinalizeFuzzyRuntimeFilterKeepsDecisionAtomic(t *testing.T) { + protocolProbe := newRuntimeFilterSingleTestBuilder(true) + rt := moruntime.ServiceRuntime( + protocolProbe.compCtx.GetProcess().GetService()) + original, hadOriginal := rt.GetGlobalVariables( + moruntime.MOProtocolVersion) + rt.SetGlobalVariables( + moruntime.MOProtocolVersion, defines.MORPCVersion8) + t.Cleanup(func() { + if hadOriginal { + rt.SetGlobalVariables( + moruntime.MOProtocolVersion, original) + } else { + rt.SetGlobalVariables( + moruntime.MOProtocolVersion, + defines.MORPCLatestVersion) + } + }) + newBuilder := func(tableCost, sinkCost float64) (*QueryBuilder, *planpb.Node, *planpb.Node, *planpb.Node) { builder := newRuntimeFilterSingleTestBuilder(true) tableScan := builder.qry.Nodes[0]