Merge #58261 #75746 #75766 #75804 #75822 #75843 #75845 #75857 #75865 #…

…75871

58261: opt,sql: use paired-joins with non-covering indexes for left joins r=rytaft a=sumeerbhola

This is done when the left outer/semi/anti join can use a
lookup join. Prior to this, when the non-covering index
could not fully evaluate the filter for left join we could
not generate a lookup join.

With this change:
- Left outer join becomes a pair of two left outer joins.
- Left semi join is a pair of inner join followed by left
  semi join.
- Left anti join is a pair of left outer join followed by
  left anti join.

Informs #55452

Release note (performance improvement): The optimizer can now
generate lookup joins in certain cases for non-covering
indexes, when performing a left outer/semi/anti join.

75746: dev: initialize submodules in `dev doctor` r=aayushshah15 a=aayushshah15

This commit adds a check to `dev doctor` to initialize submodules, like we do
in our `Makefile`.

Fixes #72247

Release note: None

75766: server: do not check decommission list for the tenant r=JeffSwenson a=JeffSwenson

Previously, the system tenant would return PermissionDenied if the
tenant's instance_id was equivalent to a decommissioned node's id.

Now, the system tenant does not check the decommissioned node list if
the incoming node_id belongs to a non-system tenant.

This PR feeds the request context down to the OnOutgoingPing and
OnIncomingPing callbacks. Previously the callbacks were using the
ambient context. The only use of the context was a storage.MVCCGet call
in nodeTombstoneStorage.IsDecommissioned.

Release note: None

75804: sql: support RESET ALL statement r=otan a=rafiss

fixes #75435

Release note (sql change): Support for the RESET ALL statement was
added. This statement resets the values of all session variables to
their default values.

75822: sql: error when setting timezone outside of postgres max utc offsets r=otan a=RichardJCai

Release note (sql change): Previously, users would be able to set
a UTC timezone offset of greater than 167 or less than -167. This
now returns an error.

Example:

SET TIME ZONE '168'
Gives error:
invalid value for parameter "timezone": "'168'": cannot find time zone "168": UTC timezone offset is out of range.

SET TIME ZONE '-168'
Gives error:
invalid value for parameter "timezone": "'-168'": cannot find time zone "-168": UTC timezone offset is out of range.

Fixes #75168

Note: If a user has already set a UTC timezone offset outside of these bounds, it will be unchanged. 

75843: c-deps/krb5: fix build for more recent versions of autoconf r=otan a=nicktrav

More recent versions of `autoconf`, when used to build `krb5`, generates
shell scripts with invalid syntax.

Fix by pulling in the [upstream patch][1] for the issue into our tree.

Closes #72529.

[1]:
krb5/krb5@f78edbe

75845: vendor: bump Pebble to 38b68e17aa97 r=jbowens a=nicktrav

Pebble commits:

```
38b68e17 internal/batchskl: return error on index overflow
8440f290 internal/manifest: use a line sweep to optimize NewL0Sublevels
0f5acb26 sstable: add direct block reading to suffix rewriter
26856d10 db: avoid stats flake in TestMemTableReservation
b452808f sstable: Make sstable Writer.Close idempotent
17fe1a65 sstable: add RewriteKeySuffixes function
c9e6edfc db: expose metrics on count and earliest seqnum of snapshots
b958d9a7 sstable: add a writeQueue to the sstable writer
c8dad06c db: disable automatic compactions in `MetricsTest`
015f5e38 internal/rangekey: fix range key iteration bug
```

The commit `38b68e17` contains the fix for #69906.

Closes #69906.

Release note: none

75857: sql: fix small race in distIndexBackfiller r=adityamaru a=stevendanna

This fixes a small race condition in
distIndexBackfiller. updateJobDetails calls SetResumeSpansInJob which
mutates the ResumeSpanList in the job details.  Normally, this is only
called from the periodic updater.  However, when the testing knob
AlwasyUpdateIndexBackfillDetails is set, we also update it on every
ProducerMetadata message we get back

