perf: avoid metadata simulation for ordinary fixed MIN/MAX values

pkg/sql/colexec/aggexec/aggexec_test.go

@@ -282,6 +282,52 @@ func TestAggExecSpillDecodeUsesExactSourceCapacity(t *testing.T) {
 	results[0].Free(mp)
 	destination.Free()
 	target.Free()
+
+	t.Run("fixed-max-plain", func(t *testing.T) {
+		f := newFixedMetadataPreflightFixture(t, AggIdOfMax, types.T_float64.ToType())
+		input := vector.NewVec(types.T_float64.ToType())
+		t.Cleanup(func() { input.Free(f.mp) })
+		require.NoError(t, vector.AppendFixedList(input, []float64{7, 4, 0, 9}, []bool{false, false, true, false}, f.mp))
+		require.NoError(t, f.exec.GroupGrow(3))
+		groups := []uint64{1, 1, 2, 3}
+		require.NoError(t, f.exec.PreflightBatchFill(0, groups, []*vector.Vector{input}))
+		require.NoError(t, f.exec.BatchFill(0, groups, []*vector.Vector{input}))
+		var encoded bytes.Buffer
+		require.NoError(t, f.exec.(SpillStateCodec).SaveSpillIntermediateRows(0, []int32{0, 1, 2}, &encoded))
+		decoded := makeMinMaxExec(f.mp, AggIdOfMax, false, types.T_float64.ToType())
+		t.Cleanup(func() {
+			if decoded != nil {
+				decoded.Free()
+			}
+		})
+		require.NoError(t, decoded.(SpillStateCodec).UnmarshalSpillFromReader(bytes.NewReader(encoded.Bytes()), f.mp))
+		decodedState := decoded.(aggregateBaseCarrier).aggregateBase().state[0].vecs[0]
+		require.Equal(t, 3, decodedState.Capacity())
+		require.Equal(t, float64(7), vector.GetFixedAtNoTypeCheck[float64](decodedState, 0))
+		require.Equal(t, float64(9), vector.GetFixedAtNoTypeCheck[float64](decodedState, 2))
+		require.True(t, decodedState.IsNull(1))
+		var before, after bytes.Buffer
+		require.NoError(t, decoded.(SpillStateCodec).SaveSpillIntermediateRows(0, []int32{0, 1, 2}, &before))
+		merged := makeMinMaxExec(f.mp, AggIdOfMax, false, types.T_float64.ToType())
+		t.Cleanup(func() { merged.Free() })
+		require.NoError(t, merged.GroupGrow(3))
+		require.NoError(t, merged.BatchMerge(decoded, 0, []uint64{1, 2, 3}))
+		require.NoError(t, decoded.(SpillStateCodec).SaveSpillIntermediateRows(0, []int32{0, 1, 2}, &after))
+		require.Equal(t, before.Bytes(), after.Bytes(), "merge source must remain immutable")
+		decoded.Free()
+		decoded = nil // consumer owns its result after the compact source is freed
+		results, err := merged.Flush()
+		require.NoError(t, err)
+		require.Len(t, results, 1)
+		t.Cleanup(func() { results[0].Free(f.mp) })
+		require.Equal(t, float64(7), vector.GetFixedAtNoTypeCheck[float64](results[0], 0))
+		require.Equal(t, float64(9), vector.GetFixedAtNoTypeCheck[float64](results[0], 2))
+		for row := range 3 {
+			require.Equal(t, row == 1, results[0].IsNull(uint64(row)))
+			require.Equal(t, vector.PrepareParamNone, results[0].GetPrepareParamKindAt(row))
+			require.Equal(t, types.StringSourceExpression, results[0].GetStringSourceAt(row))
+		}
+	})
 }
 
 func TestAggExecSpillDecodeReusesLargeSmallLargeCapacity(t *testing.T) {

pkg/sql/colexec/aggexec/capacity_preflight.go

@@ -2367,6 +2367,11 @@ func (exec *minMaxExecFixed[T]) preflightFixedStringSources(
 	return exec.applyStringSourceEvents(&events, eventCount)
 }
 
+func hasDefaultFixedMinMaxMetadata(vec *vector.Vector) bool {
+	return vec.GetPrepareParamKind() == vector.PrepareParamNone &&
+		vec.GetPrepareParamKinds() == nil && !vec.HasStringSourceMetadata()
+}
+
 func (exec *minMaxExecFixed[T]) PreflightBatchFill(
 	offset int,
 	groups []uint64,
@@ -2382,6 +2387,40 @@ func (exec *minMaxExecFixed[T]) PreflightBatchFill(
 		return err
 	}
 	values := vector.MustFixedColNoTypeCheck[T](vectors[0])
+	if hasDefaultFixedMinMaxMetadata(vectors[0]) {
+		// Ordinary fixed values cannot create a provenance sidecar when every
+		// touched destination is ordinary too. Validate the same active/standby
+		// targets without simulating winners or recording metadata events.
+		plain := true
+		for row, group := range groups {
+			if group == GroupNotMatched {
+				continue
+			}
+			physicalRow, err := preflightPhysicalRow(vectors[0], offset+row)
+			if err != nil {
+				return err
+			}
+			if vectors[0].IsNull(uint64(physicalRow)) {
+				continue
+			}
+			_ = values[physicalRow]
+			_, y, state, err := exec.validatePreflightTarget(group)
+			if err != nil {
+				return err
+			}
+			// Preserve the original metadata preflight's shape errors as well
+			// as its allocation checks; an incomplete target must fall back.
+			if len(state.vecs) == 0 || state.vecs[0] == nil ||
+				max(int(state.length), int(y)+1) < state.vecs[0].Length() ||
+				!hasDefaultFixedMinMaxMetadata(state.vecs[0]) {
+				plain = false
+				break
+			}
+		}
+		if plain {
+			return nil
+		}
+	}
 	candidate := func(row int) (T, vector.PrepareParamKind, bool, error) {
 		physicalRow, err := preflightPhysicalRow(vectors[0], offset+row)
 		if err != nil {