perf(partition): reduce consecutive Top-1 candidates before copying

pkg/sql/colexec/partition/topn.go

@@ -284,6 +284,7 @@ func (ctr *topNContainer) consume(proc *process.Process, input *batch.Batch) err
 	}
 	defer freeNormalizedHashPartitionKeys(ownedPartitionKeys, proc.Mp())
 
+	reduceSingleRuns := ctr.limit == 1 && !ctr.withTies
 	for start := 0; start < input.RowCount(); start += hashmap.UnitLimit {
 		if err, canceled := vm.CancelCheck(proc); canceled {
 			return err
@@ -293,11 +294,25 @@ func (ctr *topNContainer) consume(proc *process.Process, input *batch.Batch) err
 		if err != nil {
 			return err
 		}
-		for offset, groupID := range groupIDs[:count] {
+		for offset := 0; offset < count; offset++ {
+			groupID := groupIDs[offset]
 			for len(ctr.groups) < int(groupID) {
 				ctr.groups = append(ctr.groups, &partitionHeap{ctr: ctr, tieHead: -1})
 			}
 			row := int64(start + offset)
+			if reduceSingleRuns {
+				// Pick the first best row in this contiguous group run before
+				// copying it. Increasing input otherwise replaces the retained
+				// row (including all payload metadata) for every incoming row.
+				// The candidate never outlives this hash chunk or input batch.
+				for offset+1 < count && groupIDs[offset+1] == groupID {
+					offset++
+					candidate := int64(start + offset)
+					if ctr.compareInputRows(candidate, row) < 0 {
+						row = candidate
+					}
+				}
+			}
 			groupHeap := ctr.groups[groupID-1]
 			if uint64(groupHeap.Len()) < ctr.limit {
 				slot, err := ctr.retainRow(proc, input, row)
@@ -481,6 +496,15 @@ func (ctr *topNContainer) compareInput(inputRow, retainedRow int64) int {
 	return 0
 }
 
+func (ctr *topNContainer) compareInputRows(left, right int64) int {
+	for _, cmp := range ctr.compares {
+		if result := cmp.Compare(1, 1, left, right); result != 0 {
+			return result
+		}
+	}
+	return 0
+}
+
 func (ctr *topNContainer) compareRetained(left, right int64) int {
 	for _, cmp := range ctr.compares {
 		if result := cmp.Compare(0, 0, left, right); result != 0 {

pkg/sql/colexec/partition/topone_null_payload_test.go

@@ -0,0 +1,73 @@
+// Copyright 2026 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package partition
+
+import (
+	"testing"
+
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
+	"github.com/matrixorigin/matrixone/pkg/testutil"
+	"github.com/matrixorigin/matrixone/pkg/vm"
+	"github.com/stretchr/testify/require"
+)
+
+// Latest timestamp, rather than latest non-NULL metric, defines the winner.
+// The last input row replaces a non-NULL retained payload across both an input
+// batch and a hash chunk boundary. Reuse also checks NULL -> non-NULL cleanup.
+func TestPartitionTopOneRunsLatestNullPayload(t *testing.T) {
+	proc := testutil.NewProcessWithOwnedMPool(t, "", mpool.MustNewZero())
+	t.Cleanup(func() { require.Zero(t, proc.Mp().CurrNB()) })
+	t.Cleanup(proc.Free)
+	arg := topOneRunArgument(true, true)
+	arg.OrderBySpecs = arg.OrderBySpecs[:2] // metric payload is not an ORDER key
+	t.Cleanup(func() { arg.Free(proc, false, nil) })
+	for _, latestNull := range []bool{true, false} {
+		child := colexec.NewMockOperator()
+		t.Cleanup(func() { child.Free(proc, false, nil) })
+		arg.Children = nil
+		arg.AppendChild(child)
+		rows := make([]topOneRunRow, 514)
+		for i := range rows {
+			rows[i] = topOneRunRow{group: 1, score: int64(i), secondary: int64(i + 10), id: int64(i)}
+		}
+		rows[513].secondaryNull = latestNull
+		if latestNull {
+			rows[513].secondary = 0
+		}
+		child.WithBatchs([]*batch.Batch{
+			topOneRunBatch(t, proc, rows[:256]),
+			topOneRunBatch(t, proc, rows[256:]),
+		})
+		require.NoError(t, arg.Prepare(proc))
+		result, err := arg.Call(proc)
+		require.NoError(t, err)
+		require.NotNil(t, result.Batch)
+		require.Equal(t, 1, result.Batch.RowCount())
+		require.Equal(t, int64(513), vector.GetFixedAtWithTypeCheck[int64](result.Batch.Vecs[1], 0))
+		require.Equal(t, int64(513), vector.GetFixedAtWithTypeCheck[int64](result.Batch.Vecs[3], 0))
+		require.Equal(t, latestNull, result.Batch.Vecs[2].IsNull(0))
+		if !latestNull {
+			require.Equal(t, int64(523), vector.GetFixedAtWithTypeCheck[int64](result.Batch.Vecs[2], 0))
+		}
+		result, err = arg.Call(proc)
+		require.NoError(t, err)
+		require.True(t, result.Batch == nil || result.Status == vm.ExecStop)
+		arg.Reset(proc, false, nil)
+		child.Free(proc, false, nil)
+	}
+}

pkg/sql/colexec/partition/topone_runs_bench_test.go

@@ -0,0 +1,114 @@
+// Copyright 2026 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package partition
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/container/batch"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/sql/colexec"
+	"github.com/matrixorigin/matrixone/pkg/testutil"
+	"github.com/matrixorigin/matrixone/pkg/vm"
+	"github.com/stretchr/testify/require"
+)
+
+// Exercise the same operator/input on both revisions. GroupedIncreasing is the
+// replacement-heavy case; the other layouts expose neutral/negative cases and
+// K2 checks that the general heap path remains unchanged. These microbenchmarks
+// supplement (rather than substitute for) the complete Q4 native-client run.
+func BenchmarkPartitionTopOneRuns(b *testing.B) {
+	const rowCount = 65536
+	for _, tc := range []struct {
+		name   string
+		layout string
+		limit  uint64
+	}{
+		{"GroupedIncreasingK1", "increasing", 1},
+		{"GroupedRandomK1", "random", 1},
+		{"InterleavedK1", "interleaved", 1},
+		{"EveryRowNewGroupK1", "unique", 1},
+		{"GroupedIncreasingK2", "increasing", 2},
+	} {
+		b.Run(tc.name, func(b *testing.B) {
+			rows := make([]topOneRunRow, rowCount)
+			groupCount := rowCount / 1024
+			if tc.layout == "unique" {
+				groupCount = rowCount
+			}
+			for i := range rows {
+				group, score := i/1024, i%1024
+				switch tc.layout {
+				case "random":
+					score = (i*7919 + 17) % 1024
+				case "interleaved":
+					group = i % groupCount
+				case "unique":
+					group = i
+				}
+				rows[i] = topOneRunRow{group: int32(group), score: int64(score), id: int64(i)}
+			}
+			payload := []byte(strings.Repeat("payload", 12))
+			b.ReportAllocs()
+			b.ResetTimer()
+			for iteration := 0; iteration < b.N; iteration++ {
+				b.StopTimer()
+				proc := testutil.NewProcessWithOwnedMPool(b, "", mpool.MustNewZero())
+				child := colexec.NewMockOperator()
+				arg := topOneRunArgument(true, true)
+				arg.Limit = newTopNArgument(tc.limit).Limit
+				arg.PreReduce = true // emit packed groups, keeping client overhead small
+				arg.AppendChild(child)
+				func() {
+					defer func() {
+						arg.Free(proc, false, nil)
+						child.Free(proc, false, nil)
+						proc.Free()
+						require.Zero(b, proc.Mp().CurrNB())
+					}()
+					// Batches exceed UnitLimit and do not divide the 1024-row
+					// group runs, so both chunk and input boundaries are exercised.
+					for start := 0; start < rowCount; start += 8191 {
+						bat := topOneRunBatch(b, proc, rows[start:min(start+8191, rowCount)])
+						child.WithBatchs([]*batch.Batch{bat})
+						vec := vector.NewVec(types.T_varchar.ToType())
+						bat.Vecs = append(bat.Vecs, vec)
+						for row := 0; row < bat.RowCount(); row++ {
+							require.NoError(b, vector.AppendBytes(vec, payload, false, proc.Mp()))
+						}
+					}
+					require.NoError(b, arg.Prepare(proc))
+					b.StartTimer()
+					outputRows := 0
+					for {
+						result, err := arg.Call(proc)
+						require.NoError(b, err)
+						if result.Batch == nil || result.Status == vm.ExecStop {
+							break
+						}
+						outputRows += result.Batch.RowCount()
+					}
+					b.StopTimer()
+					wantRows := groupCount * int(tc.limit)
+					require.Equal(b, wantRows, outputRows)
+					b.ReportMetric(float64(outputRows), "output_rows")
+				}()
+			}
+		})
+	}
+}