memo: print lookup table distribution in EXPLAIN output

This commit adds the distribution of the lookup table of lookup join to
the EXPLAIN output when it is not empty.

Informs: cockroachdb#105942

Release note: None

Author: Mark Sirek

pkg/ccl/logictestccl/testdata/logic_test/multi_region_remote_access_error

@@ -1080,6 +1080,7 @@ project
       │    ├── fd: ()-->(11), (7)-->(9)
       │    ├── limit hint: 1.00
       │    ├── distribution: ap-southeast-2
+      │    ├── lookup table distribution: ap-southeast-2
       │    ├── prune: (7)
       │    ├── inner-join (inverted json_arr1_rbr@j_idx [as=t1])
       │    │    ├── columns: t2.j:3 t1.k:18 crdb_region:22

pkg/ccl/logictestccl/testdata/logic_test/regional_by_row_query_behavior

@@ -3690,6 +3690,7 @@ project
       ├── cost: 112.347141
       ├── fd: ()-->(2,5,6,12,13,17,20,21), (1)-->(3,4), (1)==(11,16), (11)==(1,16), (5)==(12,20), (12)==(5,20), (6)==(13,21), (13)==(6,21), (16)-->(18,19), (16)==(1,11), (20)==(5,12), (21)==(6,13), (2)==(17), (17)==(2), (3)==(18), (18)==(3), (4)==(19), (19)==(4)
       ├── distribution: ap-southeast-2
+      ├── lookup table distribution: ap-southeast-2,ca-central-1,us-east-1
       ├── inner-join (lookup abc@abc_id1_id2_idx [as=a])
       │    ├── columns: a.id:1 a.id1:2 a.id2:5 a.crdb_region:6 str:10 abc_id:11 x.id2:12 x.crdb_region:13 b.id:16 b.id1:17 b.created_at:18 b.updated_at:19 b.id2:20 b.crdb_region:21
       │    ├── key columns: [13 17 12 11] = [6 2 5 1]
@@ -3699,6 +3700,7 @@ project
       │    ├── cost: 112.088874
       │    ├── fd: ()-->(2,5,6,12,13,17,20,21), (16)-->(18,19), (11)==(1,16), (16)==(1,11), (12)==(5,20), (20)==(5,12), (13)==(6,21), (21)==(6,13), (6)==(13,21), (2)==(17), (17)==(2), (5)==(12,20), (1)==(11,16)
       │    ├── distribution: ap-southeast-2
+      │    ├── lookup table distribution: ap-southeast-2,ca-central-1,us-east-1
       │    ├── inner-join (lookup abc [as=b])
       │    │    ├── columns: str:10 abc_id:11 x.id2:12 x.crdb_region:13 b.id:16 b.id1:17 b.created_at:18 b.updated_at:19 b.id2:20 b.crdb_region:21
       │    │    ├── key columns: [21 16] = [21 16]
@@ -3708,6 +3710,7 @@ project
       │    │    ├── cost: 107.514948
       │    │    ├── fd: ()-->(12,13,17,20,21), (16)-->(18,19), (11)==(16), (16)==(11), (12)==(20), (20)==(12), (13)==(21), (21)==(13)
       │    │    ├── distribution: ap-southeast-2
+      │    │    ├── lookup table distribution: ap-southeast-2,ca-central-1,us-east-1
       │    │    ├── interesting orderings: (+11 opt(12,13)) (+10,+11 opt(12,13)) (+16 opt(17,20,21))
       │    │    ├── inner-join (lookup abc@abc_id1_id2_idx [as=b])
       │    │    │    ├── columns: str:10 abc_id:11 x.id2:12 x.crdb_region:13 b.id:16 b.id1:17 b.id2:20 b.crdb_region:21
@@ -3718,6 +3721,7 @@ project
       │    │    │    ├── cost: 88.4621065
       │    │    │    ├── fd: ()-->(12,13,17,20,21), (13)==(21), (21)==(13), (12)==(20), (20)==(12), (11)==(16), (16)==(11)
       │    │    │    ├── distribution: ap-southeast-2
+      │    │    │    ├── lookup table distribution: ap-southeast-2,ca-central-1,us-east-1
       │    │    │    ├── interesting orderings: (+(11|16) opt(12,13,17,20,21,28)) (+10,+(11|16) opt(12,13,17,20,21,28))
       │    │    │    ├── project
       │    │    │    │    ├── columns: "lookup_join_const_col_@17":28 str:10 abc_id:11 x.id2:12 x.crdb_region:13

pkg/sql/faketreeeval/evalctx.go

@@ -327,6 +327,11 @@ func (*DummyEvalPlanner) ExecutorConfig() interface{} {
 	return nil
 }
 