Release note: None

75865: build: address util.log.logcrash package rename r=knz a=rail

After `util.log` was renamed to `util.log.logcrash`, the build system
stopped updating the Sentry environment variable properly. Instead of
setting it to the release version, it was falling back to the default
"development" value. As a result, all Sentry reports went to the
development environment bucket.

This patch addresses the name change.

Release note: None

75871: logictestccl: fix stale issue number in TODO r=arulajmani a=arulajmani

We closed #69265 in favour of #70558, and the only remaining work
left to address locality aware planning for tenants is captured
in #75864.

Release note: None

Co-authored-by: sumeerbhola <sumeer@cockroachlabs.com>
Co-authored-by: Aayush Shah <aayush.shah15@gmail.com>
Co-authored-by: Jeff <swenson@cockroachlabs.com>
Co-authored-by: Rafi Shamim <rafi@cockroachlabs.com>
Co-authored-by: richardjcai <caioftherichard@gmail.com>
Co-authored-by: Nick Travers <travers@cockroachlabs.com>
Co-authored-by: Steven Danna <danna@cockroachlabs.com>
Co-authored-by: Rail Aliiev <rail@iqchoice.com>
Co-authored-by: arulajmani <arulajmani@gmail.com>

DEPS.bzl

@@ -1263,10 +1263,10 @@ def go_deps():
         patches = [
             "@cockroach//build/patches:com_github_cockroachdb_pebble.patch",
         ],
