execagg: extract out some aggregate helpers into a separate package

This allows us to break dependency of `colexecagg` on `execinfra`.

Release note: None

pkg/sql/BUILD.bazel

@@ -326,6 +326,7 @@ go_library(
         "//pkg/sql/distsql",
         "//pkg/sql/enum",
         "//pkg/sql/execinfra",
+        "//pkg/sql/execinfra/execagg",
         "//pkg/sql/execinfra/execopnode",
         "//pkg/sql/execinfrapb",
         "//pkg/sql/execstats",

pkg/sql/colexec/colexecagg/BUILD.bazel

@@ -21,7 +21,7 @@ go_library(
         "//pkg/sql/colexecerror",
         "//pkg/sql/colexecop",
         "//pkg/sql/colmem",
-        "//pkg/sql/execinfra",
+        "//pkg/sql/execinfra/execagg",
         "//pkg/sql/execinfrapb",
         "//pkg/sql/sem/eval",
         "//pkg/sql/sem/tree",
@@ -83,5 +83,6 @@ disallowed_imports_test(
         "//pkg/sql/colexec/colexecprojconst",
         "//pkg/sql/colexec/colexecsel",
         "//pkg/sql/colexec/colexecwindow",
+        "//pkg/sql/execinfra",
     ],
 )

pkg/sql/colexec/colexecagg/aggregate_funcs.go

@@ -19,7 +19,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/colexecerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/colexecop"
 	"github.com/cockroachdb/cockroach/pkg/sql/colmem"
-	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execagg"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -448,16 +448,16 @@ func ProcessAggregations(
 	aggregations []execinfrapb.AggregatorSpec_Aggregation,
 	inputTypes []*types.T,
 ) (
-	constructors []execinfra.AggregateConstructor,
+	constructors []execagg.AggregateConstructor,
 	constArguments []tree.Datums,
 	outputTypes []*types.T,
 	err error,
 ) {
-	constructors = make([]execinfra.AggregateConstructor, len(aggregations))
+	constructors = make([]execagg.AggregateConstructor, len(aggregations))
 	constArguments = make([]tree.Datums, len(aggregations))
 	outputTypes = make([]*types.T, len(aggregations))
 	for i, aggFn := range aggregations {
-		constructors[i], constArguments[i], outputTypes[i], err = execinfra.GetAggregateConstructor(
+		constructors[i], constArguments[i], outputTypes[i], err = execagg.GetAggregateConstructor(
 			evalCtx, semaCtx, &aggFn, inputTypes,
 		)
 		if err != nil {

pkg/sql/colexec/colexecagg/aggregators_util.go

@@ -13,7 +13,7 @@ package colexecagg
 import (
 	"github.com/cockroachdb/cockroach/pkg/sql/colexecop"
 	"github.com/cockroachdb/cockroach/pkg/sql/colmem"
-	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execagg"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -32,7 +32,7 @@ type NewAggregatorArgs struct {
 	InputTypes     []*types.T
 	Spec           *execinfrapb.AggregatorSpec
 	EvalCtx        *eval.Context
-	Constructors   []execinfra.AggregateConstructor
+	Constructors   []execagg.AggregateConstructor
 	ConstArguments []tree.Datums
 	OutputTypes    []*types.T
 }

pkg/sql/colexec/colexecagg/default_agg_tmpl.go

@@ -30,7 +30,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/colexecerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/colexecop"
 	"github.com/cockroachdb/cockroach/pkg/sql/colmem"
-	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execagg"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
@@ -118,7 +118,7 @@ func (a *default_AGGKINDAgg) Reset() {
 
 func newDefault_AGGKINDAggAlloc(
 	allocator *colmem.Allocator,
-	constructor execinfra.AggregateConstructor,
+	constructor execagg.AggregateConstructor,
 	evalCtx *eval.Context,
 	inputArgsConverter *colconv.VecToDatumConverter,
 	numArguments int,
@@ -148,7 +148,7 @@ type default_AGGKINDAggAlloc struct {
 	aggAllocBase
 	aggFuncs []default_AGGKINDAgg
 
-	constructor execinfra.AggregateConstructor
+	constructor execagg.AggregateConstructor
 	evalCtx     *eval.Context
 	// inputArgsConverter is a converter from coldata.Vecs to tree.Datums that
 	// is shared among all aggregate functions and is managed by the aggregator

pkg/sql/distsql/BUILD.bazel

@@ -71,6 +71,7 @@ go_test(
         "//pkg/sql/colflow",
         "//pkg/sql/colmem",
         "//pkg/sql/execinfra",
+        "//pkg/sql/execinfra/execagg",
         "//pkg/sql/execinfrapb",
         "//pkg/sql/flowinfra",
         "//pkg/sql/randgen",

pkg/sql/distsql/columnar_operators_test.go

@@ -25,7 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/colexec/colexecagg"
 	"github.com/cockroachdb/cockroach/pkg/sql/colexec/colexectestutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/colexec/colexecwindow"
-	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execagg"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/randgen"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
@@ -275,7 +275,7 @@ func TestAggregatorAgainstProcessor(t *testing.T) {
 								for _, typ := range aggFnInputTypes {
 									hasJSONColumn = hasJSONColumn || typ.Family() == types.JsonFamily
 								}
-								if _, outputType, err := execinfra.GetAggregateInfo(aggFn, aggFnInputTypes...); err == nil {
+								if _, outputType, err := execagg.GetAggregateInfo(aggFn, aggFnInputTypes...); err == nil {
 									outputTypes[i] = outputType
 									break
 								}
@@ -1202,7 +1202,7 @@ func TestWindowFunctionsAgainstProcessor(t *testing.T) {
 					}
 					windowerSpec.WindowFns[0].Frame = generateWindowFrame(t, rng, &ordering, inputTypes)
 
-					_, outputType, err := execinfra.GetWindowFunctionInfo(fun, argTypes...)
+					_, outputType, err := execagg.GetWindowFunctionInfo(fun, argTypes...)
 					require.NoError(t, err)
 					pspec := &execinfrapb.ProcessorSpec{
 						Input:       []execinfrapb.InputSyncSpec{{ColumnTypes: inputTypes}},

pkg/sql/distsql_physical_planner.go

@@ -35,6 +35,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/colflow"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsql"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execagg"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execopnode"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execstats"
@@ -1936,7 +1937,7 @@ func (dsp *DistSQLPlanner) planAggregators(
 					for j, c := range e.ColIdx {
 						argTypes[j] = inputTypes[c]
 					}
-					_, outputType, err := execinfra.GetAggregateInfo(localFunc, argTypes...)
+					_, outputType, err := execagg.GetAggregateInfo(localFunc, argTypes...)
 					if err != nil {
 						return err
 					}
@@ -1988,7 +1989,7 @@ func (dsp *DistSQLPlanner) planAggregators(
 							// the current aggregation e.
 							argTypes[i] = intermediateTypes[argIdxs[i]]
 						}
-						_, outputType, err := execinfra.GetAggregateInfo(finalInfo.Fn, argTypes...)
+						_, outputType, err := execagg.GetAggregateInfo(finalInfo.Fn, argTypes...)
 						if err != nil {
 							return err
 						}
@@ -2143,7 +2144,7 @@ func (dsp *DistSQLPlanner) planAggregators(
 		}
 		copy(argTypes[len(agg.ColIdx):], info.argumentsColumnTypes[i])
 		var err error
-		_, returnTyp, err := execinfra.GetAggregateInfo(agg.Func, argTypes...)
+		_, returnTyp, err := execagg.GetAggregateInfo(agg.Func, argTypes...)
 		if err != nil {
 			return err
 		}

pkg/sql/distsql_plan_window.go

@@ -11,7 +11,7 @@
 package sql
 
 import (
-	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execagg"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowexec"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -110,7 +110,7 @@ func (s *windowPlanState) createWindowFnSpec(
 	for i, argIdx := range funcInProgress.argsIdxs {
 		argTypes[i] = s.plan.GetResultTypes()[argIdx]
 	}
-	_, outputType, err := execinfra.GetWindowFunctionInfo(funcSpec, argTypes...)
+	_, outputType, err := execagg.GetWindowFunctionInfo(funcSpec, argTypes...)
 	if err != nil {
 		return execinfrapb.WindowerSpec_WindowFn{}, outputType, err
 	}

pkg/sql/execinfra/BUILD.bazel

@@ -5,7 +5,6 @@ load("//pkg/testutils/buildutil:buildutil.bzl", "disallowed_imports_test")
 go_library(
     name = "execinfra",
     srcs = [
-        "aggregatorbase.go",
         "base.go",
         "flow_context.go",
         "metadata_test_receiver.go",
@@ -48,7 +47,6 @@ go_library(
         "//pkg/sql/rowenc",
         "//pkg/sql/rowenc/valueside",
         "//pkg/sql/rowinfra",
-        "//pkg/sql/sem/builtins",
         "//pkg/sql/sem/eval",
         "//pkg/sql/sem/tree",
         "//pkg/sql/sessiondata",

pkg/sql/execinfra/execagg/BUILD.bazel

@@ -0,0 +1,16 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "execagg",
+    srcs = ["base.go"],
+    importpath = "github.com/cockroachdb/cockroach/pkg/sql/execinfra/execagg",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/sql/execinfrapb",
+        "//pkg/sql/sem/builtins",
+        "//pkg/sql/sem/eval",
+        "//pkg/sql/sem/tree",
+        "//pkg/sql/types",
+        "@com_github_cockroachdb_errors//:errors",
+    ],
+)

pkg/sql/execinfra/aggregatorbase.go → pkg/sql/execinfra/execagg/base.go

@@ -8,7 +8,7 @@
 // by the Apache License, Version 2.0, included in the file
 // licenses/APL.txt.
 
-package execinfra
+package execagg
 
 import (
 	"strings"

pkg/sql/physicalplan/BUILD.bazel

@@ -64,6 +64,7 @@ go_test(
         "//pkg/sql/catalog/desctestutils",
         "//pkg/sql/distsql",
         "//pkg/sql/execinfra",
+        "//pkg/sql/execinfra/execagg",
         "//pkg/sql/execinfrapb",
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/physicalplan/replicaoracle",

pkg/sql/physicalplan/aggregator_funcs_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/distsql"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execagg"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/physicalplan"
@@ -188,7 +189,7 @@ func checkDistAggregationInfo(
 	intermediaryTypes := make([]*types.T, numIntermediary)
 	for i, fn := range info.LocalStage {
 		var err error
-		_, returnTyp, err := execinfra.GetAggregateInfo(fn, colTypes...)
+		_, returnTyp, err := execagg.GetAggregateInfo(fn, colTypes...)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -207,7 +208,7 @@ func checkDistAggregationInfo(
 			inputTypes[i] = intermediaryTypes[localIdx]
 		}
 		var err error
-		_, finalOutputTypes[i], err = execinfra.GetAggregateInfo(finalInfo.Fn, inputTypes...)
+		_, finalOutputTypes[i], err = execagg.GetAggregateInfo(finalInfo.Fn, inputTypes...)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -509,7 +510,7 @@ func TestSingleArgumentDistAggregateFunctions(t *testing.T) {
 				continue
 			}
 			// See if this column works with this function.
-			_, _, err := execinfra.GetAggregateInfo(fn, col.GetType())
+			_, _, err := execagg.GetAggregateInfo(fn, col.GetType())
 			if err != nil {
 				continue
 			}

pkg/sql/rowexec/BUILD.bazel

@@ -58,6 +58,7 @@ go_library(
         "//pkg/sql/catalog/descs",
         "//pkg/sql/catalog/typedesc",
         "//pkg/sql/execinfra",
+        "//pkg/sql/execinfra/execagg",
         "//pkg/sql/execinfra/execopnode",
         "//pkg/sql/execinfra/execreleasable",
         "//pkg/sql/execinfrapb",

pkg/sql/rowexec/aggregator.go

@@ -15,6 +15,7 @@ import (
 	"unsafe"
 
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execagg"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execopnode"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execstats"
@@ -133,7 +134,7 @@ func (ag *aggregatorBase) init(
 				)
 			}
 		}
-		constructor, arguments, outputType, err := execinfra.GetAggregateConstructor(
+		constructor, arguments, outputType, err := execagg.GetAggregateConstructor(
 			flowCtx.EvalCtx, semaCtx, &aggInfo, ag.inputTypes,
 		)
 		if err != nil {

pkg/sql/rowexec/windower.go

@@ -14,6 +14,7 @@ import (
 	"context"
 
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execagg"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra/execopnode"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execstats"
@@ -125,7 +126,7 @@ func newWindower(
 		for i, argIdx := range windowFn.ArgsIdxs {
 			argTypes[i] = w.inputTypes[argIdx]
 		}
-		windowConstructor, outputType, err := execinfra.GetWindowFunctionInfo(windowFn.Func, argTypes...)
+		windowConstructor, outputType, err := execagg.GetWindowFunctionInfo(windowFn.Func, argTypes...)
 		if err != nil {
 			return nil, err
 		}