+// Optimizer is part of the cat.Catalog interface.
+func (*DummyEvalPlanner) Optimizer() interface{} {
+	return nil
+}
+
 var _ eval.Planner = &DummyEvalPlanner{}
 
 var errEvalPlanner = pgerror.New(pgcode.ScalarOperationCannotRunWithoutFullSessionContext,

pkg/sql/opt/cat/catalog.go

@@ -200,4 +200,8 @@ type Catalog interface {
 
 	// RoleExists returns true if the role exists.
 	RoleExists(ctx context.Context, role username.SQLUsername) (bool, error)
+
+	// Optimizer returns the query Optimizer used to optimize SQL statements
+	// referencing objects in this catalog, if any.
+	Optimizer() interface{}
 }

pkg/sql/opt/memo/expr_format.go

@@ -865,6 +865,19 @@ func (f *ExprFmtCtx) formatRelational(e RelExpr, tp treeprinter.Node) {
 		if !required.Distribution.Any() {
 			tp.Childf("distribution: %s", required.Distribution.String())
 		}
+		// Show the lookup table distribution of a lookup join, if it has been set.
+		if lookupJoinExpr, ok := e.(*LookupJoinExpr); ok && f.Catalog != nil {
+			if optimizer := f.Catalog.Optimizer(); optimizer != nil {
+				providedDistribution := GetLookupJoinLookupTableDistribution(
+					lookupJoinExpr,
+					lookupJoinExpr.RequiredPhysical(),
+					optimizer,
+				)
+				if !providedDistribution.Any() {
+					tp.Childf("lookup table distribution: %s", providedDistribution.String())
+				}
+			}
+		}
 		if distribute, ok := e.(*DistributeExpr); ok {
 			tp.Childf("input distribution: %s", distribute.Input.ProvidedPhysical().Distribution.String())
 		}

pkg/sql/opt/memo/memo.go

@@ -580,3 +580,11 @@ func (l *LiteralValuesExpr) ColList() opt.ColList {
 func (l *LiteralValuesExpr) Len() int {
 	return l.Rows.Rows.NumRows()
 }
+
+// GetLookupJoinLookupTableDistribution returns the Distribution of a lookup
+// table in a lookup join if that distribution can be statically determined.
+var GetLookupJoinLookupTableDistribution func(
+	lookupJoin *LookupJoinExpr,
+	required *physical.Required,
+	optimizer interface{},
+) (physicalDistribution physical.Distribution)

pkg/sql/opt/metadata_test.go

@@ -541,3 +541,8 @@ func (ep *fakeGetMultiregionConfigPlanner) GetRangeDescByID(
 ) (rangeDesc roachpb.RangeDescriptor, err error) {
 	return
 }
+
+// Optimizer is part of the cat.Catalog interface.
+func (ep *fakeGetMultiregionConfigPlanner) Optimizer() interface{} {
+	return nil
+}

pkg/sql/opt/testutils/testcat/test_catalog.go

@@ -316,6 +316,11 @@ func (tc *Catalog) RoleExists(ctx context.Context, role username.SQLUsername) (b
 	return true, nil
 }
 
+// Optimizer is part of the cat.Catalog interface.
+func (tc *Catalog) Optimizer() interface{} {
+	return nil
+}
+
 func (tc *Catalog) resolveSchema(toResolve *cat.SchemaName) (cat.Schema, cat.SchemaName, error) {
 	if string(toResolve.CatalogName) != testDB {
 		return nil, cat.SchemaName{}, pgerror.Newf(pgcode.InvalidSchemaName,

pkg/sql/opt/xform/physical_props.go

@@ -298,3 +298,26 @@ func BuildChildPhysicalPropsScalar(mem *memo.Memo, parent opt.Expr, nth int) *ph
 	}
 	return mem.InternPhysicalProps(&childProps)
 }
+
+func init() {
+	memo.GetLookupJoinLookupTableDistribution = func(
+		lookupJoin *memo.LookupJoinExpr,
+		required *physical.Required,
+		optimizer interface{},
+	) (physicalDistribution physical.Distribution) {
+		if optimizer == nil {
+			return physicalDistribution
+		}
+		o, ok := optimizer.(*Optimizer)
+		if !ok {
+			return physicalDistribution
+		}
+		if o.evalCtx == nil {
+			return physicalDistribution
+		}
+		_, physicalDistribution = distribution.BuildLookupJoinLookupTableDistribution(
+			o.ctx, o.evalCtx, lookupJoin, required, o.MaybeGetBestCostRelation,
+		)
+		return physicalDistribution
+	}
+}

pkg/sql/opt_catalog.go

@@ -37,6 +37,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/catconstants"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/catid"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree/treecmp"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlerrors"
@@ -489,6 +490,15 @@ func (oc *optCatalog) RoleExists(ctx context.Context, role username.SQLUsername)
 	return RoleExists(ctx, oc.planner.InternalSQLTxn(), role)
 }
 
+// Optimizer is part of the cat.Catalog interface.
+func (oc *optCatalog) Optimizer() interface{} {
+	if oc.planner == nil {
+		return nil
+	}
+	plannerInterface := eval.Planner(oc.planner)
+	return plannerInterface.Optimizer()
+}
+
 // dataSourceForDesc returns a data source wrapper for the given descriptor.
 // The wrapper might come from the cache, or it may be created now.
 func (oc *optCatalog) dataSourceForDesc(