sql: implement the any_value() aggregate function

docs/generated/sql/aggregates.md

@@ -1,6 +1,8 @@
 <table>
 <thead><tr><th>Function &rarr; Returns</th><th>Description</th><th>Volatility</th></tr></thead>
 <tbody>
+<tr><td><a name="any_value"></a><code>any_value(arg1: anyelement) &rarr; anyelement</code></td><td><span class="funcdesc"><p>Returns an arbitrary non-NULL value, or NULL if there are no non-NULL values.</p>
+</span></td><td>Immutable</td></tr>
 <tr><td><a name="array_agg"></a><code>array_agg(arg1: <a href="bool.html">bool</a>) &rarr; <a href="bool.html">bool</a>[]</code></td><td><span class="funcdesc"><p>Aggregates the selected values into an array.</p>
 </span></td><td>Immutable</td></tr>
 <tr><td><a name="array_agg"></a><code>array_agg(arg1: <a href="bool.html">bool</a>[]) &rarr; <a href="bool.html">bool</a>[][]</code></td><td><span class="funcdesc"><p>Aggregates the selected values into an array.</p>

pkg/sql/colexec/colbuilder/execplan.go

@@ -207,6 +207,12 @@ func supportedNatively(core *execinfrapb.ProcessorCoreUnion) error {
 				if !colexecagg.IsAggOptimized(*wf.Func.AggregateFunc) {
 					return errDefaultAggregateWindowFunction
 				}
+				if *wf.Func.AggregateFunc == execinfrapb.AnyNotNull {
+					// AnyNotNull has optimized hash and ordered aggregate
+					// implementations, but no window variant, so it must be
+					// run through the row-based windower.
+					return errAnyNotNullWindowFunction
+				}
 			}
 		}
 		return nil
