From 7242262e887dabbdd24703ef80b0f3de22d57a63 Mon Sep 17 00:00:00 2001 From: Lundomn <134036682+Lundomn@users.noreply.github.com> Date: Mon, 12 Oct 2026 13:35:25 +0800 Subject: [PATCH 1/4] test(partition): align TopN unsupported-order coverage with ANY NULL Fixes #29968 (cherry picked from commit fbc1abc9f7bfdda6d907904d03ef34cb4793eac8) --- pkg/sql/colexec/partition/topn_test.go | 62 ++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/pkg/sql/colexec/partition/topn_test.go b/pkg/sql/colexec/partition/topn_test.go index 86389fa8ec6a6..65f5e9ab7aa4d 100644 --- a/pkg/sql/colexec/partition/topn_test.go +++ b/pkg/sql/colexec/partition/topn_test.go @@ -159,6 +159,68 @@ func TestPartitionTopNRejectsUnsupportedOrderType(t *testing.T) { } } +func TestPartitionTopNNonLiteralAnyNullOrder(t *testing.T) { + proc := testutil.NewProcessWithOwnedMPool(t, "", mpool.MustNewZero()) + t.Cleanup(func() { require.Zero(t, proc.Mp().CurrNB()) }) + t.Cleanup(proc.Free) + arg := newTopNArgument(1) + arg.OrderBySpecs[1].Expr = topNCol(1, types.T_any) + t.Cleanup(func() { arg.Free(proc, false, nil) }) + + for generation := 0; generation < 2; generation++ { + batches := []*batch.Batch{ + makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{10, 20}, false), + makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{11, 21}, true), + } + child := colexec.NewMockOperator().WithBatchs(batches) + t.Cleanup(func() { child.Free(proc, false, nil) }) + arg.Children = nil + arg.AppendChild(child) + + require.NoError(t, arg.Prepare(proc)) + require.Len(t, arg.top.compares, 2) + require.NotNil(t, arg.top.compares[0]) + + counts := make(map[int32]int) + validPayload := map[int32][]int64{1: {10, 11}, 2: {20, 21}} + for { + result, err := arg.Call(proc) + require.NoError(t, err) + if result.Batch == nil || result.Status == vm.ExecStop { + break + } + groups := vector.MustFixedColWithTypeCheck[int32](result.Batch.Vecs[0]) + payloads := vector.MustFixedColWithTypeCheck[int64](result.Batch.Vecs[2]) + for row, group := range groups { + require.True(t, result.Batch.Vecs[1].IsNull(uint64(row))) + require.Contains(t, validPayload[group], payloads[row]) + counts[group]++ + } + } + require.Equal(t, map[int32]int{1: 1, 2: 1}, counts) + arg.Reset(proc, false, nil) + } +} + +func makeAnyNullTopNBatch(t *testing.T, proc *process.Process, groups []int32, payloads []int64, constant bool) *batch.Batch { + t.Helper() + require.Len(t, groups, len(payloads)) + bat := batch.NewWithSize(3) + bat.Vecs[0] = vector.NewVec(types.T_int32.ToType()) + require.NoError(t, vector.AppendFixedList(bat.Vecs[0], groups, nil, proc.Mp())) + if constant { + bat.Vecs[1] = vector.NewConstNull(types.T_any.ToType(), len(groups), proc.Mp()) + } else { + bat.Vecs[1] = vector.NewVec(types.T_any.ToType()) + bat.Vecs[1].SetLength(len(groups)) + nulls.AddRange(bat.Vecs[1].GetNulls(), 0, uint64(len(groups))) + } + bat.Vecs[2] = vector.NewVec(types.T_int64.ToType()) + require.NoError(t, vector.AppendFixedList(bat.Vecs[2], payloads, nil, proc.Mp())) + bat.SetRowCount(len(groups)) + return bat +} + func TestPartitionTopNSQLOrderFloatNaNs(t *testing.T) { for _, tc := range []struct { name string From eebf42fa3823e48bcc82a024f936e77ce61d76af Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Mon, 12 Oct 2026 02:21:57 -0400 Subject: [PATCH 2/4] test: fix TERM readiness and avoid ephemeral listener-port collisions --- optools/ut_schedule_test.go | 24 +++++------ pkg/logservice/test_ports_test.go | 47 +++++++++++++++++++- pkg/logservice/test_utils.go | 59 ++++++++++++++++++++++++-- pkg/sql/colexec/partition/topn_test.go | 15 ++++--- 4 files changed, 122 insertions(+), 23 deletions(-) diff --git a/optools/ut_schedule_test.go b/optools/ut_schedule_test.go index f59eee57481a7..6d5a6d589bca5 100644 --- a/optools/ut_schedule_test.go +++ b/optools/ut_schedule_test.go @@ -1533,19 +1533,26 @@ read -r observed <&8 wait "$child_pid" printf 'CHILD %s %s\n' "$observed" "$child_pid" handled=0 -ack_pending=0 -trap 'fixture_self_pid trap_pid; [[ "$trap_pid" == "$main" ]] || exit 97; handled=$((handled+1)); ack_pending=1' TERM +pending_ack=0 +trap 'fixture_self_pid trap_pid; [[ "$trap_pid" == "$main" ]] || exit 97; handled=$((handled+1)); pending_ack=1' TERM ( + # Unblock the main loop on protocol failure as well as normal completion. + trap ': > "$scratch/done"' EXIT for ((j=0; j<200; j++)); do kill -TERM "$main" - read -r acknowledgment <&7 - [[ "$acknowledgment" == ack ]] || exit 98 + read -r acknowledgment sequence pending <&7 + [[ "$acknowledgment" == ack && "$sequence" == "$((j+1))" && "$pending" == 0 ]] || exit 98 done - touch "$scratch/done" ) & sender=$! trap 'kill "$sender" 2>/dev/null || :; wait "$sender" 2>/dev/null || :' EXIT while [[ ! -f "$scratch/done" ]]; do + # ACK means the previous trap has returned, so the sender can safely issue + # the next TERM. Publishing it inside the trap can lose that next signal. + if (( pending_ack )); then + pending_ack=0 + printf 'ack %s %s\n' "$handled" "$pending_ack" >&7 + fi # Reenter BASHPID expansion in modern Bash. The legacy fallback spawns # a child, whose output can be interrupted by TERM; its PID is checked # at startup, in the subshell, and by each TERM handler instead. @@ -1553,13 +1560,6 @@ while [[ ! -f "$scratch/done" ]]; do fixture_self_pid observed [[ "$observed" == "$main" ]] || exit 99 fi - # Acknowledge only after the handler returns. Otherwise the sender can - # deliver the next TERM while TERM is still blocked inside the handler, - # and Bash can coalesce it with the signal already being handled. - if (( ack_pending )); then - ack_pending=0 - printf 'ack\n' >&7 - fi done wait "$sender" trap - EXIT diff --git a/pkg/logservice/test_ports_test.go b/pkg/logservice/test_ports_test.go index 26f9ecef42dbe..d2b33ef54553e 100644 --- a/pkg/logservice/test_ports_test.go +++ b/pkg/logservice/test_ports_test.go @@ -37,7 +37,7 @@ func TestTestPortAllocation(t *testing.T) { t.Run(tc.name, func(t *testing.T) { a := allocatedPorts{ports: make(map[int]struct{})} calls := 0 - port, err := a.allocate(func(int) error { + port, err := a.allocate(firstTestPort, 32768, func(int) error { calls++ if calls <= tc.failures { return &net.OpError{Op: "listen", Err: tc.cause} @@ -123,3 +123,48 @@ func listenUDPOnTCPAvailablePort(t *testing.T) (net.PacketConn, int) { require.FailNow(t, "could not reserve a UDP port that is also available to TCP") return nil, 0 } + +func TestTestPortRangeOutsideEphemeral(t *testing.T) { + for _, tc := range []struct { + name, value string + first, limit int + }{ + {"linux default", "32768\t60999\n", 21024, 32768}, + {"macos default", "49152 65535", 21024, 49152}, + {"only above", "20000 40000", 40001, 65535}, + {"only below", "25000 65535", 21024, 25000}, + {"below candidates", "1024 20000", 21024, 65535}, + {"last port", "65535 65535", 21024, 65535}, + {"no candidates", "1024 65535", 0, 0}, + {"reversed", "61000 32768", 0, 0}, + {"missing", "32768", 0, 0}, + {"extra", "32768 60999 0", 0, 0}, + {"not numeric", "low high", 0, 0}, + {"zero", "0 60999", 0, 0}, + {"too high", "32768 65536", 0, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + first, limit, err := testPortRangeOutsideEphemeral(tc.value) + if tc.first == 0 { + require.Error(t, err) + require.Zero(t, first) + require.Zero(t, limit) + return + } + require.NoError(t, err) + require.Equal(t, tc.first, first) + require.Equal(t, tc.limit, limit) + a := allocatedPorts{ports: make(map[int]struct{})} + for range 100 { + port, err := a.allocate(first, limit, func(candidate int) error { + require.GreaterOrEqual(t, candidate, tc.first) + require.Less(t, candidate, tc.limit) + return nil + }) + require.NoError(t, err) + require.GreaterOrEqual(t, port, first) + require.Less(t, port, limit) + } + }) + } +} diff --git a/pkg/logservice/test_utils.go b/pkg/logservice/test_utils.go index 8ca0c9a902f16..fdacd7061c609 100644 --- a/pkg/logservice/test_utils.go +++ b/pkg/logservice/test_utils.go @@ -20,6 +20,10 @@ import ( "fmt" "math/rand" "net" + "os" + stdRuntime "runtime" + "strconv" + "strings" "sync" "syscall" "testing" @@ -61,20 +65,67 @@ var randomPorts = allocatedPorts{ } func getAvailablePort() int { - port, err := randomPorts.allocate(probeTestPort) + first, limit, err := testPortRange() + if err != nil { + panic(err) + } + port, err := randomPorts.allocate(first, limit, probeTestPort) if err != nil { panic(err) // Preserve the existing fixture API, but fail promptly with the cause. } return port } -const maxPortAllocationAttempts = 128 +const ( + maxPortAllocationAttempts = 128 + firstTestPort = 21024 + testPortLimit = 65535 +) + +// Probing a port does not reserve it. Keep candidate listener ports outside +// the automatic client-port range so outbound traffic cannot claim them in +// the gap between the probe and service startup. Explicit binds can still +// race that gap; they are not coordinated by the process-local used-port map. +func testPortRange() (int, int, error) { + if stdRuntime.GOOS != "linux" { + // Conservative for the default macOS dynamic-port range (49152+). + // Unlike Linux, customized non-Linux ranges are not inspected here. + return firstTestPort, 32768, nil + } + data, err := os.ReadFile("/proc/sys/net/ipv4/ip_local_port_range") + if err != nil { + return 0, 0, &testPortAllocationError{message: "read automatic client-port range", cause: err} + } + return testPortRangeOutsideEphemeral(string(data)) +} + +func testPortRangeOutsideEphemeral(value string) (int, int, error) { + fields := strings.Fields(value) + if len(fields) == 2 { + low, lowErr := strconv.Atoi(fields[0]) + high, highErr := strconv.Atoi(fields[1]) + if lowErr == nil && highErr == nil && low > 0 && low <= high && high <= 65535 { + belowLimit := max(firstTestPort, min(testPortLimit, low)) + aboveFirst := min(testPortLimit, max(firstTestPort, high+1)) + // Choose the larger contiguous interval; neither includes the + // configured inclusive ephemeral range. + if belowLimit-firstTestPort >= testPortLimit-aboveFirst && belowLimit > firstTestPort { + return firstTestPort, belowLimit, nil + } + if aboveFirst < testPortLimit { + return aboveFirst, testPortLimit, nil + } + return 0, 0, &testPortAllocationError{message: "automatic client-port range leaves no test listener ports"} + } + } + return 0, 0, &testPortAllocationError{message: fmt.Sprintf("invalid automatic client-port range %q", value)} +} -func (a *allocatedPorts) allocate(probe func(int) error) (int, error) { +func (a *allocatedPorts) allocate(first, limit int, probe func(int) error) (int, error) { a.Lock() defer a.Unlock() for range maxPortAllocationAttempts { - port := rand.Intn(65535-21024) + 21024 + port := rand.Intn(limit-first) + first if _, exists := a.ports[port]; exists { continue } diff --git a/pkg/sql/colexec/partition/topn_test.go b/pkg/sql/colexec/partition/topn_test.go index 65f5e9ab7aa4d..a79b4e696990c 100644 --- a/pkg/sql/colexec/partition/topn_test.go +++ b/pkg/sql/colexec/partition/topn_test.go @@ -168,12 +168,10 @@ func TestPartitionTopNNonLiteralAnyNullOrder(t *testing.T) { t.Cleanup(func() { arg.Free(proc, false, nil) }) for generation := 0; generation < 2; generation++ { - batches := []*batch.Batch{ - makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{10, 20}, false), - makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{11, 21}, true), - } - child := colexec.NewMockOperator().WithBatchs(batches) + child := colexec.NewMockOperator() t.Cleanup(func() { child.Free(proc, false, nil) }) + child.WithBatchs([]*batch.Batch{makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{10, 20}, false)}) + child.WithBatchs([]*batch.Batch{makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{11, 21}, true)}) arg.Children = nil arg.AppendChild(child) @@ -202,10 +200,15 @@ func TestPartitionTopNNonLiteralAnyNullOrder(t *testing.T) { } } -func makeAnyNullTopNBatch(t *testing.T, proc *process.Process, groups []int32, payloads []int64, constant bool) *batch.Batch { +func makeAnyNullTopNBatch(t *testing.T, proc *process.Process, groups []int32, payloads []int64, constant bool) (result *batch.Batch) { t.Helper() require.Len(t, groups, len(payloads)) bat := batch.NewWithSize(3) + defer func() { + if result == nil { + bat.Clean(proc.Mp()) + } + }() bat.Vecs[0] = vector.NewVec(types.T_int32.ToType()) require.NoError(t, vector.AppendFixedList(bat.Vecs[0], groups, nil, proc.Mp())) if constant { From c38e2e50dbfec709d48e3391d7a9c39c0b37abd4 Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Mon, 12 Oct 2026 02:26:19 -0400 Subject: [PATCH 3/4] test(logservice): retain configured-port read failure diagnostics --- pkg/logservice/test_utils.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/logservice/test_utils.go b/pkg/logservice/test_utils.go index fdacd7061c609..ddfe316e07b2a 100644 --- a/pkg/logservice/test_utils.go +++ b/pkg/logservice/test_utils.go @@ -94,7 +94,7 @@ func testPortRange() (int, int, error) { } data, err := os.ReadFile("/proc/sys/net/ipv4/ip_local_port_range") if err != nil { - return 0, 0, &testPortAllocationError{message: "read automatic client-port range", cause: err} + return 0, 0, &testPortAllocationError{message: fmt.Sprintf("read automatic client-port range: %v", err), cause: err} } return testPortRangeOutsideEphemeral(string(data)) } From 9c45883ae41ba1d8192463e59cbb3c908baf1126 Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Tue, 13 Oct 2026 05:48:57 +0800 Subject: [PATCH 4/4] test(partition): consolidate rebased TopN NULL fixtures --- pkg/sql/colexec/partition/topn_test.go | 80 +++----------------------- 1 file changed, 9 insertions(+), 71 deletions(-) diff --git a/pkg/sql/colexec/partition/topn_test.go b/pkg/sql/colexec/partition/topn_test.go index a79b4e696990c..2ecd552160554 100644 --- a/pkg/sql/colexec/partition/topn_test.go +++ b/pkg/sql/colexec/partition/topn_test.go @@ -159,71 +159,6 @@ func TestPartitionTopNRejectsUnsupportedOrderType(t *testing.T) { } } -func TestPartitionTopNNonLiteralAnyNullOrder(t *testing.T) { - proc := testutil.NewProcessWithOwnedMPool(t, "", mpool.MustNewZero()) - t.Cleanup(func() { require.Zero(t, proc.Mp().CurrNB()) }) - t.Cleanup(proc.Free) - arg := newTopNArgument(1) - arg.OrderBySpecs[1].Expr = topNCol(1, types.T_any) - t.Cleanup(func() { arg.Free(proc, false, nil) }) - - for generation := 0; generation < 2; generation++ { - child := colexec.NewMockOperator() - t.Cleanup(func() { child.Free(proc, false, nil) }) - child.WithBatchs([]*batch.Batch{makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{10, 20}, false)}) - child.WithBatchs([]*batch.Batch{makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{11, 21}, true)}) - arg.Children = nil - arg.AppendChild(child) - - require.NoError(t, arg.Prepare(proc)) - require.Len(t, arg.top.compares, 2) - require.NotNil(t, arg.top.compares[0]) - - counts := make(map[int32]int) - validPayload := map[int32][]int64{1: {10, 11}, 2: {20, 21}} - for { - result, err := arg.Call(proc) - require.NoError(t, err) - if result.Batch == nil || result.Status == vm.ExecStop { - break - } - groups := vector.MustFixedColWithTypeCheck[int32](result.Batch.Vecs[0]) - payloads := vector.MustFixedColWithTypeCheck[int64](result.Batch.Vecs[2]) - for row, group := range groups { - require.True(t, result.Batch.Vecs[1].IsNull(uint64(row))) - require.Contains(t, validPayload[group], payloads[row]) - counts[group]++ - } - } - require.Equal(t, map[int32]int{1: 1, 2: 1}, counts) - arg.Reset(proc, false, nil) - } -} - -func makeAnyNullTopNBatch(t *testing.T, proc *process.Process, groups []int32, payloads []int64, constant bool) (result *batch.Batch) { - t.Helper() - require.Len(t, groups, len(payloads)) - bat := batch.NewWithSize(3) - defer func() { - if result == nil { - bat.Clean(proc.Mp()) - } - }() - bat.Vecs[0] = vector.NewVec(types.T_int32.ToType()) - require.NoError(t, vector.AppendFixedList(bat.Vecs[0], groups, nil, proc.Mp())) - if constant { - bat.Vecs[1] = vector.NewConstNull(types.T_any.ToType(), len(groups), proc.Mp()) - } else { - bat.Vecs[1] = vector.NewVec(types.T_any.ToType()) - bat.Vecs[1].SetLength(len(groups)) - nulls.AddRange(bat.Vecs[1].GetNulls(), 0, uint64(len(groups))) - } - bat.Vecs[2] = vector.NewVec(types.T_int64.ToType()) - require.NoError(t, vector.AppendFixedList(bat.Vecs[2], payloads, nil, proc.Mp())) - bat.SetRowCount(len(groups)) - return bat -} - func TestPartitionTopNSQLOrderFloatNaNs(t *testing.T) { for _, tc := range []struct { name string @@ -836,12 +771,10 @@ func TestPartitionTopNNonLiteralAnyNullOrder(t *testing.T) { t.Cleanup(func() { arg.Free(proc, false, nil) }) for generation := 0; generation < 2; generation++ { - batches := []*batch.Batch{ - makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{10, 20}, false), - makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{11, 21}, true), - } - child := colexec.NewMockOperator().WithBatchs(batches) + child := colexec.NewMockOperator() t.Cleanup(func() { child.Free(proc, false, nil) }) + child.WithBatchs([]*batch.Batch{makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{10, 20}, false)}) + child.WithBatchs([]*batch.Batch{makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{11, 21}, true)}) arg.Children = nil arg.AppendChild(child) @@ -870,10 +803,15 @@ func TestPartitionTopNNonLiteralAnyNullOrder(t *testing.T) { } } -func makeAnyNullTopNBatch(t *testing.T, proc *process.Process, groups []int32, payloads []int64, constant bool) *batch.Batch { +func makeAnyNullTopNBatch(t *testing.T, proc *process.Process, groups []int32, payloads []int64, constant bool) (result *batch.Batch) { t.Helper() require.Len(t, groups, len(payloads)) bat := batch.NewWithSize(3) + defer func() { + if result == nil { + bat.Clean(proc.Mp()) + } + }() bat.Vecs[0] = vector.NewVec(types.T_int32.ToType()) require.NoError(t, vector.AppendFixedList(bat.Vecs[0], groups, nil, proc.Mp())) if constant {