-        sha256 = "0018bcef357bf7bba06d5e3eb35277709b5fd98ee437924001531fa935d8c76d",
-        strip_prefix = "github.com/cockroachdb/pebble@v0.0.0-20220126162719-a5c1766b568a",
+        sha256 = "e411c1b5f5c7d2ef9dc337615de7b51051a182bba9c298f540d74d95d8a8f279",
+        strip_prefix = "github.com/cockroachdb/pebble@v0.0.0-20220201221612-38b68e17aa97",
         urls = [
-            "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/pebble/com_github_cockroachdb_pebble-v0.0.0-20220126162719-a5c1766b568a.zip",
+            "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/pebble/com_github_cockroachdb_pebble-v0.0.0-20220201221612-38b68e17aa97.zip",
         ],
     )
     go_repository(

Makefile

@@ -979,7 +979,7 @@ $(go-targets): override LINKFLAGS += \
 	-X "github.com/cockroachdb/cockroach/pkg/build.rev=$(shell cat .buildinfo/rev)" \
 	-X "github.com/cockroachdb/cockroach/pkg/build.cgoTargetTriple=$(TARGET_TRIPLE)" \
 	$(if $(BUILDCHANNEL),-X "github.com/cockroachdb/cockroach/pkg/build.channel=$(BUILDCHANNEL)") \
-	$(if $(BUILD_TAGGED_RELEASE),-X "github.com/cockroachdb/cockroach/pkg/util/log.crashReportEnv=$(if $(BUILDINFO_TAG),$(BUILDINFO_TAG),$(shell cat .buildinfo/tag))")
+	$(if $(BUILD_TAGGED_RELEASE),-X "github.com/cockroachdb/cockroach/pkg/util/log/logcrash.crashReportEnv=$(if $(BUILDINFO_TAG),$(BUILDINFO_TAG),$(shell cat .buildinfo/tag))")
 
 # The build.utcTime format must remain in sync with TimeFormat in
 # pkg/build/info.go. It is not installed in tests or in `buildshort` to avoid

build/bazelutil/stamp.sh

@@ -28,7 +28,7 @@ fi
 
 # TODO(ricky): Also provide a way to stamp the following variables:
 # - github.com/cockroachdb/cockroach/pkg/build.channel
-# - github.com/cockroachdb/cockroach/pkg/util/log.crashReportEnv
+# - github.com/cockroachdb/cockroach/pkg/util/log/logcrash.crashReportEnv
 
 # Variables beginning with "STABLE" will be written to stable-status.txt, and
 # others will be written to volatile-status.txt.

docs/generated/sql/bnf/reset_session_stmt.bnf

@@ -1,3 +1,4 @@
 reset_session_stmt ::=
 	'RESET' session_var
 	| 'RESET' 'SESSION' session_var
+	| 'RESET_ALL' 'ALL'

docs/generated/sql/bnf/stmt_block.bnf

@@ -579,6 +579,7 @@ pause_all_jobs_stmt ::=
 reset_session_stmt ::=
 	'RESET' session_var
 	| 'RESET' 'SESSION' session_var
+	| 'RESET_ALL' 'ALL'
 
 reset_csetting_stmt ::=
 	'RESET' 'CLUSTER' 'SETTING' var_name

go.mod

@@ -43,7 +43,7 @@ require (
 	github.com/cockroachdb/go-test-teamcity v0.0.0-20191211140407-cff980ad0a55
 	github.com/cockroachdb/gostdlib v1.13.0
 	github.com/cockroachdb/logtags v0.0.0-20211118104740-dabe8e521a4f
-	github.com/cockroachdb/pebble v0.0.0-20220126162719-a5c1766b568a
+	github.com/cockroachdb/pebble v0.0.0-20220201221612-38b68e17aa97
 	github.com/cockroachdb/redact v1.1.3
 	github.com/cockroachdb/returncheck v0.0.0-20200612231554-92cdbca611dd
 	github.com/cockroachdb/sentry-go v0.6.1-cockroachdb.2

go.sum

@@ -427,8 +427,8 @@ github.com/cockroachdb/logtags v0.0.0-20211118104740-dabe8e521a4f h1:6jduT9Hfc0n
 github.com/cockroachdb/logtags v0.0.0-20211118104740-dabe8e521a4f/go.mod h1:Vz9DsVWQQhf3vs21MhPMZpMGSht7O/2vFW2xusFUVOs=
 github.com/cockroachdb/panicparse/v2 v2.0.0-20211103220158-604c82a44f1e h1:FrERdkPlRj+v7fc+PGpey3GUiDGuTR5CsmLCA54YJ8I=
 github.com/cockroachdb/panicparse/v2 v2.0.0-20211103220158-604c82a44f1e/go.mod h1:pMxsKyCewnV3xPaFvvT9NfwvDTcIx2Xqg0qL5Gq0SjM=
-github.com/cockroachdb/pebble v0.0.0-20220126162719-a5c1766b568a h1:JY8MIjk2GyMHjCqmHkNBekLi2N/kNS3uAKheGe78huM=
-github.com/cockroachdb/pebble v0.0.0-20220126162719-a5c1766b568a/go.mod h1:buxOO9GBtOcq1DiXDpIPYrmxY020K2A8lOrwno5FetU=
+github.com/cockroachdb/pebble v0.0.0-20220201221612-38b68e17aa97 h1:zHSurQDtRibMUCQnJhUeV96D5tO8Vq9L39L/xr4BayI=
+github.com/cockroachdb/pebble v0.0.0-20220201221612-38b68e17aa97/go.mod h1:buxOO9GBtOcq1DiXDpIPYrmxY020K2A8lOrwno5FetU=
 github.com/cockroachdb/redact v1.0.8/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg=
 github.com/cockroachdb/redact v1.1.1/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg=
 github.com/cockroachdb/redact v1.1.3 h1:AKZds10rFSIj7qADf0g46UixK8NNLwWTNdCIGS5wfSQ=

pkg/base/test_server_args.go

@@ -309,4 +309,7 @@ type TestTenantArgs struct {
 
 	// TracingDefault controls whether the tracing will be on or off by default.
 	TracingDefault tracing.TracingMode
+
+	// RPCHeartbeatInterval controls how often the tenant sends Ping requests.
+	RPCHeartbeatInterval time.Duration
 }

pkg/ccl/logictestccl/testdata/logic_test/multi_region_query_behavior

@@ -1,5 +1,5 @@
 # LogicTest: multiregion-9node-3region-3azs
-# TODO(#69265): enable multiregion-9node-3region-3azs-tenant.
+# TODO(#75864): enable multiregion-9node-3region-3azs-tenant.
 
 # Set the closed timestamp interval to be short to shorten the amount of time
 # we need to wait for the system config to propagate.

pkg/ccl/logictestccl/testdata/logic_test/regional_by_row_query_behavior

@@ -1,5 +1,5 @@
 # LogicTest: multiregion-9node-3region-3azs
-# TODO(#69265): enable multiregion-9node-3region-3azs-tenant and/or revert
+# TODO(#75864): enable multiregion-9node-3region-3azs-tenant and/or revert
 # the commit that split these changes out.
 
 # Set the closed timestamp interval to be short to shorten the amount of time

pkg/ccl/serverccl/BUILD.bazel

@@ -9,12 +9,13 @@ go_library(
 
 go_test(
     name = "serverccl_test",
-    size = "medium",
+    size = "large",
     srcs = [
         "admin_test.go",
         "main_test.go",
         "role_authentication_test.go",
         "server_sql_test.go",
+        "tenant_decommissioned_host_test.go",
         "tenant_vars_test.go",
     ],
     embed = [":serverccl"],
@@ -24,6 +25,7 @@ go_test(
         "//pkg/ccl/kvccl",
         "//pkg/ccl/utilccl",
         "//pkg/ccl/utilccl/licenseccl",
+        "//pkg/kv/kvserver/liveness/livenesspb",
         "//pkg/roachpb:with-mocks",
         "//pkg/security",
         "//pkg/security/securitytest",
@@ -34,6 +36,7 @@ go_test(
         "//pkg/sql/distsql",
         "//pkg/sql/tests",
         "//pkg/testutils/serverutils",
+        "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",
         "//pkg/util",

pkg/ccl/serverccl/tenant_decommissioned_host_test.go

@@ -0,0 +1,77 @@
+// Copyright 2022 The Cockroach Authors.
+//
+// Licensed as a CockroachDB Enterprise file under the Cockroach Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+//     https://github.com/cockroachdb/cockroach/blob/master/licenses/CCL.txt
+
+package serverccl
+
+import (
+	"context"
+	gosql "database/sql"
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness/livenesspb"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTenantWithDecommissionedID(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	// This is a regression test for a multi-tenant bug. Each tenant sql server
+	// is assigned an InstanceID. The InstanceID corresponds to the id column in
+	// the system.sql_instances table. The sql process sets rpcContext.NodeID =
+	// InstanceID and PingRequest.NodeID = rpcContext.NodeID.
+	//
+	// When a KV node recieves a ping, it checks the NodeID against a
+	// decommissioned node tombstone list. Until PR #75766, this caused the KV
+	// node to reject pings from sql servers. The rejected pings would manifest
+	// as sql connection timeouts.
+
+	skip.UnderStress(t, "decommissioning times out under stress")
+
+	ctx := context.Background()
+	tc := serverutils.StartNewTestCluster(t, 4, base.TestClusterArgs{})
+	defer tc.Stopper().Stop(ctx)
+
+	server := tc.Server(0)
+	decommissionID := tc.Server(3).NodeID()
+	require.NoError(t, server.Decommission(ctx, livenesspb.MembershipStatus_DECOMMISSIONING, []roachpb.NodeID{decommissionID}))
+	require.NoError(t, server.Decommission(ctx, livenesspb.MembershipStatus_DECOMMISSIONED, []roachpb.NodeID{decommissionID}))
+
+	tenantID := serverutils.TestTenantID()
+
+	var tenantSQLServer serverutils.TestTenantInterface
+	var tenantDB *gosql.DB
+	for instanceID := 1; instanceID <= int(decommissionID); instanceID++ {
+		sqlServer, tenant := serverutils.StartTenant(t, server, base.TestTenantArgs{
+			TenantID: tenantID,
+			Existing: instanceID != 1,
+			// Set a low heartbeat interval. The first heartbeat succeeds
+			// because the tenant needs to communicate with the kv node to
+			// determine its instance id.
+			RPCHeartbeatInterval: time.Millisecond * 5,
+		})
+		if sqlServer.RPCContext().NodeID.Get() == decommissionID {
+			tenantSQLServer = sqlServer
+			tenantDB = tenant
+		} else {
+			tenant.Close()
+		}
+	}
+	require.NotNil(t, tenantSQLServer)
+	defer tenantDB.Close()
+
+	_, err := tenantDB.Exec("CREATE ROLE test_user WITH PASSWORD 'password'")
+	require.NoError(t, err)
+}

pkg/cmd/dev/doctor.go

@@ -13,6 +13,7 @@ package main
 import (
 	"errors"
 	"log"
+	"os"
 	"os/exec"
 	"path/filepath"
 	"runtime"
@@ -99,6 +100,24 @@ Please perform the following steps:
 		}
 	}
 
+	const binDir = "bin"
+	const submodulesMarkerPath = binDir + "/.submodules-initialized"
+	d.log.Println("doctor: running submodules check")
+	if _, err := os.Stat(submodulesMarkerPath); errors.Is(err, os.ErrNotExist) {
+		if _, err = d.exec.CommandContextSilent(ctx, "git", "rev-parse", "--is-inside-work-tree"); err != nil {
+			return err
+		}
+		if _, err = d.exec.CommandContextSilent(ctx, "git", "submodule", "update", "--init", "--recursive"); err != nil {
+			return err
+		}
+		if err = d.os.MkdirAll(binDir); err != nil {
+			return err
+		}
+		if err = d.os.WriteFile(submodulesMarkerPath, ""); err != nil {
+			return err
+		}
+	}
+
 	// Check whether the build is properly configured to use stamping.
 	passedStampTest := true
 	if _, err := d.exec.CommandContextSilent(ctx, "bazel", "build", "//build/bazelutil:test_stamping"); err != nil {

pkg/rpc/context.go

@@ -407,10 +407,10 @@ type ContextOptions struct {
 	// preliminary checks but before recording clock offset information.
 	//
 	// It can inject an error.
-	OnIncomingPing func(*PingRequest) error
+	OnIncomingPing func(context.Context, *PingRequest) error
 	// OnOutgoingPing intercepts outgoing PingRequests. It may inject an
 	// error.
-	OnOutgoingPing func(*PingRequest) error
+	OnOutgoingPing func(context.Context, *PingRequest) error
 	Knobs          ContextTestingKnobs
 
 	// NodeID is the node ID / SQL instance ID container shared
@@ -1419,7 +1419,7 @@ func (rpcCtx *Context) runHeartbeat(
 				ServerVersion:        rpcCtx.Settings.Version.BinaryVersion(),
 			}
 
-			interceptor := func(*PingRequest) error { return nil }
+			interceptor := func(context.Context, *PingRequest) error { return nil }
 			if fn := rpcCtx.OnOutgoingPing; fn != nil {
 				interceptor = fn
 			}
@@ -1429,7 +1429,7 @@ func (rpcCtx *Context) runHeartbeat(
 			ping := func(ctx context.Context) error {
 				// NB: We want the request to fail-fast (the default), otherwise we won't
 				// be notified of transport failures.
-				if err := interceptor(request); err != nil {
+				if err := interceptor(ctx, request); err != nil {
 					returnErr = true
 					return err
 				}

pkg/rpc/context_test.go

@@ -179,13 +179,13 @@ func TestPingInterceptors(t *testing.T) {
 		Clock:    hlc.NewClock(hlc.UnixNano, 500*time.Millisecond),
 		Stopper:  stop.NewStopper(),
 		Settings: cluster.MakeTestingClusterSettings(),
-		OnOutgoingPing: func(req *PingRequest) error {
+		OnOutgoingPing: func(ctx context.Context, req *PingRequest) error {
 			if req.TargetNodeID == blockedTargetNodeID {
 				return errBoomSend
 			}
 			return nil
 		},
-		OnIncomingPing: func(req *PingRequest) error {
+		OnIncomingPing: func(ctx context.Context, req *PingRequest) error {
 			if req.OriginNodeID == blockedOriginNodeID {
 				return errBoomRecv
 			}

pkg/rpc/heartbeat.go

@@ -54,7 +54,7 @@ type HeartbeatService struct {
 	clusterName                    string
 	disableClusterNameVerification bool
 
-	onHandlePing func(*PingRequest) error // see ContextOptions.OnIncomingPing
+	onHandlePing func(context.Context, *PingRequest) error // see ContextOptions.OnIncomingPing
 
 	// TestingAllowNamedRPCToAnonymousServer, when defined (in tests),
 	// disables errors in case a heartbeat requests a specific node ID but
@@ -169,7 +169,7 @@ func (hs *HeartbeatService) Ping(ctx context.Context, args *PingRequest) (*PingR
 	}
 
 	if fn := hs.onHandlePing; fn != nil {
-		if err := fn(args); err != nil {
+		if err := fn(ctx, args); err != nil {
 			return nil, err
 		}
 	}

pkg/server/server.go

@@ -211,13 +211,18 @@ func NewServer(cfg Config, stopper *stop.Stopper) (*Server, error) {
 		Clock:     clock,
 		Stopper:   stopper,
 		Settings:  cfg.Settings,
-		OnOutgoingPing: func(req *rpc.PingRequest) error {
+		OnOutgoingPing: func(ctx context.Context, req *rpc.PingRequest) error {
 			// Outgoing ping will block requests with codes.FailedPrecondition to
 			// notify caller that this replica is decommissioned but others could
 			// still be tried as caller node is valid, but not the destination.
 			return checkPingFor(ctx, req.TargetNodeID, codes.FailedPrecondition)
 		},
-		OnIncomingPing: func(req *rpc.PingRequest) error {
+		OnIncomingPing: func(ctx context.Context, req *rpc.PingRequest) error {
+			// Decommission state is only tracked for the system tenant.
+			if tenantID, isTenant := roachpb.TenantFromContext(ctx); isTenant &&
+				!roachpb.IsSystemTenantID(tenantID.ToUint64()) {
+				return nil
+			}
 			// Incoming ping will reject requests with codes.PermissionDenied to
 			// signal remote node that it is not considered valid anymore and
 			// operations should fail immediately.

pkg/server/testserver.go

@@ -716,6 +716,9 @@ func (ts *TestServer) StartTenant(
 			tenantKnobs.ClusterSettingsUpdater = st.MakeUpdater()
 		}
 	}
+	if params.RPCHeartbeatInterval != 0 {
+		baseCfg.RPCHeartbeatInterval = params.RPCHeartbeatInterval
+	}
 	sqlServer, addr, httpAddr, err := StartTenant(
 		ctx,
 		stopper,

pkg/sql/backfill.go

@@ -1056,7 +1056,7 @@ func (sc *SchemaChanger) distIndexBackfill(
 		if updatedTodoSpans == nil {
 			return nil
 		}
-		nRanges, err := numRangesInSpans(ctx, sc.db, sc.distSQLPlanner, mu.updatedTodoSpans)
+		nRanges, err := numRangesInSpans(ctx, sc.db, sc.distSQLPlanner, updatedTodoSpans)
 		if err != nil {
 			return err
 		}
@@ -1077,9 +1077,16 @@ func (sc *SchemaChanger) distIndexBackfill(
 		})
 	}
 
+	// updateJobMu ensures only one goroutine is calling
+	// updateJobDetails at a time to avoid a data race in
+	// SetResumeSpansInJob. This mutex should be uncontended when
+	// sc.testingKnobs.AlwaysUpdateIndexBackfillDetails is false.
+	var updateJobMu syncutil.Mutex
 	updateJobDetails = func() error {
 		updatedTodoSpans := getTodoSpansForUpdate()
 		return sc.db.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
+			updateJobMu.Lock()
+			defer updateJobMu.Unlock()
 			// No processor has returned completed spans yet.
 			if updatedTodoSpans == nil {
 				return nil

pkg/sql/distsql_physical_planner.go

@@ -2228,14 +2228,18 @@ func (dsp *DistSQLPlanner) createPlanForLookupJoin(
 	}
 
 	joinReaderSpec := execinfrapb.JoinReaderSpec{
-		Table:                    *n.table.desc.TableDesc(),
-		Type:                     n.joinType,
-		LockingStrength:          n.table.lockingStrength,
-		LockingWaitPolicy:        n.table.lockingWaitPolicy,
-		MaintainOrdering:         len(n.reqOrdering) > 0,
-		HasSystemColumns:         n.table.containsSystemColumns,
-		LeftJoinWithPairedJoiner: n.isSecondJoinInPairedJoiner,
-		LookupBatchBytesLimit:    dsp.distSQLSrv.TestingKnobs.JoinReaderBatchBytesLimit,
+		Table:             *n.table.desc.TableDesc(),
+		Type:              n.joinType,
+		LockingStrength:   n.table.lockingStrength,
+		LockingWaitPolicy: n.table.lockingWaitPolicy,
+		// TODO(sumeer): specifying ordering here using isFirstJoinInPairedJoiner
+		// is late in the sense that the cost of this has not been taken into
+		// account. Make this decision earlier in CustomFuncs.GenerateLookupJoins.
+		MaintainOrdering:                  len(n.reqOrdering) > 0 || n.isFirstJoinInPairedJoiner,
+		HasSystemColumns:                  n.table.containsSystemColumns,
+		LeftJoinWithPairedJoiner:          n.isSecondJoinInPairedJoiner,
+		OutputGroupContinuationForLeftRow: n.isFirstJoinInPairedJoiner,
+		LookupBatchBytesLimit:             dsp.distSQLSrv.TestingKnobs.JoinReaderBatchBytesLimit,
 	}
 	joinReaderSpec.IndexIdx, err = getIndexIdx(n.table.index, n.table.desc)
 	if err != nil {
@@ -2251,7 +2255,7 @@ func (dsp *DistSQLPlanner) createPlanForLookupJoin(
 	joinReaderSpec.LookupColumnsAreKey = n.eqColsAreKey
 
 	numInputNodeCols, planToStreamColMap, post, types :=
-		mappingHelperForLookupJoins(plan, n.input, n.table, false /* addContinuationCol */)
+		mappingHelperForLookupJoins(plan, n.input, n.table, n.isFirstJoinInPairedJoiner)
 
 	// Set the lookup condition.
 	var indexVarMap []int

pkg/sql/distsql_spec_exec_factory.go

@@ -658,6 +658,7 @@ func (e *distSQLSpecExecFactory) ConstructLookupJoin(
 	remoteLookupExpr tree.TypedExpr,
 	lookupCols exec.TableColumnOrdinalSet,
 	onCond tree.TypedExpr,
+	isFirstJoinInPairedJoiner bool,
 	isSecondJoinInPairedJoiner bool,
 	reqOrdering exec.OutputOrdering,
 	locking *tree.LockingItem,