@@ -245,6 +251,7 @@ var (
 	errNonInnerMergeJoinWithOnExpr    = errors.New("can't plan vectorized non-inner merge joins with ON expressions")
 	errWindowFunctionFilterClause     = errors.New("window functions with FILTER clause are not supported")
 	errDefaultAggregateWindowFunction = errors.New("default aggregate window functions not supported")
+	errAnyNotNullWindowFunction       = errors.New("any_not_null aggregate window function not supported")
 	// TODO(yuzefovich): #55758 has been resolved, re-evaluate whether it's
 	// worth unskipping stream ingestion processors from being wrapped.
 	errStreamIngestionWrap = errors.New("core.StreamIngestion{Data,Frontier} is not supported because of #55758")

pkg/sql/logictest/testdata/logic_test/aggregate

@@ -4034,3 +4034,68 @@ query R
 SELECT percentile_cont(ARRAY[.4::FLOAT]) WITHIN GROUP (ORDER BY i::FLOAT4) FROM t90519;
 ----
 {2.2}
+
+# any_value returns an arbitrary non-NULL value from the group (#172590).
+subtest any_value
+
+statement ok
+CREATE TABLE any_value_t (k INT, v INT, s STRING)
+
+# No rows produces NULL.
+query I
+SELECT any_value(v) FROM any_value_t
+----
+NULL
+
+statement ok
+INSERT INTO any_value_t VALUES (1, NULL, NULL), (1, 10, 'x'), (2, NULL, NULL)
+
+# NULL values are skipped; the only non-NULL value must be returned.
+query I
+SELECT any_value(v) FROM any_value_t WHERE k = 1
+----
+10
+
+# A group with only NULL values returns NULL.
+query I
+SELECT any_value(v) FROM any_value_t WHERE k = 2
+----
+NULL
+
+query II rowsort
+SELECT k, any_value(v) FROM any_value_t GROUP BY k
+----
+1  10
+2  NULL
+
+# any_value works on any type.
+query T
+SELECT any_value(s) FROM any_value_t
+----
+x
+
+# The arbitrary value must still be one of the input values.
+query B
+SELECT any_value(v) IN (2, 4, 6) FROM (VALUES (2), (4), (6)) AS t(v)
+----
+true
+
+# any_value respects FILTER and DISTINCT.
+query I
+SELECT any_value(v) FILTER (WHERE v > 8) FROM (VALUES (2), (9)) AS t(v)
+----
+9
+
+query I
+SELECT any_value(DISTINCT v) FROM (VALUES (7), (7), (7)) AS t(v)
+----
+7
+
+# any_value can be used as a window function.
+query II rowsort
+SELECT k, any_value(v) OVER (PARTITION BY k) FROM any_value_t WHERE v IS NOT NULL
+----
+1  10
+
+statement ok
+DROP TABLE any_value_t

pkg/sql/opt/memo/typing_test.go

@@ -127,9 +127,12 @@ func TestTypingComparisonAssumptions(t *testing.T) {
 func TestTypingAggregateAssumptions(t *testing.T) {
 	for _, name := range builtins.AllAggregateBuiltinNames() {
 		if name == builtins.AnyNotNull ||
+			name == "any_value" ||
 			name == "percentile_disc" ||
 			name == "percentile_cont" {
-			// These are treated as special cases.
+			// These are treated as special cases. any_value shares the
+			// any_not_null implementation, whose return type mirrors its input
+			// rather than being fixed.
 			continue
 		}
 		_, overloads := builtinsregistry.GetBuiltinProperties(name)

pkg/sql/opt/optbuilder/groupby.go

@@ -813,6 +813,10 @@ func (b *Builder) constructWindowFn(name string, args []opt.ScalarExpr) opt.Scal
 
 func (b *Builder) constructAggregate(name string, args []opt.ScalarExpr) opt.ScalarExpr {
 	switch name {
+	case "any_value":
+		// any_value has the same semantics as the internal any_not_null
+		// aggregate: it returns an arbitrary non-NULL input value.
+		return b.factory.ConstructAnyNotNullAgg(args[0])
 	case "array_agg":
 		return b.factory.ConstructArrayAgg(args[0])
 	case "array_cat_agg":

pkg/sql/opt/optbuilder/testdata/aggregate

@@ -4859,3 +4859,18 @@ scalar-group-by
  └── aggregations
       └── const-agg [as=string_agg:8]
            └── string_agg:8
+
+# any_value is lowered to the internal any-not-null-agg operator.
+build
+SELECT k, any_value(v) FROM kv GROUP BY k
+----
+group-by (hash)
+ ├── columns: k:1!null any_value:7
+ ├── grouping columns: k:1!null
+ ├── project
+ │    ├── columns: k:1!null v:2
+ │    └── scan kv
+ │         └── columns: k:1!null v:2 w:3 s:4 crdb_internal_mvcc_timestamp:5 tableoid:6
+ └── aggregations
+      └── any-not-null-agg [as=any_value:7]
+           └── v:2

pkg/sql/sem/builtins/aggregate_builtins.go

@@ -599,6 +599,16 @@ var aggregates = map[string]builtinDefinition{
 			"Returns an arbitrary not-NULL value, or NULL if none exists.",
 		))),
 
+	// any_value shares its implementation with the internal any_not_null
+	// aggregate; it is the user-facing name introduced by PostgreSQL 16.
+	"any_value": makeBuiltin(tree.FunctionProperties{},
+		makeImmutableAggOverloadWithReturnType(
+			[]*types.T{types.AnyElement},
+			tree.IdentityReturnType(0),
+			newAnyNotNullAggregate,
+			"Returns an arbitrary non-NULL value, or NULL if there are no non-NULL values.",
+		)),
+
 	// Ordered-set aggregations.
 	"percentile_disc": makeBuiltin(tree.FunctionProperties{},
 		makeImmutableAggOverloadWithReturnType(

pkg/sql/sem/builtins/fixed_oids.go

@@ -2961,6 +2961,7 @@ var builtinOidsArray = []string{
 	3006: `pg_indexes_size(relation_oid: oid) -> int`,
 	3007: `crdb_internal.tsdb_query(name: string, start_time: timestamptz, end_time: timestamptz) -> tuple{timestamptz AS timestamp, float AS value, string AS source}`,
 	3008: `crdb_internal.tsdb_query(name: string, start_time: timestamptz, end_time: timestamptz, options: jsonb) -> tuple{timestamptz AS timestamp, float AS value, string AS source}`,
+	3009: `any_value(arg1: anyelement) -> anyelement`,
 }
 
 var builtinOidsBySignature map[string]oid.Oid