colexec: clean up materializer a bit

pkg/sql/colexec/execplan.go

@@ -66,7 +66,6 @@ func wrapRowSources(
 				processorID,
 				input,
 				inputTypes[i],
-				&execinfrapb.PostProcessSpec{},
 				nil, /* output */
 				nil, /* metadataSourcesQueue */
 				nil, /* toClose */

pkg/sql/colexec/materializer.go

@@ -30,6 +30,7 @@ type Materializer struct {
 	NonExplainable
 
 	input colexecbase.Operator
+	typs  []*types.T
 
 	da sqlbase.DatumAlloc
 
@@ -74,12 +75,13 @@ const materializerProcName = "materializer"
 // the context of the flow (i.e. it is Flow.ctxCancel). It should only be
 // non-nil in case of a root Materializer (i.e. not when we're wrapping a row
 // source).
+// NOTE: the constructor does *not* take in an execinfrapb.PostProcessSpec
+// because we expect input to handle that for us.
 func NewMaterializer(
 	flowCtx *execinfra.FlowCtx,
 	processorID int32,
 	input colexecbase.Operator,
 	typs []*types.T,
-	post *execinfrapb.PostProcessSpec,
 	output execinfra.RowReceiver,
 	metadataSourcesQueue []execinfrapb.MetadataSource,
 	toClose []IdempotentCloser,
@@ -88,13 +90,16 @@ func NewMaterializer(
 ) (*Materializer, error) {
 	m := &Materializer{
 		input:   input,
+		typs:    typs,
 		row:     make(sqlbase.EncDatumRow, len(typs)),
 		closers: toClose,
 	}
 
 	if err := m.ProcessorBase.Init(
 		m,
-		post,
+		// input must have handled any post-processing itself, so we pass in
+		// an empty post-processing spec.
+		&execinfrapb.PostProcessSpec{},
 		typs,
 		flowCtx,
 		processorID,
@@ -170,12 +175,14 @@ func (m *Materializer) next() (sqlbase.EncDatumRow, *execinfrapb.ProducerMetadat
 		}
 		m.curIdx++
 
-		typs := m.OutputTypes()
-		for colIdx := 0; colIdx < len(typs); colIdx++ {
+		for colIdx := 0; colIdx < len(m.typs); colIdx++ {
 			col := m.batch.ColVec(colIdx)
-			m.row[colIdx].Datum = PhysicalTypeColElemToDatum(col, rowIdx, &m.da, typs[colIdx])
+			m.row[colIdx].Datum = PhysicalTypeColElemToDatum(col, rowIdx, &m.da, m.typs[colIdx])
 		}
-		return m.ProcessRowHelper(m.row), nil
+		// Note that there is no post-processing to be done in the
+		// materializer, so we do not use ProcessRowHelper and emit the row
+		// directly.
+		return m.row, nil
 	}
 	return nil, m.DrainHelper()
 }

pkg/sql/colexec/materializer_test.go

@@ -18,7 +18,6 @@ import (
 	"github.com/cockroachdb/apd"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
-	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
@@ -54,7 +53,6 @@ func TestColumnarizeMaterialize(t *testing.T) {
 		1, /* processorID */
 		c,
 		typs,
-		&execinfrapb.PostProcessSpec{},
 		nil, /* output */
 		nil, /* metadataSourcesQueue */
 		nil, /* toClose */
@@ -140,7 +138,6 @@ func TestMaterializeTypes(t *testing.T) {
 		1, /* processorID */
 		c,
 		typs,
-		&execinfrapb.PostProcessSpec{},
 		nil, /* output */
 		nil, /* metadataSourcesQueue */
 		nil, /* toClose */
@@ -195,7 +192,6 @@ func BenchmarkColumnarizeMaterialize(b *testing.B) {
 			1, /* processorID */
 			c,
 			types,
-			&execinfrapb.PostProcessSpec{},
 			nil, /* output */
 			nil, /* metadataSourcesQueue */
 			nil, /* toClose */

pkg/sql/colexec/types_integration_test.go

@@ -23,7 +23,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/colexecbase"
 	"github.com/cockroachdb/cockroach/pkg/sql/colexecbase/colexecerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
-	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
@@ -90,7 +89,6 @@ func TestSupportedSQLTypesIntegration(t *testing.T) {
 				1, /* processorID */
 				arrowOp,
 				typs,
-				&execinfrapb.PostProcessSpec{},
 				output,
 				nil, /* metadataSourcesQueue */
 				nil, /* toClose */

pkg/sql/colflow/vectorized_flow.go

@@ -890,7 +890,6 @@ func (s *vectorizedFlowCreator) setupOutput(
 			pspec.ProcessorID,
 			op,
 			columnTypes,
-			&execinfrapb.PostProcessSpec{},
 			s.syncFlowConsumer,
 			metadataSourcesQueue,
 			toClose,

pkg/sql/colflow/vectorized_flow_shutdown_test.go

@@ -303,7 +303,6 @@ func TestVectorizedFlowShutdown(t *testing.T) {
 					1, /* processorID */
 					materializerInput,
 					typs,
-					&execinfrapb.PostProcessSpec{},
 					nil, /* output */
 					materializerMetadataSources,
 					[]colexec.IdempotentCloser{callbackCloser{closeCb: func() error {

pkg/sql/colflow/vectorized_meta_propagation_test.go

@@ -76,7 +76,6 @@ func TestVectorizedMetaPropagation(t *testing.T) {
 		2, /* processorID */
 		noop,
 		types,
-		&execinfrapb.PostProcessSpec{},
 		nil, /* output */
 		[]execinfrapb.MetadataSource{col},
 		nil, /* toClose */

pkg/sql/colflow/vectorized_panic_propagation_test.go

@@ -58,7 +58,6 @@ func TestVectorizedInternalPanic(t *testing.T) {
 		1, /* processorID */
 		vee,
 		types,
-		&execinfrapb.PostProcessSpec{},
 		nil, /* output */
 		nil, /* metadataSourceQueue */
 		nil, /* toClose */
@@ -106,7 +105,6 @@ func TestNonVectorizedPanicPropagation(t *testing.T) {
 		1, /* processorID */
 		nvee,
 		types,
-		&execinfrapb.PostProcessSpec{},
 		nil, /* output */
 		nil, /* metadataSourceQueue */
 		nil, /* toClose */

pkg/sql/distsql/columnar_utils_test.go

@@ -144,7 +144,6 @@ func verifyColOperator(args verifyColOperatorArgs) error {
 		int32(len(args.inputs))+2,
 		result.Op,
 		args.outputTypes,
-		&execinfrapb.PostProcessSpec{},
 		nil, /* output */
 		result.MetadataSources,
 		nil, /* toClose */

pkg/sql/distsql/vectorized_panic_propagation_test.go

@@ -21,7 +21,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/colexecbase"
 	"github.com/cockroachdb/cockroach/pkg/sql/colflow"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
-	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/flowinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/testutils/distsqlutils"
@@ -60,7 +59,6 @@ func TestNonVectorizedPanicDoesntHangServer(t *testing.T) {
 			},
 		},
 		nil, /* typs */
-		&execinfrapb.PostProcessSpec{},
 		&distsqlutils.RowBuffer{},
 		nil, /* metadataSourceQueue */
 		nil, /* toClose */

pkg/sql/sem/tree/eval_test.go

@@ -244,7 +244,6 @@ func TestEval(t *testing.T) {
 				0, /* processorID */
 				result.Op,
 				typs,
-				&execinfrapb.PostProcessSpec{},
 				nil, /* output */
 				nil, /* metadataSourcesQueue */
 				nil, /* toClose */