diff --git a/pkg/BUILD.bazel b/pkg/BUILD.bazel
index 887fba2e7b22..24c621de80a0 100644
--- a/pkg/BUILD.bazel
+++ b/pkg/BUILD.bazel
@@ -178,6 +178,7 @@ ALL_TESTS = [
     "//pkg/sql/colexec:colexec_test",
     "//pkg/sql/colexecerror:colexecerror_test",
     "//pkg/sql/colexecop:colexecop_test",
+    "//pkg/sql/colfetcher:colfetcher_test",
     "//pkg/sql/colflow/colrpc:colrpc_test",
     "//pkg/sql/colflow:colflow_test",
     "//pkg/sql/colmem:colmem_test",
diff --git a/pkg/sql/colexec/colbuilder/execplan.go b/pkg/sql/colexec/colbuilder/execplan.go
index 7b87275f3b53..7c4e71f9b4dd 100644
--- a/pkg/sql/colexec/colbuilder/execplan.go
+++ b/pkg/sql/colexec/colbuilder/execplan.go
@@ -735,7 +735,11 @@ func NewColOperator(
 			if err := checkNumIn(inputs, 0); err != nil {
 				return r, err
 			}
-			scanOp, err := colfetcher.NewColBatchScan(ctx, streamingAllocator, flowCtx, evalCtx, core.TableReader, post)
+
+			estimatedRowCount := spec.EstimatedRowCount
+			scanOp, err := colfetcher.NewColBatchScan(
+				ctx, streamingAllocator, flowCtx, evalCtx, core.TableReader, post, estimatedRowCount,
+			)
 			if err != nil {
 				return r, err
 			}
diff --git a/pkg/sql/colfetcher/BUILD.bazel b/pkg/sql/colfetcher/BUILD.bazel
index d23fd96b5884..0a4adda8fe1f 100644
--- a/pkg/sql/colfetcher/BUILD.bazel
+++ b/pkg/sql/colfetcher/BUILD.bazel
@@ -1,4 +1,4 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 load("//build:STRINGER.bzl", "stringer")
 
 go_library(
@@ -47,3 +47,24 @@ stringer(
     src = "cfetcher.go",
     typ = "fetcherState",
 )
+
+go_test(
+    name = "colfetcher_test",
+    srcs = [
+        "main_test.go",
+        "vectorized_batch_size_test.go",
+    ],
+    deps = [
+        "//pkg/base",
+        "//pkg/security",
+        "//pkg/security/securitytest",
+        "//pkg/server",
+        "//pkg/testutils/serverutils",
+        "//pkg/testutils/skip",
+        "//pkg/testutils/testcluster",
+        "//pkg/util/leaktest",
+        "//pkg/util/log",
+        "//pkg/util/randutil",
+        "@com_github_stretchr_testify//assert",
+    ],
+)
diff --git a/pkg/sql/colfetcher/cfetcher.go b/pkg/sql/colfetcher/cfetcher.go
index cd4be2ab0f01..7688abc9c766 100644
--- a/pkg/sql/colfetcher/cfetcher.go
+++ b/pkg/sql/colfetcher/cfetcher.go
@@ -253,6 +253,10 @@ type cFetcher struct {
 	// fetcher is the underlying fetcher that provides KVs.
 	fetcher *row.KVFetcher
 
+	// estimatedRowCount is the optimizer-derived number of expected rows that
+	// this fetch will produce, if non-zero.
+	estimatedRowCount uint64
+
 	// machine contains fields that get updated during the run of the fetcher.
 	machine struct {
 		// state is the queue of next states of the state machine. The 0th entry
@@ -269,6 +273,10 @@ type cFetcher struct {
 		// seekPrefix is the prefix to seek to in stateSeekPrefix.
 		seekPrefix roachpb.Key
 
+		// limitHint is a hint as to the number of rows that the caller expects to
+		// be returned from this fetch.
+		limitHint int
+
 		// remainingValueColsByIdx is the set of value columns that are yet to be
 		// seen during the decoding of the current row.
 		remainingValueColsByIdx util.FastIntSet
@@ -306,12 +314,19 @@ type cFetcher struct {
 	}
 }
 
-const cFetcherBatchMinCapacity = 1
-
 func (rf *cFetcher) resetBatch(timestampOutputIdx, tableOidOutputIdx int) {
 	var reallocated bool
+	var estimatedRowCount int
+	// We need to transform our rf.estimatedRowCount, which is a uint64, into
+	// an int. We have to be careful: if we just cast it directly, a giant
+	// estimate will wrap around and become negative.
+	if rf.estimatedRowCount > uint64(coldata.BatchSize()) {
+		estimatedRowCount = coldata.BatchSize()
+	} else {
+		estimatedRowCount = int(rf.estimatedRowCount)
+	}
 	rf.machine.batch, reallocated = rf.allocator.ResetMaybeReallocate(
-		rf.typs, rf.machine.batch, cFetcherBatchMinCapacity, rf.memoryLimit,
+		rf.typs, rf.machine.batch, estimatedRowCount, rf.memoryLimit,
 	)
 	if reallocated {
 		rf.machine.colvecs = rf.machine.batch.ColVecs()
@@ -649,6 +664,7 @@ func (rf *cFetcher) StartScan(
 	}
 	rf.fetcher = f
 	rf.machine.lastRowPrefix = nil
+	rf.machine.limitHint = int(limitHint)
 	rf.machine.state[0] = stateInitFetch
 	return nil
 }
@@ -1038,7 +1054,19 @@ func (rf *cFetcher) nextBatch(ctx context.Context) (coldata.Batch, error) {
 			}
 			rf.machine.rowIdx++
 			rf.shiftState()
+
+			var emitBatch bool
 			if rf.machine.rowIdx >= rf.machine.batch.Capacity() {
+				// We have no more room in our batch, so output it immediately.
+				emitBatch = true
+			} else if rf.machine.limitHint > 0 && rf.machine.rowIdx >= rf.machine.limitHint {
+				// If we made it to our limit hint, output our batch early to make sure
+				// that we don't bother filling in extra data if we don't need to.
+				emitBatch = true
+				rf.machine.limitHint = 0
+			}
+
+			if emitBatch {
 				rf.pushState(stateResetBatch)
 				rf.machine.batch.SetLength(rf.machine.rowIdx)
 				rf.machine.rowIdx = 0
diff --git a/pkg/sql/colfetcher/colbatch_scan.go b/pkg/sql/colfetcher/colbatch_scan.go
index 787a61e3734e..0efe733f323c 100644
--- a/pkg/sql/colfetcher/colbatch_scan.go
+++ b/pkg/sql/colfetcher/colbatch_scan.go
@@ -216,6 +216,7 @@ func NewColBatchScan(
 	evalCtx *tree.EvalContext,
 	spec *execinfrapb.TableReaderSpec,
 	post *execinfrapb.PostProcessSpec,
+	estimatedRowCount uint64,
 ) (*ColBatchScan, error) {
 	// NB: we hit this with a zero NodeID (but !ok) with multi-tenancy.
 	if nodeID, ok := flowCtx.NodeID.OptionalNodeID(); nodeID == 0 && ok {
@@ -264,6 +265,7 @@ func NewColBatchScan(
 	}
 
 	fetcher := cFetcherPool.Get().(*cFetcher)
+	fetcher.estimatedRowCount = estimatedRowCount
 	if _, _, err := initCRowFetcher(
 		flowCtx.Codec(), allocator, execinfra.GetWorkMemLimit(flowCtx.Cfg),
 		fetcher, table, columnIdxMap, neededColumns, spec, spec.HasSystemColumns,
diff --git a/pkg/sql/colfetcher/main_test.go b/pkg/sql/colfetcher/main_test.go
new file mode 100644
index 000000000000..27e4a8b6b3c7
--- /dev/null
+++ b/pkg/sql/colfetcher/main_test.go
@@ -0,0 +1,33 @@
+// Copyright 2021 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package colfetcher_test
+
+import (
+	"os"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/security"
+	"github.com/cockroachdb/cockroach/pkg/security/securitytest"
+	"github.com/cockroachdb/cockroach/pkg/server"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
+	"github.com/cockroachdb/cockroach/pkg/util/randutil"
+)
+
+//go:generate ../../util/leaktest/add-leaktest.sh *_test.go
+
+func TestMain(m *testing.M) {
+	security.SetAssetLoader(securitytest.EmbeddedAssets)
+	randutil.SeedForTests()
+	serverutils.InitTestServerFactory(server.TestServerFactory)
+	serverutils.InitTestClusterFactory(testcluster.TestClusterFactory)
+	os.Exit(m.Run())
+}
diff --git a/pkg/sql/colfetcher/vectorized_batch_size_test.go b/pkg/sql/colfetcher/vectorized_batch_size_test.go
new file mode 100644
index 000000000000..40b9ca04f719
--- /dev/null
+++ b/pkg/sql/colfetcher/vectorized_batch_size_test.go
@@ -0,0 +1,74 @@
+// Copyright 2021 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package colfetcher_test
+
+import (
+	"context"
+	"regexp"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
+	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/stretchr/testify/assert"
+)
+
+// TestScanBatchSize tests that the the cfetcher's dynamic batch size algorithm
+// uses the optimizer's estimated row count for its initial batch size. This
+// test sets up a scan against a table with a known row count, and makes sure
+// that the optimizer uses its statistics to produce an estimated row count that
+// is equal to the number of rows in the table, allowing the fetcher to create
+// a single batch for the scan.
+func TestScanBatchSize(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	skip.UnderMetamorphic(t, "This test doesn't work with metamorphic batch sizes.")
+
+	testClusterArgs := base.TestClusterArgs{
+		ReplicationMode: base.ReplicationAuto,
+	}
+	tc := testcluster.StartTestCluster(t, 1, testClusterArgs)
+	ctx := context.Background()
+	defer tc.Stopper().Stop(ctx)
+
+	conn := tc.Conns[0]
+
+	_, err := conn.ExecContext(ctx, `CREATE TABLE t (a PRIMARY KEY) AS SELECT generate_series(1, 511)`)
+	assert.NoError(t, err)
+
+	rows, err := conn.QueryContext(ctx, `EXPLAIN ANALYZE (VERBOSE, DISTSQL) SELECT * FROM t`)
+	assert.NoError(t, err)
+	batchCountRegex := regexp.MustCompile(`vectorized batch count: (\d+)`)
+	var found bool
+	var sb strings.Builder
+	for rows.Next() {
+		var res string
+		assert.NoError(t, rows.Scan(&res))
+		sb.WriteString(res)
+		sb.WriteByte('\n')
+		matches := batchCountRegex.FindStringSubmatch(res)
+		if len(matches) == 0 {
+			continue
+		}
+		foundBatches, err := strconv.Atoi(matches[1])
+		assert.NoError(t, err)
+		assert.Equal(t, 1, foundBatches, "should use just 1 batch to scan 511 rows")
+		found = true
+		break
+	}
+	if !found {
+		t.Fatalf("expected to find a vectorized batch count; found nothing. text:\n%s", sb.String())
+	}
+}
diff --git a/pkg/sql/colmem/allocator.go b/pkg/sql/colmem/allocator.go
index 263f8ecc8fe6..db1de607d764 100644
--- a/pkg/sql/colmem/allocator.go
+++ b/pkg/sql/colmem/allocator.go
@@ -163,10 +163,9 @@ func (a *Allocator) NewMemBatchNoCols(typs []*types.T, capacity int) coldata.Bat
 func (a *Allocator) ResetMaybeReallocate(
 	typs []*types.T, oldBatch coldata.Batch, minCapacity int, maxBatchMemSize int64,
 ) (newBatch coldata.Batch, reallocated bool) {
-	if minCapacity < 1 {
+	if minCapacity < 0 {
 		colexecerror.InternalError(errors.AssertionFailedf("invalid minCapacity %d", minCapacity))
-	}
-	if minCapacity > coldata.BatchSize() {
+	} else if minCapacity == 0 || minCapacity > coldata.BatchSize() {
 		minCapacity = coldata.BatchSize()
 	}
 	reallocated = true
diff --git a/pkg/sql/distsql_physical_planner.go b/pkg/sql/distsql_physical_planner.go
index 26c390a01fc8..834ce071ee6d 100644
--- a/pkg/sql/distsql_physical_planner.go
+++ b/pkg/sql/distsql_physical_planner.go
@@ -1327,6 +1327,7 @@ func (dsp *DistSQLPlanner) planTableReaders(
 		p.TotalEstimatedScannedRows += info.estimatedRowCount
 
 		corePlacement[i].NodeID = sp.Node
+		corePlacement[i].EstimatedRowCount = info.estimatedRowCount
 		corePlacement[i].Core.TableReader = tr
 	}
 
diff --git a/pkg/sql/execinfrapb/processors.pb.go b/pkg/sql/execinfrapb/processors.pb.go
index 0676136e38be..5031c6f8bc95 100644
--- a/pkg/sql/execinfrapb/processors.pb.go
+++ b/pkg/sql/execinfrapb/processors.pb.go
@@ -71,13 +71,16 @@ type ProcessorSpec struct {
 	// This can be aliased with InputSyncSpec.ColumnTypes, so it must not be
 	// modified in-place during planning.
 	ResultTypes []*types.T `protobuf:"bytes,7,rep,name=result_types,json=resultTypes" json:"result_types,omitempty"`
+	// estimated_row_count contains the number of rows that the optimizer expects
+	// will be emitted from this processor, or 0 if the estimate wasn't populated.
+	EstimatedRowCount uint64 `protobuf:"varint,8,opt,name=estimated_row_count,json=estimatedRowCount" json:"estimated_row_count"`
 }
 
 func (m *ProcessorSpec) Reset()         { *m = ProcessorSpec{} }
 func (m *ProcessorSpec) String() string { return proto.CompactTextString(m) }
 func (*ProcessorSpec) ProtoMessage()    {}
 func (*ProcessorSpec) Descriptor() ([]byte, []int) {
-	return fileDescriptor_processors_5e45e1d8c39180fc, []int{0}
+	return fileDescriptor_processors_840af64cccdf7fe1, []int{0}
 }
 func (m *ProcessorSpec) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -141,7 +144,7 @@ func (m *ProcessorCoreUnion) Reset()         { *m = ProcessorCoreUnion{} }
 func (m *ProcessorCoreUnion) String() string { return proto.CompactTextString(m) }
 func (*ProcessorCoreUnion) ProtoMessage()    {}
 func (*ProcessorCoreUnion) Descriptor() ([]byte, []int) {
-	return fileDescriptor_processors_5e45e1d8c39180fc, []int{1}
+	return fileDescriptor_processors_840af64cccdf7fe1, []int{1}
 }
 func (m *ProcessorCoreUnion) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -176,7 +179,7 @@ func (m *NoopCoreSpec) Reset()         { *m = NoopCoreSpec{} }
 func (m *NoopCoreSpec) String() string { return proto.CompactTextString(m) }
 func (*NoopCoreSpec) ProtoMessage()    {}
 func (*NoopCoreSpec) Descriptor() ([]byte, []int) {
-	return fileDescriptor_processors_5e45e1d8c39180fc, []int{2}
+	return fileDescriptor_processors_840af64cccdf7fe1, []int{2}
 }
 func (m *NoopCoreSpec) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -218,7 +221,7 @@ func (m *LocalPlanNodeSpec) Reset()         { *m = LocalPlanNodeSpec{} }
 func (m *LocalPlanNodeSpec) String() string { return proto.CompactTextString(m) }
 func (*LocalPlanNodeSpec) ProtoMessage()    {}
 func (*LocalPlanNodeSpec) Descriptor() ([]byte, []int) {
-	return fileDescriptor_processors_5e45e1d8c39180fc, []int{3}
+	return fileDescriptor_processors_840af64cccdf7fe1, []int{3}
 }
 func (m *LocalPlanNodeSpec) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -251,7 +254,7 @@ func (m *MetadataTestSenderSpec) Reset()         { *m = MetadataTestSenderSpec{}
 func (m *MetadataTestSenderSpec) String() string { return proto.CompactTextString(m) }
 func (*MetadataTestSenderSpec) ProtoMessage()    {}
 func (*MetadataTestSenderSpec) Descriptor() ([]byte, []int) {
-	return fileDescriptor_processors_5e45e1d8c39180fc, []int{4}
+	return fileDescriptor_processors_840af64cccdf7fe1, []int{4}
 }
 func (m *MetadataTestSenderSpec) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -284,7 +287,7 @@ func (m *MetadataTestReceiverSpec) Reset()         { *m = MetadataTestReceiverSp
 func (m *MetadataTestReceiverSpec) String() string { return proto.CompactTextString(m) }
 func (*MetadataTestReceiverSpec) ProtoMessage()    {}
 func (*MetadataTestReceiverSpec) Descriptor() ([]byte, []int) {
-	return fileDescriptor_processors_5e45e1d8c39180fc, []int{5}
+	return fileDescriptor_processors_840af64cccdf7fe1, []int{5}
 }
 func (m *MetadataTestReceiverSpec) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -390,6 +393,9 @@ func (m *ProcessorSpec) MarshalTo(dAtA []byte) (int, error) {
 			i += n
 		}
 	}
+	dAtA[i] = 0x40
+	i++
+	i = encodeVarintProcessors(dAtA, i, uint64(m.EstimatedRowCount))
 	return i, nil
 }
 
@@ -911,6 +917,7 @@ func (m *ProcessorSpec) Size() (n int) {
 			n += 1 + l + sovProcessors(uint64(l))
 		}
 	}
+	n += 1 + sovProcessors(uint64(m.EstimatedRowCount))
 	return n
 }
 
@@ -1503,6 +1510,25 @@ func (m *ProcessorSpec) Unmarshal(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 8:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field EstimatedRowCount", wireType)
+			}
+			m.EstimatedRowCount = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowProcessors
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.EstimatedRowCount |= (uint64(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := skipProcessors(dAtA[iNdEx:])
@@ -3061,89 +3087,90 @@ var (
 )
 
 func init() {
-	proto.RegisterFile("sql/execinfrapb/processors.proto", fileDescriptor_processors_5e45e1d8c39180fc)
+	proto.RegisterFile("sql/execinfrapb/processors.proto", fileDescriptor_processors_840af64cccdf7fe1)
 }
 
-var fileDescriptor_processors_5e45e1d8c39180fc = []byte{
-	// 1265 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x97, 0xdf, 0x72, 0x13, 0xb7,
-	0x17, 0xc7, 0xe3, 0xc4, 0x24, 0xb6, 0x9c, 0x10, 0x23, 0x02, 0xe8, 0xe7, 0x5f, 0xeb, 0xa4, 0x2e,
-	0xb4, 0x2e, 0x05, 0xd3, 0x32, 0x6d, 0x2f, 0x98, 0x76, 0x5a, 0x6c, 0x97, 0xc9, 0xa6, 0x34, 0xd0,
-	0x75, 0x0a, 0x33, 0xdc, 0xb8, 0xca, 0xae, 0xe2, 0x2c, 0xac, 0x57, 0x8b, 0x24, 0x13, 0xa0, 0x2f,
-	0xd1, 0x47, 0xe8, 0x03, 0xf4, 0x41, 0x72, 0xd5, 0xe1, 0x92, 0x2b, 0xa6, 0x35, 0x2f, 0xd2, 0xd1,
-	0xd1, 0x7a, 0x77, 0xed, 0xc4, 0xbb, 0xb9, 0xc9, 0x38, 0xf2, 0xf7, 0xfb, 0xd1, 0xd1, 0x39, 0x47,
-	0x7f, 0x8c, 0xb6, 0xe4, 0x73, 0xff, 0x16, 0x7b, 0xc9, 0x1c, 0x2f, 0x38, 0x10, 0x34, 0xdc, 0xbf,
-	0x15, 0x0a, 0xee, 0x30, 0x29, 0xb9, 0x90, 0xad, 0x50, 0x70, 0xc5, 0x31, 0x71, 0xb8, 0xf3, 0x4c,
-	0x70, 0xea, 0x1c, 0xb6, 0xe4, 0x73, 0xbf, 0xe5, 0x7a, 0x52, 0xc9, 0xe7, 0xbe, 0x18, 0x05, 0xb5,
-	0xda, 0xac, 0xd7, 0xa5, 0x8a, 0x1a, 0x57, 0xed, 0xda, 0x7c, 0x6e, 0x7f, 0x9f, 0x4a, 0x16, 0xc9,
-	0xae, 0x66, 0xc8, 0xf4, 0x6c, 0x46, 0xd5, 0xcc, 0x82, 0x8d, 0xfc, 0x67, 0x7d, 0x8f, 0x47, 0xca,
-	0x1b, 0x19, 0x4a, 0xe7, 0x90, 0x06, 0x03, 0x76, 0xc0, 0x98, 0x2b, 0xcf, 0xa0, 0x56, 0x74, 0xdf,
-	0x67, 0x7d, 0xa9, 0xa8, 0x9a, 0xa8, 0x2f, 0x69, 0xb5, 0x7a, 0x15, 0x32, 0x69, 0xfe, 0x46, 0xc3,
-	0x1b, 0x03, 0x3e, 0xe0, 0xf0, 0xf1, 0x96, 0xfe, 0x64, 0x46, 0x1b, 0x7f, 0x2f, 0xa1, 0xb5, 0x87,
-	0x13, 0x5a, 0x2f, 0x64, 0x0e, 0xee, 0xa0, 0x73, 0x5e, 0x10, 0x8e, 0x14, 0x29, 0x6c, 0x2d, 0x35,
-	0x2b, 0xb7, 0x3f, 0x6d, 0xcd, 0xcb, 0x6b, 0xcb, 0xd2, 0xb2, 0xde, 0xab, 0xc0, 0xd1, 0xbe, 0x76,
-	0xf1, 0xf8, 0xdd, 0xe6, 0x82, 0x6d, 0xbc, 0xf8, 0x1e, 0x2a, 0x3a, 0x5c, 0x30, 0xb2, 0xb8, 0x55,
-	0x68, 0x56, 0x6e, 0xdf, 0x98, 0xcf, 0x88, 0xe7, 0xee, 0x70, 0xc1, 0x7e, 0x0d, 0x3c, 0x1e, 0x44,
-	0x20, 0xf0, 0xe3, 0x6d, 0xb4, 0xcc, 0x47, 0x4a, 0x47, 0xb3, 0x04, 0xd1, 0x5c, 0x9f, 0x4f, 0x7a,
-	0x00, 0x3a, 0x9b, 0x8f, 0x14, 0x13, 0xa9, 0x80, 0x22, 0x3f, 0xee, 0xa0, 0x62, 0xc8, 0xa5, 0x22,
-	0x45, 0x88, 0xe8, 0xb3, 0x8c, 0x88, 0xb8, 0x54, 0x51, 0x54, 0x29, 0x0c, 0x98, 0xf1, 0x75, 0x54,
-	0x92, 0x8a, 0x0e, 0x58, 0xdf, 0x73, 0xc9, 0xb9, 0xad, 0x42, 0xf3, 0x5c, 0x7b, 0x5d, 0x7f, 0x3b,
-	0x7e, 0xb7, 0xb9, 0xd2, 0xd3, 0xe3, 0x56, 0xd7, 0x5e, 0x01, 0x81, 0xe5, 0xe2, 0x6f, 0xd0, 0x6a,
-	0x5c, 0x26, 0xad, 0x5f, 0x06, 0xfd, 0xc5, 0x48, 0x5f, 0x89, 0x17, 0x6e, 0x75, 0xed, 0x4a, 0x2c,
-	0xb4, 0x5c, 0xfc, 0x1d, 0x5a, 0x15, 0x4c, 0x8e, 0x7c, 0xd5, 0x87, 0xea, 0x91, 0x15, 0x58, 0x78,
-	0x6d, 0x26, 0x60, 0xc9, 0x86, 0x2d, 0x53, 0xdd, 0x3d, 0xbb, 0x62, 0xf4, 0x7b, 0xfa, 0xdf, 0xc6,
-	0x5f, 0x1b, 0x08, 0x9f, 0x4c, 0x2a, 0xbe, 0x83, 0x8a, 0x01, 0xe7, 0x21, 0x29, 0xc0, 0xf2, 0x3f,
-	0x99, 0xbf, 0xfc, 0x5d, 0xce, 0x43, 0x6d, 0xd3, 0x6b, 0xb7, 0xc1, 0x83, 0x7f, 0x42, 0x15, 0xe8,
-	0x32, 0x9b, 0x51, 0x97, 0x89, 0xa8, 0xa6, 0x19, 0x19, 0xdc, 0x4b, 0xc4, 0x40, 0x49, 0xbb, 0xf1,
-	0x36, 0x42, 0x4f, 0xb9, 0x17, 0x44, 0xac, 0x25, 0x60, 0x35, 0xe7, 0xb3, 0x76, 0x62, 0x2d, 0xa0,
-	0x52, 0x5e, 0xfc, 0x2d, 0x5a, 0x96, 0x5c, 0x28, 0x26, 0xa2, 0x9a, 0x5e, 0x9d, 0x4f, 0xe9, 0x81,
-	0x0e, 0x08, 0x91, 0x47, 0xc7, 0x41, 0x07, 0x03, 0xc1, 0x06, 0x54, 0x71, 0x01, 0xc5, 0xcc, 0x8c,
-	0xe3, 0x6e, 0xac, 0x35, 0x71, 0x24, 0x5e, 0xdc, 0x46, 0x25, 0x2d, 0xf4, 0x02, 0x47, 0x91, 0x95,
-	0xbc, 0xf4, 0x76, 0x23, 0x25, 0x50, 0x62, 0x9f, 0x4e, 0xf1, 0x90, 0x89, 0x01, 0xd3, 0xcb, 0x65,
-	0x82, 0x94, 0xf2, 0x52, 0xfc, 0x73, 0x22, 0x36, 0x29, 0x4e, 0xb9, 0xf5, 0xd2, 0x0e, 0xa9, 0x3c,
-	0x8c, 0x58, 0xe5, 0xbc, 0xa5, 0x6d, 0xc7, 0x5a, 0xb3, 0xb4, 0xc4, 0x8b, 0x7f, 0x40, 0xcb, 0x2f,
-	0xa8, 0x3f, 0x62, 0x92, 0xa0, 0x3c, 0xca, 0x23, 0xd0, 0xc5, 0x9d, 0x13, 0xf9, 0x74, 0x2c, 0xfb,
-	0xd4, 0x79, 0x76, 0xe0, 0xf9, 0x3e, 0x13, 0xa4, 0x92, 0x47, 0x69, 0xc7, 0x5a, 0x13, 0x4b, 0xe2,
-	0xc5, 0xf7, 0x11, 0x12, 0x8c, 0xba, 0xd6, 0x30, 0xe4, 0x42, 0x91, 0xb5, 0xbc, 0x83, 0xc5, 0x8e,
-	0xb5, 0x5d, 0xaa, 0xa8, 0xa1, 0x25, 0x7e, 0xfc, 0x3d, 0x5a, 0xe9, 0xd1, 0x61, 0xa8, 0x83, 0x5a,
-	0x07, 0xd4, 0xb5, 0x8c, 0xee, 0x31, 0x42, 0x60, 0x4c, 0x5c, 0xf8, 0x09, 0xaa, 0x9a, 0x8f, 0x49,
-	0x67, 0x90, 0x2a, 0x90, 0x5a, 0x79, 0xa4, 0x99, 0x5e, 0x3a, 0xc1, 0xc1, 0xbf, 0x21, 0x3c, 0x64,
-	0x8a, 0xea, 0x6b, 0x6a, 0x8f, 0x49, 0xd5, 0x63, 0x81, 0xde, 0x2b, 0x18, 0xe8, 0x5f, 0x64, 0x35,
-	0xc5, 0xac, 0x07, 0xf8, 0xa7, 0xb0, 0xf0, 0x01, 0xda, 0x48, 0x8f, 0xda, 0xcc, 0x61, 0xde, 0x0b,
-	0x26, 0xc8, 0x45, 0x98, 0xe3, 0xf6, 0xd9, 0xe6, 0x98, 0xb8, 0x60, 0x96, 0x53, 0x79, 0xf8, 0x47,
-	0x54, 0xee, 0xf4, 0x1e, 0x3d, 0x16, 0x9e, 0xde, 0xa6, 0x1b, 0x00, 0xcf, 0xb8, 0x50, 0x62, 0x29,
-	0x10, 0x13, 0x27, 0xde, 0x45, 0xab, 0xaf, 0xbd, 0xc1, 0x6b, 0x3a, 0x88, 0x7a, 0xfa, 0x12, 0x90,
-	0x32, 0x2e, 0x83, 0x27, 0x29, 0x35, 0xc0, 0xa6, 0xfc, 0xba, 0x2b, 0x43, 0xc1, 0x9f, 0x32, 0x47,
-	0xf5, 0x98, 0x22, 0x97, 0xf3, 0xba, 0xf2, 0x61, 0xac, 0x35, 0x7d, 0x94, 0x78, 0xf5, 0xe6, 0x3f,
-	0xf2, 0x02, 0x97, 0x1f, 0x31, 0x41, 0xae, 0xe4, 0x6d, 0xfe, 0xc7, 0x91, 0xd2, 0x6c, 0xfe, 0x89,
-	0x0f, 0xff, 0x82, 0xd6, 0x7c, 0xee, 0x50, 0xff, 0xa1, 0x4f, 0x83, 0x5d, 0xee, 0x32, 0x42, 0x00,
-	0xf4, 0xf9, 0x7c, 0xd0, 0xfd, 0xb4, 0x1c, 0x68, 0xd3, 0x04, 0xdd, 0x9d, 0xe6, 0x19, 0x91, 0xea,
-	0xce, 0xff, 0xe5, 0x75, 0x67, 0x67, 0xc6, 0x61, 0xba, 0x73, 0x96, 0x83, 0xf7, 0xd0, 0x79, 0x33,
-	0x76, 0x4f, 0xf0, 0x40, 0x79, 0x4c, 0x90, 0x5a, 0xde, 0x66, 0xec, 0x4c, 0xe9, 0x81, 0x3b, 0xc3,
-	0xd0, 0x25, 0xe1, 0xc2, 0xf5, 0x02, 0xea, 0x7b, 0xea, 0x15, 0xf9, 0x7f, 0x5e, 0x49, 0x1e, 0xc4,
-	0x5a, 0x53, 0x92, 0xc4, 0xab, 0xd3, 0xa9, 0x1f, 0x5b, 0x36, 0x3f, 0x8a, 0xfa, 0xee, 0x83, 0xbc,
-	0x74, 0xb6, 0xd3, 0x72, 0x93, 0xce, 0x29, 0x82, 0x4e, 0xa7, 0x17, 0xbc, 0x60, 0x42, 0x31, 0xf7,
-	0x9e, 0xe7, 0x2b, 0x26, 0x98, 0x20, 0x1f, 0xe6, 0xa5, 0xd3, 0x9a, 0x71, 0x98, 0x74, 0xce, 0x72,
-	0x74, 0x3a, 0x27, 0x63, 0x51, 0x77, 0xd7, 0xf3, 0xd2, 0x69, 0x4d, 0xe9, 0x4d, 0x3a, 0xa7, 0x19,
-	0x93, 0x73, 0x77, 0x14, 0xea, 0xd3, 0x8f, 0x6c, 0x9e, 0xe5, 0xdc, 0x35, 0xda, 0xe4, 0xdc, 0x35,
-	0xff, 0xe3, 0xc7, 0x68, 0x5d, 0x86, 0xbe, 0xa7, 0xee, 0x06, 0x6e, 0xcf, 0xa1, 0x4a, 0x27, 0x74,
-	0x0b, 0x70, 0x37, 0x33, 0xce, 0xb9, 0x69, 0x03, 0x30, 0x67, 0x29, 0xfa, 0xce, 0x13, 0x4c, 0x2a,
-	0x2e, 0x18, 0xc4, 0xf8, 0x51, 0xde, 0x9d, 0x67, 0x27, 0x62, 0x73, 0xe7, 0xa5, 0xdc, 0x7a, 0x1f,
-	0x1e, 0x4c, 0x2a, 0xd3, 0xc8, 0xdb, 0x87, 0x53, 0x15, 0x89, 0x7d, 0xd8, 0x41, 0x17, 0xa5, 0x12,
-	0x8c, 0x0e, 0xad, 0x60, 0xc0, 0xa4, 0xf2, 0x78, 0x00, 0x81, 0x7d, 0x0c, 0xb8, 0x2f, 0x33, 0x56,
-	0x7b, 0xd2, 0x04, 0xe4, 0xd3, 0x68, 0x98, 0xa3, 0x2b, 0x33, 0xc3, 0xf1, 0x36, 0xba, 0x0a, 0x13,
-	0x7d, 0x7d, 0xe6, 0x89, 0xa6, 0xf6, 0xd3, 0x3c, 0xea, 0x9d, 0xe2, 0xf1, 0x9f, 0x9b, 0x85, 0x9d,
-	0x62, 0xe9, 0x7c, 0x75, 0x7d, 0xa7, 0x58, 0xba, 0x50, 0xc5, 0x3b, 0xc5, 0xd2, 0x72, 0x75, 0x65,
-	0xa7, 0x58, 0x5a, 0xad, 0xae, 0x35, 0xce, 0xa3, 0xd5, 0xf4, 0x8b, 0xaf, 0xf1, 0x3b, 0xba, 0x70,
-	0xe2, 0x70, 0xc1, 0x4d, 0xb4, 0x6a, 0xf3, 0xa3, 0x1e, 0x1f, 0x09, 0x87, 0x59, 0xee, 0x4b, 0x78,
-	0x44, 0xae, 0x45, 0x0f, 0xe3, 0xa9, 0x6f, 0x70, 0x03, 0x95, 0x77, 0x47, 0x43, 0xf8, 0x61, 0x20,
-	0xe1, 0xa1, 0x38, 0x91, 0x25, 0xc3, 0x98, 0xa0, 0xe2, 0x2e, 0x1d, 0x32, 0x78, 0xfb, 0x95, 0x27,
-	0xcf, 0x6b, 0x3d, 0xd2, 0xf8, 0x0a, 0x5d, 0x3e, 0xfd, 0x0e, 0xc3, 0x35, 0xb4, 0xe8, 0xb9, 0x30,
-	0x6f, 0xb9, 0x8d, 0xa2, 0x27, 0xf4, 0xa2, 0xd5, 0xb5, 0x17, 0x3d, 0xb7, 0xb1, 0x8d, 0xc8, 0xbc,
-	0x5b, 0x09, 0xdf, 0x40, 0x48, 0x02, 0xa5, 0xef, 0xb9, 0x12, 0x7e, 0xd1, 0x94, 0xdb, 0x6b, 0xe3,
-	0x77, 0x9b, 0x65, 0xc3, 0xb6, 0xba, 0xd2, 0x2e, 0x1b, 0x81, 0xe5, 0xca, 0xf6, 0xcd, 0xe3, 0x7f,
-	0xeb, 0x0b, 0xc7, 0xe3, 0x7a, 0xe1, 0xcd, 0xb8, 0x5e, 0x78, 0x3b, 0xae, 0x17, 0xfe, 0x19, 0xd7,
-	0x0b, 0x7f, 0xbc, 0xaf, 0x2f, 0xbc, 0x79, 0x5f, 0x5f, 0x78, 0xfb, 0xbe, 0xbe, 0xf0, 0xa4, 0x92,
-	0xfa, 0x15, 0xf6, 0x5f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x34, 0x76, 0x7e, 0x5f, 0x94, 0x0e, 0x00,
-	0x00,
+var fileDescriptor_processors_840af64cccdf7fe1 = []byte{
+	// 1296 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x97, 0xdf, 0x72, 0xd3, 0x46,
+	0x14, 0xc6, 0xe3, 0x20, 0x12, 0x7b, 0x9d, 0x90, 0xb0, 0x09, 0xb0, 0x4d, 0x5b, 0x27, 0x75, 0xa1,
+	0x75, 0x29, 0x98, 0x96, 0xa1, 0xbd, 0x60, 0xda, 0x69, 0xb1, 0x53, 0x26, 0x4a, 0x69, 0xa0, 0x72,
+	0x0a, 0x33, 0xdc, 0xb8, 0x1b, 0x69, 0xe3, 0x08, 0x64, 0xad, 0xd8, 0x5d, 0x13, 0xa0, 0xb7, 0x7d,
+	0x80, 0x3e, 0x42, 0x1f, 0xa0, 0x0f, 0x92, 0x4b, 0x2e, 0xb9, 0x62, 0x5a, 0xe7, 0x45, 0x3a, 0x7b,
+	0x56, 0x96, 0x64, 0x27, 0x96, 0x72, 0xe3, 0x91, 0x57, 0xdf, 0xf7, 0xdb, 0xb3, 0x67, 0xcf, 0xfe,
+	0x11, 0xda, 0x90, 0x2f, 0x82, 0x5b, 0xec, 0x15, 0x73, 0xfd, 0x70, 0x5f, 0xd0, 0x68, 0xef, 0x56,
+	0x24, 0xb8, 0xcb, 0xa4, 0xe4, 0x42, 0x36, 0x23, 0xc1, 0x15, 0xc7, 0xc4, 0xe5, 0xee, 0x73, 0xc1,
+	0xa9, 0x7b, 0xd0, 0x94, 0x2f, 0x82, 0xa6, 0xe7, 0x4b, 0x25, 0x5f, 0x04, 0x62, 0x10, 0xae, 0xad,
+	0x4d, 0x7a, 0x3d, 0xaa, 0xa8, 0x71, 0xad, 0x5d, 0x9b, 0xce, 0xed, 0xee, 0x51, 0xc9, 0x62, 0xd9,
+	0xd5, 0x1c, 0x99, 0xee, 0xcd, 0xa8, 0x1a, 0x79, 0xb0, 0x41, 0xf0, 0xbc, 0xeb, 0xf3, 0x58, 0x79,
+	0x23, 0x47, 0xe9, 0x1e, 0xd0, 0xb0, 0xc7, 0xf6, 0x19, 0xf3, 0xe4, 0x19, 0xd4, 0x8a, 0xee, 0x05,
+	0xac, 0x2b, 0x15, 0x55, 0x23, 0xf5, 0x25, 0xad, 0x56, 0xaf, 0x23, 0x26, 0xcd, 0x6f, 0xdc, 0xbc,
+	0xda, 0xe3, 0x3d, 0x0e, 0x8f, 0xb7, 0xf4, 0x93, 0x69, 0xad, 0xff, 0x69, 0xa1, 0xc5, 0x47, 0x23,
+	0x5a, 0x27, 0x62, 0x2e, 0x6e, 0xa3, 0xf3, 0x7e, 0x18, 0x0d, 0x14, 0x29, 0x6d, 0x9c, 0x6b, 0x54,
+	0x6f, 0x7f, 0xde, 0x9c, 0x96, 0xd7, 0xa6, 0xad, 0x65, 0x9d, 0xd7, 0xa1, 0xab, 0x7d, 0x2d, 0xeb,
+	0xe8, 0xfd, 0xfa, 0x8c, 0x63, 0xbc, 0xf8, 0x3e, 0xb2, 0x5c, 0x2e, 0x18, 0x99, 0xdd, 0x28, 0x35,
+	0xaa, 0xb7, 0x6f, 0x4c, 0x67, 0x24, 0x7d, 0xb7, 0xb9, 0x60, 0xbf, 0x85, 0x3e, 0x0f, 0x63, 0x10,
+	0xf8, 0xf1, 0x16, 0x9a, 0xe3, 0x03, 0xa5, 0xa3, 0x39, 0x07, 0xd1, 0x5c, 0x9f, 0x4e, 0x7a, 0x08,
+	0x3a, 0x87, 0x0f, 0x14, 0x13, 0x99, 0x80, 0x62, 0x3f, 0x6e, 0x23, 0x2b, 0xe2, 0x52, 0x11, 0x0b,
+	0x22, 0xfa, 0x22, 0x27, 0x22, 0x2e, 0x55, 0x1c, 0x55, 0x06, 0x03, 0x66, 0x7c, 0x1d, 0x95, 0xa5,
+	0xa2, 0x3d, 0xd6, 0xf5, 0x3d, 0x72, 0x7e, 0xa3, 0xd4, 0x38, 0xdf, 0x5a, 0xd2, 0x6f, 0x87, 0xef,
+	0xd7, 0xe7, 0x3b, 0xba, 0xdd, 0xde, 0x74, 0xe6, 0x41, 0x60, 0x7b, 0xf8, 0x5b, 0xb4, 0x90, 0x4c,
+	0x93, 0xd6, 0xcf, 0x81, 0x7e, 0x25, 0xd6, 0x57, 0x93, 0x81, 0xdb, 0x9b, 0x4e, 0x35, 0x11, 0xda,
+	0x1e, 0xfe, 0x1e, 0x2d, 0x08, 0x26, 0x07, 0x81, 0xea, 0xc2, 0xec, 0x91, 0x79, 0x18, 0xf8, 0xda,
+	0x44, 0xc0, 0x92, 0xf5, 0x9b, 0x66, 0x76, 0x77, 0x9d, 0xaa, 0xd1, 0xef, 0xea, 0xbf, 0xf8, 0x0e,
+	0x5a, 0x61, 0x52, 0xf9, 0x7d, 0xaa, 0x98, 0xd7, 0x15, 0xfc, 0xb0, 0xeb, 0xf2, 0x41, 0xa8, 0x48,
+	0x79, 0xa3, 0xd4, 0xb0, 0xe2, 0xb1, 0x5c, 0x4c, 0x04, 0x0e, 0x3f, 0x6c, 0xeb, 0xd7, 0xf5, 0x7f,
+	0x56, 0x11, 0x3e, 0x39, 0x15, 0xf8, 0x2e, 0xb2, 0x42, 0xce, 0x23, 0x52, 0x82, 0xa4, 0x7d, 0x36,
+	0x3d, 0x69, 0x3b, 0x9c, 0x47, 0xda, 0xa6, 0x33, 0xe6, 0x80, 0x07, 0xff, 0x8c, 0xaa, 0x50, 0x9b,
+	0x0e, 0xa3, 0x1e, 0x13, 0x71, 0x25, 0xe4, 0xe4, 0x7d, 0x37, 0x15, 0x03, 0x25, 0xeb, 0xc6, 0x5b,
+	0x08, 0x3d, 0xe3, 0x7e, 0x18, 0xb3, 0xce, 0x01, 0xab, 0x31, 0x9d, 0xb5, 0x9d, 0x68, 0x01, 0x95,
+	0xf1, 0xe2, 0xef, 0xd0, 0x9c, 0xe4, 0x42, 0x31, 0x11, 0x57, 0xc2, 0xd5, 0xe9, 0x94, 0x0e, 0xe8,
+	0x80, 0x10, 0x7b, 0x74, 0x1c, 0xb4, 0xd7, 0x13, 0xac, 0x47, 0x15, 0x17, 0x50, 0x02, 0xb9, 0x71,
+	0xdc, 0x4b, 0xb4, 0x26, 0x8e, 0xd4, 0x8b, 0x5b, 0xa8, 0xac, 0x85, 0x7e, 0xe8, 0x2a, 0x32, 0x5f,
+	0x94, 0xde, 0xcd, 0x58, 0x09, 0x94, 0xc4, 0xa7, 0x53, 0xdc, 0x67, 0xa2, 0xc7, 0xf4, 0x70, 0x99,
+	0x80, 0x39, 0xce, 0x4d, 0xf1, 0x2f, 0xa9, 0xd8, 0xa4, 0x38, 0xe3, 0xd6, 0x43, 0x3b, 0xa0, 0xf2,
+	0x20, 0x66, 0x55, 0x8a, 0x86, 0xb6, 0x95, 0x68, 0xcd, 0xd0, 0x52, 0x2f, 0xfe, 0x11, 0xcd, 0xbd,
+	0xa4, 0xc1, 0x80, 0x49, 0x82, 0x8a, 0x28, 0x8f, 0x41, 0x97, 0x54, 0x4e, 0xec, 0xd3, 0xb1, 0xec,
+	0x51, 0xf7, 0xf9, 0xbe, 0x1f, 0x04, 0x4c, 0x90, 0x6a, 0x11, 0xa5, 0x95, 0x68, 0x4d, 0x2c, 0xa9,
+	0x17, 0x3f, 0x40, 0x48, 0x30, 0xea, 0xd9, 0xfd, 0x88, 0x0b, 0x45, 0x16, 0x8b, 0xb6, 0x23, 0x27,
+	0xd1, 0x6e, 0x52, 0x45, 0x0d, 0x2d, 0xf5, 0xe3, 0x1f, 0xd0, 0x7c, 0x87, 0xf6, 0x23, 0x1d, 0xd4,
+	0x12, 0xa0, 0xae, 0xe5, 0x54, 0x8f, 0x11, 0x02, 0x63, 0xe4, 0xc2, 0x4f, 0xd1, 0xb2, 0x79, 0x4c,
+	0x2b, 0x83, 0x2c, 0x03, 0xa9, 0x59, 0x44, 0x9a, 0xa8, 0xa5, 0x13, 0x1c, 0xfc, 0x3b, 0xc2, 0x7d,
+	0xa6, 0xa8, 0x3e, 0xdc, 0x76, 0x99, 0x54, 0x1d, 0x16, 0xea, 0xb5, 0x82, 0x81, 0xfe, 0x55, 0x5e,
+	0x51, 0x4c, 0x7a, 0x80, 0x7f, 0x0a, 0x0b, 0xef, 0xa3, 0xd5, 0x6c, 0xab, 0xc3, 0x5c, 0xe6, 0xbf,
+	0x64, 0x82, 0xac, 0x40, 0x1f, 0xb7, 0xcf, 0xd6, 0xc7, 0xc8, 0x05, 0xbd, 0x9c, 0xca, 0xc3, 0x3f,
+	0xa1, 0x4a, 0xbb, 0xf3, 0xf8, 0x89, 0xf0, 0xf5, 0x32, 0x5d, 0x05, 0x78, 0xce, 0x31, 0x94, 0x48,
+	0x81, 0x98, 0x3a, 0xf1, 0x0e, 0x5a, 0x78, 0xe3, 0xf7, 0xde, 0xd0, 0x5e, 0x5c, 0xd3, 0x97, 0x80,
+	0x94, 0x73, 0x84, 0x3c, 0xcd, 0xa8, 0x01, 0x36, 0xe6, 0xd7, 0x55, 0x19, 0x09, 0xfe, 0x8c, 0xb9,
+	0xaa, 0xc3, 0x14, 0xb9, 0x5c, 0x54, 0x95, 0x8f, 0x12, 0xad, 0xa9, 0xa3, 0xd4, 0xab, 0x17, 0xff,
+	0xa1, 0x1f, 0x7a, 0xfc, 0x90, 0x09, 0x72, 0xa5, 0x68, 0xf1, 0x3f, 0x89, 0x95, 0x66, 0xf1, 0x8f,
+	0x7c, 0xf8, 0x57, 0xb4, 0x18, 0x70, 0x97, 0x06, 0x8f, 0x02, 0x1a, 0xee, 0x70, 0x8f, 0x11, 0x02,
+	0xa0, 0x2f, 0xa7, 0x83, 0x1e, 0x64, 0xe5, 0x40, 0x1b, 0x27, 0xe8, 0xea, 0x34, 0x97, 0x8f, 0x4c,
+	0x75, 0x7e, 0x50, 0x54, 0x9d, 0xed, 0x09, 0x87, 0xa9, 0xce, 0x49, 0x0e, 0xde, 0x45, 0x17, 0x4c,
+	0xdb, 0x7d, 0xc1, 0x43, 0xe5, 0x33, 0x41, 0xd6, 0x8a, 0x16, 0x63, 0x7b, 0x4c, 0x0f, 0xdc, 0x09,
+	0x86, 0x9e, 0x12, 0x2e, 0x3c, 0x3f, 0xa4, 0x81, 0xaf, 0x5e, 0x93, 0x0f, 0x8b, 0xa6, 0xe4, 0x61,
+	0xa2, 0x35, 0x53, 0x92, 0x7a, 0x75, 0x3a, 0xf5, 0x15, 0xcd, 0xe1, 0x87, 0x71, 0xdd, 0x7d, 0x54,
+	0x94, 0xce, 0x56, 0x56, 0x6e, 0xd2, 0x39, 0x46, 0xd0, 0xe9, 0xf4, 0xc3, 0x97, 0x4c, 0x28, 0xe6,
+	0xdd, 0xf7, 0x03, 0xc5, 0x04, 0x13, 0xe4, 0xe3, 0xa2, 0x74, 0xda, 0x13, 0x0e, 0x93, 0xce, 0x49,
+	0x8e, 0x4e, 0xe7, 0xa8, 0x2d, 0xae, 0xee, 0x5a, 0x51, 0x3a, 0xed, 0x31, 0xbd, 0x49, 0xe7, 0x38,
+	0x63, 0xb4, 0xef, 0x0e, 0x22, 0xbd, 0xfb, 0x91, 0xf5, 0xb3, 0xec, 0xbb, 0x46, 0x9b, 0xee, 0xbb,
+	0xe6, 0x3f, 0x7e, 0x82, 0x96, 0x64, 0x14, 0xf8, 0xea, 0x5e, 0xe8, 0x75, 0x5c, 0xaa, 0x74, 0x42,
+	0x37, 0x00, 0x77, 0x33, 0x67, 0x9f, 0x1b, 0x37, 0x00, 0x73, 0x92, 0xa2, 0xcf, 0x3c, 0xc1, 0xa4,
+	0xe2, 0x82, 0x41, 0x8c, 0x9f, 0x14, 0x9d, 0x79, 0x4e, 0x2a, 0x36, 0x67, 0x5e, 0xc6, 0xad, 0xd7,
+	0xe1, 0xfe, 0x68, 0x66, 0xea, 0x45, 0xeb, 0x70, 0x6c, 0x46, 0x12, 0x1f, 0x76, 0xd1, 0x8a, 0x54,
+	0x82, 0xd1, 0xbe, 0x1d, 0xf6, 0xf4, 0xc5, 0x8a, 0x87, 0x10, 0xd8, 0xa7, 0x80, 0xfb, 0x3a, 0x67,
+	0xb4, 0x27, 0x4d, 0x40, 0x3e, 0x8d, 0x86, 0x39, 0xba, 0x32, 0xd1, 0x9c, 0x2c, 0xa3, 0xab, 0xd0,
+	0xd1, 0x37, 0x67, 0xee, 0x68, 0x6c, 0x3d, 0x4d, 0xa3, 0xde, 0xb5, 0x8e, 0xfe, 0x5e, 0x2f, 0x6d,
+	0x5b, 0xe5, 0x0b, 0xcb, 0x4b, 0xdb, 0x56, 0xf9, 0xe2, 0x32, 0xde, 0xb6, 0xca, 0x73, 0xcb, 0xf3,
+	0xdb, 0x56, 0x79, 0x61, 0x79, 0xb1, 0x7e, 0x01, 0x2d, 0x64, 0x6f, 0x7c, 0xf5, 0x3f, 0xd0, 0xc5,
+	0x13, 0x9b, 0x0b, 0x6e, 0xa0, 0x05, 0x87, 0x1f, 0x76, 0xf8, 0x40, 0xb8, 0xcc, 0xf6, 0x5e, 0xc1,
+	0x25, 0x72, 0x31, 0xbe, 0x82, 0x8e, 0xbd, 0xc1, 0x75, 0x54, 0xd9, 0x19, 0xf4, 0xe1, 0x73, 0x42,
+	0xc2, 0x45, 0x71, 0x24, 0x4b, 0x9b, 0x31, 0x41, 0xd6, 0x0e, 0xed, 0x33, 0xb8, 0xfb, 0x55, 0x46,
+	0x97, 0x72, 0xdd, 0x52, 0xbf, 0x83, 0x2e, 0x9f, 0x7e, 0x86, 0xe1, 0x35, 0x34, 0xeb, 0x7b, 0xd0,
+	0x6f, 0xa5, 0x85, 0xe2, 0x8b, 0xf7, 0xac, 0xbd, 0xe9, 0xcc, 0xfa, 0x5e, 0x7d, 0x0b, 0x91, 0x69,
+	0xa7, 0x12, 0xbe, 0x81, 0x90, 0x04, 0x4a, 0xd7, 0xf7, 0x24, 0x7c, 0x07, 0x55, 0x5a, 0x8b, 0xc3,
+	0xf7, 0xeb, 0x15, 0xc3, 0xb6, 0x37, 0xa5, 0x53, 0x31, 0x02, 0xdb, 0x93, 0xad, 0x9b, 0x47, 0xff,
+	0xd5, 0x66, 0x8e, 0x86, 0xb5, 0xd2, 0xdb, 0x61, 0xad, 0xf4, 0x6e, 0x58, 0x2b, 0xfd, 0x3b, 0xac,
+	0x95, 0xfe, 0x3a, 0xae, 0xcd, 0xbc, 0x3d, 0xae, 0xcd, 0xbc, 0x3b, 0xae, 0xcd, 0x3c, 0xad, 0x66,
+	0xbe, 0xdd, 0xfe, 0x0f, 0x00, 0x00, 0xff, 0xff, 0xbe, 0x2a, 0x7a, 0x01, 0xca, 0x0e, 0x00, 0x00,
 }
diff --git a/pkg/sql/execinfrapb/processors.proto b/pkg/sql/execinfrapb/processors.proto
index f797f629fcf4..6cb1f5911a1e 100644
--- a/pkg/sql/execinfrapb/processors.proto
+++ b/pkg/sql/execinfrapb/processors.proto
@@ -80,6 +80,10 @@ message ProcessorSpec {
   // This can be aliased with InputSyncSpec.ColumnTypes, so it must not be
   // modified in-place during planning.
   repeated sql.sem.types.T result_types = 7;
+
+  // estimated_row_count contains the number of rows that the optimizer expects
+  // will be emitted from this processor, or 0 if the estimate wasn't populated.
+  optional uint64 estimated_row_count = 8 [(gogoproto.nullable) = false];
 }
 
 message ProcessorCoreUnion {
diff --git a/pkg/sql/instrumentation.go b/pkg/sql/instrumentation.go
index 8da0346acd73..49cc8936c5a6 100644
--- a/pkg/sql/instrumentation.go
+++ b/pkg/sql/instrumentation.go
@@ -511,6 +511,7 @@ func (m execNodeTraceMetadata) annotateExplain(
 				nodeStats.RowCount.MaybeAdd(stats.Output.NumTuples)
 				nodeStats.KVBytesRead.MaybeAdd(stats.KV.BytesRead)
 				nodeStats.KVRowsRead.MaybeAdd(stats.KV.TuplesRead)
+				nodeStats.VectorizedBatchCount.MaybeAdd(stats.Output.NumBatches)
 			}
 			// If we didn't get statistics for all processors, we don't show the
 			// incomplete results. In the future, we may consider an incomplete flag
diff --git a/pkg/sql/logictest/testdata/logic_test/explain_analyze b/pkg/sql/logictest/testdata/logic_test/explain_analyze
index b2f9bdb5a6da..5b7b9862af1c 100644
--- a/pkg/sql/logictest/testdata/logic_test/explain_analyze
+++ b/pkg/sql/logictest/testdata/logic_test/explain_analyze
@@ -1,5 +1,8 @@
-# LogicTest: default-configs !local-spec-planning !fakedist-spec-planning
+# LogicTest: local fakedist fakedist-disk !local-spec-planning !fakedist-spec-planning
 #
+# TODO(yuzefovich): we run this file only with vectorize=on configs because
+# EXPLAIN ANALYZE (PLAN, VERBOSE) output differs depending on the execution
+# engine used. Clean this up.
 # TODO(yuzefovich): for some reason spec-planning configs are not happy. Figure
 # it out.
 
@@ -67,6 +70,7 @@ network usage: <hidden>
 │ columns: (k, v, a, b)
 │ cluster nodes: <hidden>
 │ actual row count: 2
+│ vectorized batch count: 0
 │ estimated row count: 990 (missing stats)
 │ equality: (v) = (a)
 │ right cols are key
@@ -75,6 +79,7 @@ network usage: <hidden>
 │     columns: (k, v)
 │     cluster nodes: <hidden>
 │     actual row count: 4
+│     vectorized batch count: 0
 │     KV rows read: 4
 │     KV bytes read: 32 B
 │     estimated row count: 1,000 (missing stats)
@@ -85,6 +90,7 @@ network usage: <hidden>
       columns: (a, b)
       cluster nodes: <hidden>
       actual row count: 3
+      vectorized batch count: 0
       KV rows read: 3
       KV bytes read: 24 B
       estimated row count: 1,000 (missing stats)
diff --git a/pkg/sql/opt/exec/explain/emit.go b/pkg/sql/opt/exec/explain/emit.go
index fa0fc41b9cef..15ed6f123096 100644
--- a/pkg/sql/opt/exec/explain/emit.go
+++ b/pkg/sql/opt/exec/explain/emit.go
@@ -338,6 +338,12 @@ func (e *emitter) emitNodeAttributes(n *Node) error {
 		if s.RowCount.HasValue() {
 			e.ob.AddField("actual row count", humanizeutil.Count(s.RowCount.Value()))
 		}
+		// Omit vectorized batches in non-verbose mode.
+		if e.ob.flags.Verbose {
+			if s.VectorizedBatchCount.HasValue() {
+				e.ob.AddField("vectorized batch count", humanizeutil.Count(s.VectorizedBatchCount.Value()))
+			}
+		}
 		if s.KVRowsRead.HasValue() {
 			e.ob.AddField("KV rows read", humanizeutil.Count(s.KVRowsRead.Value()))
 		}
diff --git a/pkg/sql/opt/exec/factory.go b/pkg/sql/opt/exec/factory.go
index 215d0ef74a30..f1b3c735d5b1 100644
--- a/pkg/sql/opt/exec/factory.go
+++ b/pkg/sql/opt/exec/factory.go
@@ -311,6 +311,10 @@ type ExecutionStats struct {
 	// RowCount is the number of rows produced by the operator.
 	RowCount optional.Uint
 
+	// VectorizedBatchCount is the number of vectorized batches produced by the
+	// operator.
+	VectorizedBatchCount optional.Uint
+
 	KVBytesRead optional.Uint
 	KVRowsRead  optional.Uint
 
diff --git a/pkg/sql/physicalplan/physical_plan.go b/pkg/sql/physicalplan/physical_plan.go
index 7cadc13e9ad9..146023afff9d 100644
--- a/pkg/sql/physicalplan/physical_plan.go
+++ b/pkg/sql/physicalplan/physical_plan.go
@@ -225,6 +225,9 @@ func (p *PhysicalPlan) SetMergeOrdering(o execinfrapb.Ordering) {
 type ProcessorCorePlacement struct {
 	NodeID roachpb.NodeID
 	Core   execinfrapb.ProcessorCoreUnion
+	// EstimatedRowCount, if set to non-zero, is the optimizer's guess of how
+	// many rows will be emitted from this processor.
+	EstimatedRowCount uint64
 }
 
 // AddNoInputStage creates a stage of processors that don't have any input from
@@ -248,8 +251,9 @@ func (p *PhysicalPlan) AddNoInputStage(
 				Output: []execinfrapb.OutputRouterSpec{{
 					Type: execinfrapb.OutputRouterSpec_PASS_THROUGH,
 				}},
-				StageID:     stageID,
-				ResultTypes: outputTypes,
+				StageID:           stageID,
+				ResultTypes:       outputTypes,
+				EstimatedRowCount: corePlacements[i].EstimatedRowCount,
 			},
 		}