diff --git a/pkg/sql/distsql_physical_planner.go b/pkg/sql/distsql_physical_planner.go
index d6eef86c4329..e5b6ce73e219 100644
--- a/pkg/sql/distsql_physical_planner.go
+++ b/pkg/sql/distsql_physical_planner.go
@@ -2862,19 +2862,25 @@ func (dsp *DistSQLPlanner) createPlanForZero(
 	return dsp.createValuesPlan(types, 0 /* numRows */, nil /* rawBytes */)
 }
 
-func createDistinctSpec(n *distinctNode, cols []int) *execinfrapb.DistinctSpec {
+func createDistinctSpec(
+	distinctOnColIdxs util.FastIntSet,
+	columnsInOrder util.FastIntSet,
+	nullsAreDistinct bool,
+	errorOnDup string,
+	cols []int,
+) *execinfrapb.DistinctSpec {
 	var orderedColumns []uint32
-	if !n.columnsInOrder.Empty() {
-		orderedColumns = make([]uint32, 0, n.columnsInOrder.Len())
-		for i, ok := n.columnsInOrder.Next(0); ok; i, ok = n.columnsInOrder.Next(i + 1) {
+	if !columnsInOrder.Empty() {
+		orderedColumns = make([]uint32, 0, columnsInOrder.Len())
+		for i, ok := columnsInOrder.Next(0); ok; i, ok = columnsInOrder.Next(i + 1) {
 			orderedColumns = append(orderedColumns, uint32(cols[i]))
 		}
 	}
 
 	var distinctColumns []uint32
-	if !n.distinctOnColIdxs.Empty() {
+	if !distinctOnColIdxs.Empty() {
 		for planCol, streamCol := range cols {
-			if streamCol != -1 && n.distinctOnColIdxs.Contains(planCol) {
+			if streamCol != -1 && distinctOnColIdxs.Contains(planCol) {
 				distinctColumns = append(distinctColumns, uint32(streamCol))
 			}
 		}
@@ -2890,8 +2896,8 @@ func createDistinctSpec(n *distinctNode, cols []int) *execinfrapb.DistinctSpec {
 	return &execinfrapb.DistinctSpec{
 		OrderedColumns:   orderedColumns,
 		DistinctColumns:  distinctColumns,
-		NullsAreDistinct: n.nullsAreDistinct,
-		ErrorOnDup:       n.errorOnDup,
+		NullsAreDistinct: nullsAreDistinct,
+		ErrorOnDup:       errorOnDup,
 	}
 }
 
@@ -2902,26 +2908,39 @@ func (dsp *DistSQLPlanner) createPlanForDistinct(
 	if err != nil {
 		return nil, err
 	}
-	currentResultRouters := plan.ResultRouters
+	spec := createDistinctSpec(
+		n.distinctOnColIdxs,
+		n.columnsInOrder,
+		n.nullsAreDistinct,
+		n.errorOnDup,
+		plan.PlanToStreamColMap,
+	)
+	dsp.addDistinctProcessors(plan, spec, n.reqOrdering)
+	return plan, nil
+}
 
+func (dsp *DistSQLPlanner) addDistinctProcessors(
+	plan *PhysicalPlan, spec *execinfrapb.DistinctSpec, reqOrdering ReqOrdering,
+) {
 	distinctSpec := execinfrapb.ProcessorCoreUnion{
-		Distinct: createDistinctSpec(n, plan.PlanToStreamColMap),
-	}
-
-	if len(currentResultRouters) == 1 {
-		plan.AddNoGroupingStage(distinctSpec, execinfrapb.PostProcessSpec{}, plan.ResultTypes, plan.MergeOrdering)
-		return plan, nil
+		Distinct: spec,
 	}
-
-	// TODO(arjun): This is potentially memory inefficient if we don't have any sorted columns.
+	defer func() {
+		plan.SetMergeOrdering(dsp.convertOrdering(reqOrdering, plan.PlanToStreamColMap))
+	}()
 
 	// Add distinct processors local to each existing current result processor.
 	plan.AddNoGroupingStage(distinctSpec, execinfrapb.PostProcessSpec{}, plan.ResultTypes, plan.MergeOrdering)
+	if !plan.IsLastStageDistributed() {
+		return
+	}
 
-	// TODO(arjun): We could distribute this final stage by hash.
-	plan.AddSingleGroupStage(dsp.gatewayNodeID, distinctSpec, execinfrapb.PostProcessSpec{}, plan.ResultTypes)
-
-	return plan, nil
+	nodes := getNodesOfRouters(plan.ResultRouters, plan.Processors)
+	plan.AddStageOnNodes(
+		nodes, distinctSpec, execinfrapb.PostProcessSpec{},
+		distinctSpec.Distinct.DistinctColumns, plan.ResultTypes,
+		plan.MergeOrdering, plan.ResultRouters,
+	)
 }
 
 func (dsp *DistSQLPlanner) createPlanForOrdinality(
diff --git a/pkg/sql/distsql_spec_exec_factory.go b/pkg/sql/distsql_spec_exec_factory.go
index ca02e8c244e2..44065b0b364a 100644
--- a/pkg/sql/distsql_spec_exec_factory.go
+++ b/pkg/sql/distsql_spec_exec_factory.go
@@ -593,7 +593,18 @@ func (e *distSQLSpecExecFactory) ConstructDistinct(
 	nullsAreDistinct bool,
 	errorOnDup string,
 ) (exec.Node, error) {
-	return nil, unimplemented.NewWithIssue(47473, "experimental opt-driven distsql planning: distinct")
+	physPlan, plan := getPhysPlan(input)
+	spec := createDistinctSpec(
+		distinctCols,
+		orderedCols,
+		nullsAreDistinct,
+		errorOnDup,
+		physPlan.PlanToStreamColMap,
+	)
+	e.dsp.addDistinctProcessors(physPlan, spec, ReqOrdering(reqOrdering))
+	// Since addition of distinct processors doesn't change any properties of
+	// the physical plan, we don't need to update any of those.
+	return plan, nil
 }
 
 func (e *distSQLSpecExecFactory) ConstructSetOp(
diff --git a/pkg/sql/logictest/testdata/logic_test/experimental_distsql_planning b/pkg/sql/logictest/testdata/logic_test/experimental_distsql_planning
index 23227bd8f49b..af827a445c00 100644
--- a/pkg/sql/logictest/testdata/logic_test/experimental_distsql_planning
+++ b/pkg/sql/logictest/testdata/logic_test/experimental_distsql_planning
@@ -109,3 +109,12 @@ generate_series
 1
 2
 3
+
+# Check that distinct is supported.
+query I rowsort
+SELECT DISTINCT v FROM kv
+----
+1
+2
+NULL
+3
diff --git a/pkg/sql/logictest/testdata/logic_test/explain_analyze_plans b/pkg/sql/logictest/testdata/logic_test/explain_analyze_plans
index 8e98f13121e4..070dc9108dd4 100644
--- a/pkg/sql/logictest/testdata/logic_test/explain_analyze_plans
+++ b/pkg/sql/logictest/testdata/logic_test/explain_analyze_plans
@@ -66,7 +66,7 @@ https://cockroachdb.github.io/distsqlplan/decode.html#eJzcml1v4jgYhe_3V1i-oprQxE
 query T
 SELECT url FROM [EXPLAIN ANALYZE (DISTSQL) SELECT DISTINCT(kw.w) FROM kv JOIN kw ON kv.k = kw.w ORDER BY kw.w]
 ----
-https://cockroachdb.github.io/distsqlplan/decode.html#eJzMWU1v4zYQvfdXEHNKUDkS9eE4AhZIunFRL1J7a-fQ7cIHxWJjwYrkknSyQeD_XkjehSNrQ-ojFnmLJA71ZvTevAz9Auy_GHyYDW-GH2_Rhsbo9-nkT_R1-Pfnm6vRGF2Nr26-_DNEJ9ej2e3sr5tT9H1pdj0af7xFJ6uns6fTXdjqEX2ajMZo9YQmY7R6PFuhDyh7jibT6-EU_fYlv5qDAUkaknHwQBj4XwGDATYY4IABLhjgwdyANU0XhLGUZkte8oBR-A18y4AoWW94dntuwCKlBPwX4BGPCfhwG9zFZEqCkFDTAgNCwoMozl-zerxc0-ghoM9gwGwdJMxHPTN78WTDfXSZwbgL-GJJGEo3fJ3dzLbgm3X86la2jJGYLHj0GPFnH1lnVraM8SCOEY8eiI8sBvOtAbuQXQI_AN49o2XAlkVolxjm23m-xz0BH2-NZhn3DzJ-Kmds4tc529rkbL-Z836fTZLSkFASFnaaZ5GyJT8p3B8BW35Ko4RQEx9QJSb_8pNLfPqBRvfL_K_3qhj5RhYbHqXJW1XbV8Spw4LriPEoWXATe6UydwXYbfEJRSnZVrOUvPYpeXW-wSylnFDTxodwf-0Mb7-AF1fvHFjeK027Zzrad8saOZ9X6JaFnHXtlrjjbom175YSFuy7ZV-TbmlXp61dQapOz3S1l2qNnAcVpFrIWVep2h1L1dZeqhIW7KV6rolUneq0dSpI1e2ZnvZSrZHzRQWpFnLWVapOx1J1tJeqhAV7qQ40kapbnbZuBal6Pe2FWiNjTy7UnolRkIQIo5QvCdVesm7HknW1l6yED3vJXmgiWcmMPSVsnSaMHH7Kn-5sZd-PhPdkxweWbuiCfKbpIn_N7nKSx-UTQ0gY3z21dxej5McjxoN87yolsSqzvg6kgX6QcKsyVW8OtTD19auTbanAhGsw3NYB0kA_SLhVmZoyXIKpr1-dDhjeESb7EJP1GpNTgGQdBjvCYLeo2MNgVxjsid_sCYPtIuWsbkrZ188OxZCU2KEYUjs7PBImJXYoYVMrO2zaVM_1s0MxJCV2KIbUzg6PhEmJHUrY1MoOmzJ8IPSVC7EpXdSxw47-DdJwRpNgUuJKEkxqbEkGqpUvHYtRauY0HQc1HSe1Y45qxwLVypwa01zHYQ2LpzUsGddwrXmtozqXxkANDEqMqZVBHalOigxKAkrJ4CRjlBqDKh1YaGBQYkytDKoxzcWYFBmUBJSaw0QJo9QYVOnUomhQ5xKDKo2EGhhUaSR8P4M6EiY1E5QYkyKDkoBSY1ASRqkxqNLhxfsZVGOaizGpmaDEmBQZlASUGoOSMErNz12l04viD0dYbFB2aSpUYFDz7S__BwAA__-NY9YN
+https://cockroachdb.github.io/distsqlplan/decode.html#eJzUWlFv4kYQfu-vWO1ToprYu7ZJsHRS0gtVOaVwhTz0euLBgW2wcGxqL8lFUf57ZThEjLkdGx-ezVsIXvPNfrPffDP2C03_C6lHR92b7sdbskxC8vtw8Cf52v37881Vr0-u-lc3X_7pkpPr3uh29NfNKfl-afa51_94S07mT2dPp-tl80fyadDrk_kTGfTJ_PFsTj6Q7HsyGF53h-S3L6tPY2rQKJ6Kvv8gUup9pYwalFOD2tSgDjWoS8cGXSTxRKRpnGSXvKwW9KbfqGcZNIgWS5n9e2zQSZwI6r1QGchQUI_e-nehGAp_KhLTogadCukH4epn5o-XiyR48JNnatDRwo9Sj7TM7IcHS-mRywzGnS8nM5GSeCkX2T-zW8jlInzzr-yyVIRiIoPHQD57xDqzsstS6YchkcGD8IiV0vGrQddL1gFsAN49k5mfzvLQLhkdv45X97gX1GOvxmERt3cifipGbLK3MXNtYuY_jHl7n2UUJ1ORiGnuTuNsJXTJno37w09nn-IgEonJdlIlFP_Kk0t2-iEJ7merv37WjolvYrKUQRz9aNe2O2JXyYLrIJVBNJEmcwvbfFzA1Wh2atCsCptbDYe9DcmtwtMoTqRITF5g6dfG8LZLULCPgBXIvSz041a8MO0dCkpE49aP5jwXDSuvlQyuDiZvmbb29aFCzOcl6kMuZl3rA2u4PjDt6wOQBdv60H5H9aEazcr6wDSpDwBPm_pQYAmrPvDy6sJLKKrdMh3tFbVCzBclFDUXs66KyhtWVK69ogJZsFXU83ekqNVoVioq10RRAZ42ilpgCUtR7fLqYpdQVKdlutoraoWYOyUUNRezropqN6yotvaKCmTBVlEv3pGiVqNZqai2JooK8LRR1AJLWIrqlFcXp4Siui3t9bRCxC6spy2TET-aEkZiOROJ9srqNKysjvbKCuTDVlk770hZq9GsVFZHE2UFeNooa4ElHabDe_AORbqIo1Tsnri9d7YyisT0Xqz5TONlMhGfk3iy-pn1x8Fq3WpKMhWpXH_L1x960earVPqy7IjZKi1OVSB19IPE3DqYymt4JeaYfvvEzzEwsQoZznWA1NEP0k6GV8R0aIYDzDH99mknwxvCxHcxWW8x2TlI1u5iWxmQg6FsakgMRdkATCjKpsbEbQ0xoVRvR3k-XPX5cJWL2_l4dhe31Sczn8lWM7txrp_jU0NCyRk1pHqO70jMoegigKmWLh5aPy70c3xqSCiOTw2pnuM7DiYcxwdgquX4Ds3wjrKuMEtd0pi6Tavl-Q721UCLXUvcDm5AAFA4_SzQZKPYPggUSg1nhWYtf0w4cEzUfVUbJSMBf42yzWpMONQD_RqKf4PIw9E4ABSOxhVaNnwLB2DCmdqpMeGYOIi8Wi7u4DQHQKEM7ph6uMDaQIEqjBc08HFqTDjDOwgUSpcKgELycQAonGJeGDLkj8kFcEwKHZwGPq7QwWng49SYalF_pH1C8nEAeTgaB4BC8XG8yusFTT0tq_J-QUPuBMCE4-Mg8lCmcRAonAew6kEDBwYNvDBowPdxACYcHweBwtE4Nah6Pu5gQQFAofg4Xhg15I-JAxwT9XNcFB_HCx3cz_NxR8KEQ70aE46Pg8jD0TgAFI7GAW0himdSY0KZxwGYkHwcQB6OjwNA4fg49aCBA4MGrn5UjOPjgMf8OBoHgMLROOBBP8o8DgKFMpTh6tcPbOD1A7vQwiH4uPHrL_8HAAD__7oqa9U=
 
 # This query verifies stats collection for WITH ORDINALITY and the hashJoiner.
 query T
diff --git a/pkg/sql/opt/exec/execbuilder/testdata/distsql_agg b/pkg/sql/opt/exec/execbuilder/testdata/distsql_agg
index ecee80723595..0c69f42a8654 100644
--- a/pkg/sql/opt/exec/execbuilder/testdata/distsql_agg
+++ b/pkg/sql/opt/exec/execbuilder/testdata/distsql_agg
@@ -133,12 +133,12 @@ https://cockroachdb.github.io/distsqlplan/decode.html#eJyslF2L2kAUhu_7K4ZzpTCSzC
 query T
 SELECT url FROM [EXPLAIN (DISTSQL) SELECT DISTINCT (a) FROM data]
 ----
-https://cockroachdb.github.io/distsqlplan/decode.html#eJyslEFvmzAYhu_7FdZ3ajUjMJA05ZSpzSSkLOmSHCZVHDz8qUOimNlG2hTlv0_AlC7RYqySo8Evz_vwWd6D_llCAtvFcvGwI40qyefN-gt5Xnx7Wn5KV-TmMd3utl-Xt-Tvlnadrh525Ibf9nsFNzwDCpUUuOKvqCF5BgYUQqAQAYUYKEwgo1ArmaPWUrVb9l0gFb8gCSgUVd2Y9nFGIZcKIdmDKUyJkMCOfy9xg1yg8gOgINDwouwwLXteq-KVq99AYVvzSifE8xnhlSCMSPMDFVBYNyYhcwbZgYJszBtLG_6CkLADde_zWGhTVLnxJ6dl5q30WglUKKy08CLtDSL775wTPkJ2sFViwTs7RSedmPtEmMtEfOb54aiZDDQ6_oDpVWYSuvuHTv6h50ej_AcaHf3vruIfuftHTv6R58ej_AcaHf1nV_GP3f1jJ__Y8yej_AcaHf3vr34n_Ye2QV3LSuMJ69KXg_bCQvGC_QWnZaNyfFIy7zD9ct3lugcCtenfsn6RVv2rtuC_YWYNhydhdh4O7eQBdGRNx_ZwPKb3xBqe2snTMeQ7a3hmJ8_GkO_tswoGjon9kJ2zs8OHPwEAAP__EaTtnQ==
+https://cockroachdb.github.io/distsqlplan/decode.html#eJy8lU1r4zwQx-_PpxBzanlkHPklbX3q0nYhkE26SQ4LxQc1Fo0htbySDFtCvvvieEmal2ocZHKUrZ_mb_2G8Qr07yUkMH0aPj3MSKWW5Ptk_IO8PP16Hn4bjMjV42A6m_4cXpN_W-r1YPQwI1f8utmbccNToFDITIz4u9CQvAADCgFQCIFCBBRiSCmUSs6F1lLVW1YbYJD9gaRHIS_KytSPUwpzqQQkKzC5WQpIYMZfl2IieCaU3wMKmTA8X27K1LXvS5W_c_UBFKYlL3RCPJ8RXmSEEWkWQgGFcWUScs8gXVOQldnV0oa_CUjYmrbP85hrkxdz48f7Ye7rjx6rTCiRnai2O-D1gyy4XhzR6XqXKPgy0e4c2dQ6POf_5qAvY7PeWbl3mcIWmariVKqTgUbSk6XP4oOdp2tHe7VZ-45hbTrGZ54fOPUMkmh7-f2L9QzrtGdYJz0TtPcWtPIWeH7o5A1JtL2Am4t5Czr1FnTiLWzvLWzlLfT8yMkbkmh7AbcX8xZ26i3sxFvU3lvUylvk-bGTNyTR9gLuLuYt6tRb1Pm_9US5idClLLRo9efs1YFF9iaaD9SyUnPxrOR8U6ZZjjfc5kEmtGnesmYxKJpXdcDPMDuE2Wc42IPZeXDfBb5zgZlTbhbb6cB636EdDu2y-nZbkZWO7XDsotoOI6rtMKLaDmOqERpR3XdRfWOFb-2ybl1k2WFElh1GZNlhTBZCI7LuXGQxZIpiY9RtjroNUrdJ6jhK3WYpcxqmDJmmESLtaJyeJc1OY9LsNCbNTqPSEByTdjRUrdLS9X9_AwAA__9Q4BZ6
 
 query T
 SELECT url FROM [EXPLAIN (DISTSQL) SELECT SUM (DISTINCT A) FROM data]
 ----
-https://cockroachdb.github.io/distsqlplan/decode.html#eJyslEFvmzAYhu_7FdZ3ajRHxECahlOqNpMipUkXMmlSxcHDnxgSwcw20qYo_30CpnRUi6EjR4xfnvfhs3wE_SODAMLlevmwJ6XKyKfd9om8LL8-r-9XG3LzuAr34ef1iPzZostDvbbaPOwJHzXbBTc8Agq5FLjhB9QQvAADCi5Q8ICCDxSmEFEolIxRa6mqLcc6sBI_IZhQSPOiNNVyRCGWCiE4gklNhhDAnn_LcIdcoHImQEGg4WlWYyr2olDpgatfQCEseK4DMnYY4bkgjEjzHRVQ2JYmIAsG0YmCLM0rSxueIATsRPv3eUy1SfPYONN2mUUlvVUCFQorzb1Ie4XI5jtvCR8hOtkqscl_dvLe8wfuk0Rhwo1UDmNtYPjl6WbBRhc5fovD-k-e9Zm8w8aOO2j2HY3OP_r2KrN3-_u7vfzdseMN8u9odPafXcXf6-_v9fL3xo4_yL-j0dn_7ir-fn9_v5e_P3amg_w7Gp3951e_-_5B26EuZK6xxbr05Ul1MaJIsLlItSxVjM9KxjWmedzWuXpBoDbNW9Y8rPLmVVXw7zCzht1WmL0Nu9awZyd79tquPe1b01N7eDpE-tYantnJsyHkO2t4bifPh5BZxxnrOmTvO2XR6cPvAAAA___0BiUG
+https://cockroachdb.github.io/distsqlplan/decode.html#eJy8ll9r4kAUxd_3U4T71LKReCfxX54sbReEtnarCwvFh6kZrGATd2YCW4rffdEsGv_dm5DgozFn7p1zfhzyBebPAkIY3T_c346dVC-cHy_DR-f1_vfzw83gybm6G4zGo58P187_V0z6sXk2eLodO_I6ez2SVk7AhTiJ1JP8UAbCV0BwQYALPrgQgAstmLiw1MlUGZPo9StfG8Eg-gth04V5vEzt-vHEhWmiFYRfYOd2oSCEsXxbqBclI6W9JrgQKSvni82Y9ez-Us8_pP4EF0ZLGZvQaXjoyDhy0Ensu9LgwjC1odNHmKxcSFK7m2WsnCkIceUW3-dubuw8nlqvtb9Mf33poY6UVtGJabsD3j6dd2nej9ST1W4jcXaj3TlJNuvwnO_ZQWfXxmapvXc7-WVcupnNtJpJm2gPD4wa_Xq86uP12TlBgbun8anbn7x4bhPRLLdJa28TLM4pFuHUw4YnKpHKbLSNvH0xUrFWUrEWUhmX8qS2q5AqivMhCvEhGp5fiQ9mo63RnYvxIWrlQ9TCB-NSno9OFT784nz4hfjwG15QiQ9mo63R3Yvx4dfKh18LH4xLeT66VfgIivMRFOIjaHitSnwwG22N7l2Mj6BWPoJa-GBcyvPRq-tL6MScF2WWSWzUwRfR6ZOba2NUNFOZkSZJ9VQ962S6GZP9HG50mweRMjb7F7Mfgzj7a71gXoyHYsyLxZ4Yy4k7VcSIldStSuoerRak4T5tuE-KA3pyQIoFk3WLVLdpcbsKKLSYAYUWc6AwagYURs2A0iEN79KGd6uA0qM7ocmUwlGllGoFWs3VAq1me4GRc8XAyJnA8ahY9n0XjO90szCZI10tGDDDj8qlVOi0mgudVrOhM3IudEbOhU73KjLFikcdUyp0umOQKRk8aplSodNqLnRazYbOyLnQGTkXOt2wgmlYQX-0HYY-WX37FwAA__89ryzV
 
 query T
 SELECT url FROM [EXPLAIN (DISTSQL) SELECT SUM (DISTINCT A), SUM (DISTINCT B) FROM data]
@@ -148,12 +148,12 @@ https://cockroachdb.github.io/distsqlplan/decode.html#eJyslFFr2zAUhd_3K8R9WkDBke
 query T
 SELECT url FROM [EXPLAIN (DISTSQL) SELECT DISTINCT a, b FROM data WHERE (a + b + c::INT) = 27 ORDER BY a,b]
 ----
-https://cockroachdb.github.io/distsqlplan/decode.html#eJy8lFFr2zAUhd_3Ky73KSUKjmSnTQUDb43LDFnSOYFtFD8otugMrp1JCmyE_Pdhe6RNWGVDsjxeSUffuecKbVH_zJHjIpgGd0vYqBzuo_lneAy-PUw_hDPoTcLFcvFlegV_j1R1OLtbgiCwag6nwgj4-imIAuj1BPRhdQV9SDgPZ8vxFbwHdgPzaBJE8PF7LYuRYFGmciaepUb-iBQJMiToIkEPCY4wJrhWZSK1LlV1ZFsLwvQX8iHBrFhvTLUcE0xKJZFv0WQml8hxKVa5jKRIpXKGSDCVRmR5jamM-muVPQv1Gwku1qLQHAYOBVGkQKE0P6RCgvdZbqTi0Ov5FPrgs6of333dEG8KJDjfGA4-JT7DeEew3JgXb9qIJ4mc7kh3_5NMm6xIjDM6NN8gCM5VKpVM25jsTeYLqmyuOub0ic_6GO9s5ujwJHfugTvafaK0y0QdOnDYRWfa0sE-tuszzpR1T411So0NHPeiqbV0sE_t5oypud1Tczul5g4c76KptXSwT218xtS87ql5nVLzBs7ooqm1dLBP7fY__br_YEZSr8tCywPiWzcPq-9Ypk-y-cJ1uVGJfFBlUmOacl7r6oVUatPs0qYIi2arMvhaTK1idiCmx2JmJ7egXavas4u9U3yPrOJrO_n6FPKNVTy2k8enkG_tsxq2PBP7Iztmx7t3fwIAAP__j8c3KA==
+https://cockroachdb.github.io/distsqlplan/decode.html#eJy8ll9r2zwUh-_fT3E4VylRcCQ7aWp4wVubskCXdElgGyUXSixaQ2pnkgIrJd992B5J0z-SO7m-lO1H-h09h4MfUf1aY4iz4dXwfA5buYbL6eQr3Ax_XF99Go2hdTGazWffrk7g7yf5ejQ-nwMnsCw_jrnm8P3LcDqEVotDG5Yn0IZVGI7G88EJ_A_sFCbTi-EUPv8ssAUSTLNYjPm9UBjeIEWCDAn6SDBAgj1cENzIbCWUymT-yWMBjOLfGHYJJulmq_PHC4KrTAoMH1Enei0wxDlfrsVU8FhIr4sEY6F5si6OyYNGG5ncc_mABGcbnqoQOh4FnsZAIdN3QiLBy2SthQyh1YootCFieT2R_7SgsFwgwclWhxBREjFc7AhmW33IpjS_FRjSHame_yJROklX2usdhy-PIDiRsZAifv3MwzbLB7jj6u61PRa7QzT2ZrTDVll54vOt2iRi7XKzN2ug3X8o4pDO_5B046yTbTx6fL9vRQiOItDqvUer9J5HOx5rtPssFezN9ZvvPlp799Eau49VV88qqWcdz29UvaWC_bWdNq-e1a6e1ajer67er6Te73hBo-otFeyvbdC8er929X6N6oPq6oNK6oOO12tUvaWC_bWdNa8-qF198EG_G68cOhVqk6VKVPqL6OaxRXwrylJVtpUrcS2zVXFMuZwUXPEgFkqXb2m5GKXlqzzgU5g-h-lTmB3B9H1w3wU-c4GpU27aM9PMeN--GfbNsvpmW4GR7pnhnotqM2xRbYYtqs2wTbWFtqjuu6g-NcIDs6yBiywzbJFlhi2yzLBNloW2yDpzkUUtU9Q2Rt3mqNsgdZukjqPUbZZSp2FKLdM0sEh7MU7fJc1M26SZaZs0M22VZsFt0l4MVaO0xe6_PwEAAP__881r5g==
 
 query T
 SELECT url FROM [EXPLAIN (DISTSQL) SELECT DISTINCT a, b FROM data WHERE (a + b + c::INT) = 27 ORDER BY b,a]
 ----
-https://cockroachdb.github.io/distsqlplan/decode.html#eJy8lVGL2kAUhd_7Ky73ySUjcSZx1YFC2jVLA1a3UWjLkocxGbaBbGInI7SI_73EFHeVOsmi9fHOzJnv3JPLZIPlzww5zv2Jf7eAtcrgPpx9hkf_28PkQzCFzjiYL-ZfJjfw90hVB9O7BQgCy_pwIrSAr5_80IdOR4AFyxuwIOY8mC6GN_Ae2ABm4dgP4eN3WBIQERLMi0ROxbMskT8iRYIMCTpI0EWCfYwIrlQRy7IsVHVksxMEyS_kPYJpvlrrajkiGBdKIt-gTnUmkeNCLDMZSpFIZfeQYCK1SLMdpjLqrVT6LNRvJDhfibzk0LUpiDwBCoX-IRUSvE8zLRWHTsejYIHHqn4853VDvC6Q4GytOXiUeAyjLcFirV-8lVo8SeR0S9r7H6elTvNY2_1D8zWC4EwlUsmkiclOMl9QRX3VMcciHrMw2prM0d5Z7py3JDIvlJbKpvQIySziUeskwz1g0PZTQ9tMjU27Nrvq3DR0sP80txecG9Y-NdYqNda1naum1tDBPrXBBVNz2qfmtErN6druVVNr6GCf2vCCqbntU3NbpeZ27f5VU2voYJ_a6D-97P9ghrJcFXkpD4inbu5VT75MnmT9myiLtYrlgyriHaYuZzvdbiGRpa53aV0Eeb1VGXwtpkYxOxDTYzEzih0z2THbZma1a1T3zeL-OU3fGsUDM3lwDnloFI_M5NE5ZNowY01D9rYpi7bv_gQAAP__AKVrVw==
+https://cockroachdb.github.io/distsqlplan/decode.html#eJy8ltFr4koUxt_vX3E4TxZH4kxitYELube1rNDVrgq7S_FhNEMrWONORthS_N8Xzao16pykhjxG882c830_PvKO8a8Z-jhoP7Rvh7DUM7jv977CU_vH48N_nS5U7jqD4eDbwxX8fWX93OneDkEyGCcvh9JI-P6l3W9DpSKhCuMrqMLE9zvdYesK_gXRhF7_rt2H_3_CmIEcIcN5FKqufFUx-k_IkaFAhi4y9JBhA0cMFzqaqDiO9PqV942gE_5Gv85wOl8szfrnEcNJpBX672imZqbQx6Ecz1RfyVBpp44MQ2XkdLa5Zj1osNDTV6nfkOFgIeexDzWHg5yHwCEyL0ojw_vpzCjtQ6UScKhCINb7BO7HhfzkARn2lsaHgLNA4GjFMFqa_Wyxkc8Kfb5i2ee_m8ZmOp8Yp3E4fHIFw54OlVbh6Tv3x4zf4EXGL6fOGK32o4mzo-2PipIb00dVWSCqyWFnd-D1Tyyxn87NY9wg0kZph6d9E1UW8OrZO7xPO7A7-MQw3agWLRxxuP65ERoHI_DsfPMsfDu85ohSCSc22NFxXT7hvHDCeYGEE8ZtCU_7lotwkR0vkQkvUXPcUvEiNthF0ywfL1E4XqJAvAjjtnilfcuFl5sdLzcTXm7N8UrFi9hgF02rfLzcwvFyC8SLMG6LV9q3XHh52fHyMuHl1ZxGqXgRG-yiuSkfL69wvLwC8SKM2-KV9u3Tn38n7uireBHNY5Xpq66-tkaFzyqxM46WeqIedTTZXJM89ja6zQ-hik3yL08eOvPkr_WAH8U8LeYfxeJAzPOJm5eIOb9I3bhIfWNXC6vhrt1w1yr27Dd7VrEgsm5Y1dd28fUloNjFBCh2MQUKoSZAIdQEKE2r4S274a1LQLmxd0KdKIWjSsnVCnY1VQt2NdkLhJwqBkJOBM6PiuXQd0H4bm8WInNurxbuEZcflUuu0O1qKnS7mgydkFOhE3IqdHuvcqJY-VHH5Ard3jGcKBl-1DK5QrerqdDtajJ0Qk6FTsip0O0NK4iGFfaPtnToo9U_fwIAAP__UdBziQ==
 
 query T
 SELECT url FROM [EXPLAIN (DISTSQL) SELECT c, d, sum(a+c::INT) + avg(b+d) FROM data GROUP BY c, d]
diff --git a/pkg/sql/opt/exec/execbuilder/testdata/distsql_distinct_on b/pkg/sql/opt/exec/execbuilder/testdata/distsql_distinct_on
index 3ba8c8056c80..871fbb757922 100644
--- a/pkg/sql/opt/exec/execbuilder/testdata/distsql_distinct_on
+++ b/pkg/sql/opt/exec/execbuilder/testdata/distsql_distinct_on
@@ -78,7 +78,7 @@ distinct   ·             ·            (x, y, z)  ·
 query T
 SELECT url FROM [EXPLAIN (DISTSQL) SELECT DISTINCT ON (x,y,z) x, y, z FROM xyz]
 ----
-https://cockroachdb.github.io/distsqlplan/decode.html#eJyslF1vmzAUhu_3K6xz1UpGYCAf5YqpzSSkLOmSXEyquGBwVCFRzGwj5UP57xNQpUvUGC_sLgZeP-_jWOcA8ncBAaxn89njhtSiIN9Wy-_kZfbzef41WpC7p2i9Wf-Y35P3T5p1tHjckOWC3G0p2VGyvyfvP7rwdrePgULJM1wkbygheAEGFFyg4AEFHyiMIKZQCZ6ilFw0nxzaQJRtIXAo5GVVq-ZxTCHlAiE4gMpVgRDAJvlV4AqTDIXtAIUMVZIXLWa724eVyN8SsQMK6yopZUAsuyEvaxWQ0KWhR0Mf4iMFXqsPiFTJK0LAjtS8yFMuVV6myh6dtwgZbUFXKe5VysfmdclFhgKzs73jo64Hc_6xiHdWhJmfOzM4d9u1bP_Gk--pcjIeDzp511zYNRH2rbbQLcI9VU7Ck0HCnrmwZyI8ttpCtwj3VDkJTwcJ--bCvonwxLpRt6fISffhv42STygrlBUvJV6MlM93dppRg9krdnNJ8lqk-Cx42mK65bLNtQ8ylKp7y7pFVHavmoJ_h5k27J6F2WXY1ZN70J427evD_pDeI214rCePh5An2vBUT54OIT_o_yun55roL9klOz5--RMAAP__glbS5g==
+https://cockroachdb.github.io/distsqlplan/decode.html#eJy8lU1v4jAQhu_7K6w5tZKj4HxQyIlVy0pILHSBw0oVh5RYbSSaZO1E4kP891WSCgotHhdH3Ajx43ntZzTZgvy3hACm_WH_fkYKsSS_JuPf5Kn_93H4czAiNw-D6Wz6Z3hL3peUz4PR_YyMR-RmRcmaks0tef9Rw6v1Zg4UkjTio_CNSwiegAEFByi4QMEDCj7MKWQiXXApU1Eu2VbAIFpB0KIQJ1mRl3_PKSxSwSHYQh7nSw4BzMLnJZ_wMOLCbgGFiOdhvKzKrNabXibit1CsgcI0CxMZEMsuK4-LPCA9h_Zc2vNgvqOQFvmhiMzDFw4B21H9IA-xzONkkdv-cYoeo1WhkyoH8HlNXkP5epY65HHO5jnsViSpiLjg0dF-1S5nE7OWVuRDELfZIKPUSjOb-Scrv67tHdVm-t3BNLrDdizbu7A_kCj7225fqT9Yc_3BjPrD0Xfk6DjyrOoOL3GERNmf-O5KjpzmHDlGjlx9R66Oo7ZV3eEljpAo-xN3ruTIbc6Ra-TI03fk6Ti6sy40hATZn7d7JUNec4a8xr6EX5SZcJmlieRa37lWGZRHL7w-lUwLseCPIl1UZerHccVVf0Rc5vVbVj8MkvpVGfAjzE5h9hF2jmD2PbhtAndNYGaUm_lq2lHet6uGXbWsttqWp6R9NeybqFbDiGo1jKhWw5hqhEZUt01U3ynhjlpWx0SWGkZkqWFElhrGZCE0IqtrIoshUxQbo2Zz1GyQmk1Sw1FqNkuZ0TBlyDT1EGmfxum3pKlpTJqaxqSpaVQagmPSPg1VpbT57sf_AAAA__-kuPbl
 
 # Ensure that ordering propagates past local DISTINCT processors.
 query TTTTT
@@ -98,7 +98,7 @@ distinct        ·             ·            (x, y, z)  +x
 query T
 SELECT url FROM [EXPLAIN (DISTSQL) SELECT DISTINCT ON (x,y,z) x, y, z FROM xyz ORDER BY x]
 ----
-https://cockroachdb.github.io/distsqlplan/decode.html#eJy0lU2P2jwUhffvr7DuakavUXBivrJKO1AJicIUWLQasUgTaxQpE6e2IwUQ_73Kh5RCBzsT6C5Ocu4597lX8hHkrxhc2MwWs6ctykSMvqxXX9HL7Pvz4tN8iR6m8812823xiOpfivN8-bRFqyV6yDHaY3R4RPVDJc73B7RaT2dr9PkHyneAIeEhW_pvTIL7AgQw2IDBAQwUMAxghyEVPGBSclH8ciwF8zAHt48hStJMFa93GAIuGLhHUJGKGbiw9X_GbM38kAmrDxhCpvwoLm3y_cFLRfTmiz1g2KR-Il3UswrnVaZc5NnYc7BHYXfCwDPVmEjlvzJwyQm3D7LhQjFhDc4zeOT_q-Xtj5SfRlJFSaAs0r90wGUjRVciZIKFLvLIVVPnqmnjxas673WiTfZX7x9KRs-SkfZjJy3Gbtk9i3YcvCFKPfhh18Ebyjd4yR0Hb7fHa7fBS3slgC54DVFqvKOueA3lG7z2HfE67fE6bfAOeyWALngNUWq84654DeUbvM4d8dL2eGkbvKNeR7iGIDXcSVe4hvINXPqP7oR3TNdMpjyR7MzxWuV-cWGw8JVVF4zkmQjYs-BBaVMdV6WufBEyqaqvpDrMk-pTEfBPMdGKbb3Y1oqdMzG5FDv62EO9NdWqB3rxQCs2OA9vaXqkFY_1zmOteKIXT26JTQw7Zloy_ZYRw5qRm_aMGBaNGsz1m0YMq0b0u3aZfXf673cAAAD__1h174Y=
+https://cockroachdb.github.io/distsqlplan/decode.html#eJzElk9v4jwQh-_vp7Dm1Oo1CnbCn-bEbstKSF3oAoddVRxSYrVIFLN2kKCI775KgpRCykxIkLgR4PGM5_dolC3Yv3PwYdR97N6P2crM2Y_h4Cd77v5-evzW67Obh95oPPr1eMv2f4mfe_37MRv02c2asw1nH7ds_yGF15sPNhg-dIfs-x-2ngCHhQ5VP3hXFvxnEMBBAgcXOHjAoQETDkujp8pabeK_bBOgF67Br3OYLZarKP56wmGqjQJ_C9EsmivwYRy8zNVQBaEyTh04hCoKZvOkzHrz0Vma2XtgNsBhtAwW1mc1J648WEU-60jecXnHg8mOg15FWREbBa8KfLHjxRsZaRMp4zQOe-iI_08eL885_mFmo9liGjmiflyBJxeJb2VCZVTos444Kpqd87Jhb4F9O3HGZJe1555sLztNpxW_ujN6h9yUCtwh68y7eGd9XdNLRx6O9lT5xkF5UdxPUcBPR9Ycr6ShRCt7Q5tlDSWOz9IVVzFUXNTQ3JQqGCqLKyKLKOLVkhDLKEK0slekVVYR4vhsvPIqisiLKpKbUgVF3OKKuEUUadaSEMsoQrSyV6RdVhHi-Gy87lUUcS-qSG5KFRTxiiviFVGkVSspCNHIXpC7soIQx2fD9a4iiHdRQXJTutCL0BdFh8ou9cKqQu849bhtFb6q9JpWr8xUPRk9Tcqkj4OES74IlY3SX0X60FukP8UNfoYFCksclsew-Ay7B7A4D25XgYWsRDer0LKO0y46cA-HPTwtIusGSjdxuInCLRxuVREFhwlRcJgShaAJUXCaEqVdRZQ7fCfUiaVArBRqp-SWyjlxEzSRN0FTgVM4kTiBU5GL3Go5J3OBrxbhEanhy0U0CDy3Xc4KHaep0HGaDJ3AqdBxnAwd36xU6Lklc5ham0gN3zLijsBze-as0HGaCh2nydAJnAodx6nQJb5hj0Of7P77FwAA__8PIyWL
 
 # Ensure that even with more ordering columns, ordering propagates past local
 # DISTINCT processors.
@@ -119,7 +119,7 @@ distinct        ·             ·            (x, y)  +y
 query T
 SELECT url FROM [EXPLAIN (DISTSQL) SELECT DISTINCT ON (y) x, y FROM xyz ORDER BY y, x]
 ----
-https://cockroachdb.github.io/distsqlplan/decode.html#eJyslV9v2jAUxd_3Kaz71KpGwU6gkCe2wiQkBh3wsKniISNWFYnGme1ISRHffcofKYMttml4dJKTc-7vHiVHkL8P4MNmtpg9bVEqDujrevUNvcx-PC8-z5fobjrfbDffF_eofqQ4z5dPW7Raorv8HmUY5ZUmy9_Raj2drdGXnyjHKNsBhpiHbBm8MQn-CxDAQAGDCxg8wDCAHYZE8D2TkovikWMpmIcZ-H0MUZykqri8w7DngoF_BBWpAwMftsGvA1uzIGTC6QOGkKkgOpQ2Wf4-SUT0FogcMGySIJY-6jmF8ypVPppQPHFhd8LAU9U4SBW8MvDJCdun2HChmHAG5wEm9AFPyEOrBb3GYhpJFcV75ZD-pUsxkAiZYGExVKud22rXuPDqPW1zaHP9M71lLu8sF7HfN7HYt0N7jveRjRty1Bsfdtm4waIhS26ycWpPltqQ9Xrl-FeTNeSoyT52IWuwaMjSm5B17cm6NmSHvXL8q8kactRkR13IGiwasu5NyHr2ZD0bso-9j3A1pKi5jrtwNVg0XL2bf_3_Y7dmMuGxZGdebW_uF78FFr6y6lcieSr27FnwfWlTHVelrrwQMqmqu6Q6zOPqVhHwbzHRiqleTLVi90xMLsWuPvZQb-1p1QO9eKAVG5yHXYZ-1IpHeueRVjzWi8ddYhNDx0wl07eMGGpGOvWMGIrmGcz1TSOGqhF91y6z706f_gQAAP__doXoLw==
+https://cockroachdb.github.io/distsqlplan/decode.html#eJy8lk1v2kAQhu_9Fas5Jcois2vzEZ9oEyohpZACh1YRBwevEiSC3V0j4UT898o2kgMOMzZGHP3xeGbnfTTyB5h_S3Bh0n_o303ZWi_Zz_HoF3vq_3l8-D4Ysqv7wWQ6-f1wzXavJNeD4d2UjYbsKr5mG87ijNnE72w0vu-P2Y-_LOZsMwMOq8BXQ-9NGXCfQAAHCRxs4OAAhxbMOIQ6mCtjAp288pECA38DbpPDYhWuo-T2jMM80ArcD4gW0VKBC1PveanGyvOVtprAwVeRt1imZTbxey_UizdPx8BhEnor47KGlVQerSOX9STv2TDbcgjWUV7BRN6LAldsefkuJoGOlLZa-w305A3viZujJWSVEvcLEy1W88gSzcMqyYG0r7Tyk0MdlMu_8ByzV8-8FujZNm_JPtpS_p0gq3XsrGjvhQmhved9OTX6-rqnYdAIQkvuD_NY-dZeeVFeSVFCSUs2LOcUKYk-dlK260hJlMiDFReTUpxdysKETpJSlrdClrHCaaTRVbaC6GNnRaeOFUSJfLLyYlbIs1tRmNBJVtjlrbDLWNFupNFVtoLoY2dFt44VRIl8svbFrLDPbkVhQidZ4ZS3wiljRadxihNEFzsnbus4QZTI5-pczAnn7E4UJlT7p-aLcmNlwmBlVKn_lWbSsPJfVHZIE6z1XD3qYJ6WyS5HKZfe8JWJsqciuxisskdJg59hgcISh-UhLD7D9h4sqsHdOrCQteh2HVo2cdpGB-7gsIOnRWTdQuk2DrdRuIPDnTqi4DAhCg5TohA0IQpOU6J064hyi--EJrEUiJVC7ZTCUqkSN0ETeRM0FTiFE4kTOBW5KKyWKpkLfLUIh0gNXy6iReCF7VIpdJymQsdpMnQCp0LHcTJ0fLNSoReWzH5qXSI1fMuIWwIv7JlKoeM0FTpOk6ETOBU6jlOhS3zDHoY-2377HwAA__8VCBZB
 
 # Distinct processors elided becaue of strong key.
 query TTTTT
@@ -154,4 +154,4 @@ render          ·             ·            (a, b)     +a,+b
 query T
 SELECT url FROM [EXPLAIN (DISTSQL) SELECT DISTINCT ON (a, b) a, b FROM abc ORDER BY a, b, c]
 ----
-https://cockroachdb.github.io/distsqlplan/decode.html#eJy0lFFvmzAUhd_3K6z71KqOEgNJU56yNZkUiUFHMmlTw4MDVodEMbONtCnKf5-AiSRsMa6SvVi6F47P4bvo7kD-yMCF1cJbPK5RKTL0MQw-oefF1yfv_dJHN_Plar367N2iP69U9dJ_XKPARzcUo-0tqs5GRrcxCsL5IkQfvtVtjOIIMOQ8YT59ZRLcZyCAwQIMNmBwAMMYIgyF4DGTkovqlV0tWCY_wR1hSPOiVFU7whBzwcDdgUpVxsCFNd1mLGQ0YWJY3ZswRdOstqHbeFaI9JWKX4BhVdBcumjof_G8wzEYboBsoD2tDUC0x8BLdfCUir4wcMkem-eap1KleayGk9NQM4Jn1bcHImGCJS5qGuc8rbOeByveXNX1ucMz6w7P7DuI9rqAZGSUEENQqr649klcYj46y2R0f41qcFy8aXo90Vo491ecnmWOw34LDusYR1PY9ekY4-iJ1uKYXhGHbY7DMcPR-fbBcTHcwNgYR0-0FsfDFXE45jhGBjgGnUWHaJ4ggrj6zoQhhZ5ELYXxf9pw__AMmSx4LtmJ47mbR9XaY8kLa9al5KWI2ZPgcW3TlEGtqxsJk6p5SppimTePqoDHYqIVWydi0hVbeucea1urdvRi55LcY614oneeXOJ8rxVP9c7TS5wf9LMa9fwm-p-s6x3t3_0OAAD__y1-A6s=
+https://cockroachdb.github.io/distsqlplan/decode.html#eJy8lk9v4jwQxu_vp7Dm1KpG4PyhbU6827ISEgtdYKVdFQ4msVokGrN2kLaq-O6rJCsgFDxGVrhYGifP-Bn_RpN8gP69hAjG3X73YULWakm-jobfyHP351P__96AXD32xpPx9_41-fdKHvcGDxMyHJArTsn8muRrKePzmAxHj90R-fKr2KYkngGFVCZiwN-EhugZGFDwgIIPFAKgEMKMwkrJWGgtVf7KRyHoJX8galFYpKt1lm_PKMRSCYg-IFtkSwERTPh8KUaCJ0I187yJyPhiWRzD53FnpRZvXL0DhfGKpzoizcGPfn-3NJpTYFPYrt4UYLahINfZ7kyd8RcBEdtQe1-PC50t0jhrtqumOox28tqHKhFKJBEpN6pn7tLM38kr16_Hcsw2O2veSWu7VLI88TDVDe14N7Tj35QJT9bBmFUhFIbr7HhVO7u-o93jVgeyIVdNFlYkpywEFQvMvsk8myb71FSN_eCsPkOsbfncXr7PWC195tXVZ549ZP8cyN4-5DLwizWwhoxY217O3eUhe7VA9uuC7NtDDuwgHxBt7AfNKYTWkBFr28u5vzxkvxbIQV2QA3vILQvIjYOfAsLThDAis1ehLNkijrZ3El6ebVAL29Yl_gaOOBgJvZKpFlYf-VZeg0heRFm7lmsViycl4-KYMhwWumIjETorn7Iy6KXlo9zgvpgdillFHFbU7Dy15yJuu4jvXcQM8e0Z79s3i30zrLaZVmBUh2Zx6ITarEauzCxGUJvFCGqkaMR32wX1rVF8Z4Z15wTLrEaKNosRWGYxAgspGvF97wKLIVMUG6OOc9RtkLpNUrdR6jZLmdMwZcg0DRBon8bpedDMcqxysxqDZlZj0JDCMeufhqoR2mzz398AAAD__w5OR9U=
diff --git a/pkg/sql/opt/exec/execbuilder/testdata/lookup_join b/pkg/sql/opt/exec/execbuilder/testdata/lookup_join
index f80f1394b7df..937b27397869 100644
--- a/pkg/sql/opt/exec/execbuilder/testdata/lookup_join
+++ b/pkg/sql/opt/exec/execbuilder/testdata/lookup_join
@@ -285,7 +285,7 @@ NULL       NULL     {5}       5
 query T
 SELECT url FROM [EXPLAIN (DISTSQL) SELECT DISTINCT authors.name FROM books AS b1, books2 AS b2, authors WHERE b1.title = b2.title AND authors.book = b1.title AND b1.shelf <> b2.shelf]
 ----
-https://cockroachdb.github.io/distsqlplan/decode.html#eJyck9Fv2j4Qx99_f4V_95RIbsFO6KRIlVwVpjKx0AHSJlU8JOQKWYOd2Y60CfG_T0noSlATxt58vvucv_e1vQPzI4MA5qPJ6H5BCp2Rj7PpZ_I0-vY4uRuHxBmO54v5l4lLDiVlPA7vFyQq7EZpcy2jLdZQrNSLIXdzEjNaB7yKOH0tJl8fRrMRcZyYXdvUZkhuSczrpUvuwiFxXtuWfJk9FLqHdMyuzQazZ_J_RVZrdwkUpEowjLZoIHgCBhQGsKSQa7VCY5Qut3dV0Tj5CUGfQirzwpbbSworpRGCHVRHQQChulJ5bwAUErRRmlVlewqqsG-QsdEaIbjZ06PGrLvxIooznGGUoO71G-2hMkzkOt1G-hdQmOeRNAG5AgrTwgZEMCo8aFPB_lUFa6o42H9GB2_VwVt1vB1fSKUT1Jic-nu-5J1hHiKz-aRSibrHm7Nk-Gwdwdxbna431hHcbcxAhUeF3zqJd4mjpYCDod4718qP_Jwo9VLk5LtKJVEyIMIvVYVE8PJNiw9_NLZftn-JtGFqbCpXtuc3hQnW2n_Q6H_mr8zQ5Eoa_KvP0i9vEJM11i_CqEKv8FGrVXVMHU4rrtpI0Ng6e1MHY1mnSoHHMOuEvQbMTmF-AcxPYa8T9rtl-53woBsedML9E3i5_-93AAAA___haelL
+https://cockroachdb.github.io/distsqlplan/decode.html#eJyck1Fv2jAQx9_3Kbx7SiS34CRQKVIlo8JUJhY6QNqkioeEXCFrsDPbkTYhvvuUhK4ENaH0zee739__O9s70L9T8GE-mozuFiRXKfkym34jj6OfD5PBOCDWcDxfzL9PbHIoKeJxcLcgYW42UulrEW6xgiIpnzUZzEnEaBU4ZeTQl2Ly4340GxHLiti1SUyK5JZETrW0ySAYEutFtuCL7KHQPqQjdq03mD6RzyVZru0lUBAyxiDcogb_ERhQ6MGSQqbkCrWWqtjelUXj-A_4XQqJyHJTbC8prKRC8HdQHgU-BPJKZp0-UIjRhElalu0pyNy8QtqEawT_Zk-PhFm78CKMUpxhGKPqdGvyUA6MZyrZhuovUJhnodA-uQIK09z4hDPKXWhywT7qgtVdHMZ_xofT6MNp9PF6fC6kilFhfDrf8yVvNHMf6s1XmQhUHafeS4pPxuLMvlXJemMs7ti1Hih3KfcaO3EvmWhh4DBQ941rdY7mOZHyOc_IL5kIIoVPuFe4Cgh3ijfNb_57bL5s7xJrw0SbRKxMx6sb46xRv_ch_d679fs1_TN_cYY6k0Ljuz5jt3ghGK-xenFa5mqFD0quymOqcFpy5UaM2lTZmyoYiypVGDyGWSvs1mB2CjsXwM4p7LbCXrttrxXutcO9VrjfDvdb4e4JvNx_-hcAAP__NdMdAw==
 
 query TTTTT colnames
 EXPLAIN (VERBOSE) SELECT a.name FROM authors AS a JOIN books2 AS b2 ON a.book = b2.title ORDER BY a.name
diff --git a/pkg/sql/physicalplan/physical_plan.go b/pkg/sql/physicalplan/physical_plan.go
index 90ff200886c0..3091deebc2bd 100644
--- a/pkg/sql/physicalplan/physical_plan.go
+++ b/pkg/sql/physicalplan/physical_plan.go
@@ -1111,10 +1111,64 @@ func (p *PhysicalPlan) AddJoinStage(
 	}
 }
 
+// AddStageOnNodes adds a stage of processors that take in a single input
+// logical stream on the specified nodes and connects them to the previous
+// stage via a hash router.
+func (p *PhysicalPlan) AddStageOnNodes(
+	nodes []roachpb.NodeID,
+	core execinfrapb.ProcessorCoreUnion,
+	post execinfrapb.PostProcessSpec,
+	hashCols []uint32,
+	types []*types.T,
+	mergeOrd execinfrapb.Ordering,
+	routers []ProcessorIdx,
+) {
+	pIdxStart := len(p.Processors)
+	newStageID := p.NewStageOnNodes(nodes)
+
+	for _, n := range nodes {
+		proc := Processor{
+			Node: n,
+			Spec: execinfrapb.ProcessorSpec{
+				Input: []execinfrapb.InputSyncSpec{
+					{ColumnTypes: types},
+				},
+				Core:    core,
+				Post:    post,
+				Output:  []execinfrapb.OutputRouterSpec{{Type: execinfrapb.OutputRouterSpec_PASS_THROUGH}},
+				StageID: newStageID,
+			},
+		}
+		p.AddProcessor(proc)
+	}
+
+	if len(nodes) > 1 {
+		// Set up the routers.
+		for _, resultProc := range routers {
+			p.Processors[resultProc].Spec.Output[0] = execinfrapb.OutputRouterSpec{
+				Type:        execinfrapb.OutputRouterSpec_BY_HASH,
+				HashColumns: hashCols,
+			}
+		}
+	}
+
+	// Connect the result streams to the processors.
+	for bucket := 0; bucket < len(nodes); bucket++ {
+		pIdx := ProcessorIdx(pIdxStart + bucket)
+		p.MergeResultStreams(routers, bucket, mergeOrd, pIdx, 0)
+	}
+
+	// Set the new result routers.
+	p.ResultRouters = p.ResultRouters[:0]
+	for i := 0; i < len(nodes); i++ {
+		p.ResultRouters = append(p.ResultRouters, ProcessorIdx(pIdxStart+i))
+	}
+}
+
 // AddDistinctSetOpStage creates a distinct stage and a join stage to implement
 // INTERSECT and EXCEPT plans.
 //
-// TODO(abhimadan): If there's a strong key on the left or right side, we
+// TODO(yuzefovich): If there's a strong key on the left or right side, we
 // can elide the distinct stage on that side.
 func (p *PhysicalPlan) AddDistinctSetOpStage(
 	nodes []roachpb.NodeID,
@@ -1126,65 +1180,28 @@ func (p *PhysicalPlan) AddDistinctSetOpStage(
 	leftMergeOrd, rightMergeOrd execinfrapb.Ordering,
 	leftRouters, rightRouters []ProcessorIdx,
 ) {
-	const numSides = 2
-	inputResultTypes := [numSides][]*types.T{leftTypes, rightTypes}
-	inputMergeOrderings := [numSides]execinfrapb.Ordering{leftMergeOrd, rightMergeOrd}
-	inputResultRouters := [numSides][]ProcessorIdx{leftRouters, rightRouters}
-
 	// Create distinct stages for the left and right sides, where left and right
 	// sources are sent by hash to the node which will contain the join processor.
 	// The distinct stage must be before the join stage for EXCEPT queries to
 	// produce correct results (e.g., (VALUES (1),(1),(2)) EXCEPT (VALUES (1))
 	// would return (1),(2) instead of (2) if there was no distinct processor
 	// before the EXCEPT ALL join).
-	distinctIdxStart := len(p.Processors)
 	distinctProcs := make(map[roachpb.NodeID][]ProcessorIdx)
-
-	for side, types := range inputResultTypes {
-		distinctStageID := p.NewStageOnNodes(nodes)
-		for _, n := range nodes {
-			proc := Processor{
-				Node: n,
-				Spec: execinfrapb.ProcessorSpec{
-					Input: []execinfrapb.InputSyncSpec{
-						{ColumnTypes: types},
-					},
-					Core:    distinctCores[side],
-					Post:    execinfrapb.PostProcessSpec{},
-					Output:  []execinfrapb.OutputRouterSpec{{Type: execinfrapb.OutputRouterSpec_PASS_THROUGH}},
-					StageID: distinctStageID,
-				},
-			}
-			pIdx := p.AddProcessor(proc)
-			distinctProcs[n] = append(distinctProcs[n], pIdx)
-		}
-	}
-
-	if len(nodes) > 1 {
-		// Set up the left routers.
-		for _, resultProc := range leftRouters {
-			p.Processors[resultProc].Spec.Output[0] = execinfrapb.OutputRouterSpec{
-				Type:        execinfrapb.OutputRouterSpec_BY_HASH,
-				HashColumns: eqCols,
-			}
-		}
-		// Set up the right routers.
-		for _, resultProc := range rightRouters {
-			p.Processors[resultProc].Spec.Output[0] = execinfrapb.OutputRouterSpec{
-				Type:        execinfrapb.OutputRouterSpec_BY_HASH,
-				HashColumns: eqCols,
-			}
-		}
+	p.AddStageOnNodes(
+		nodes, distinctCores[0], execinfrapb.PostProcessSpec{}, eqCols,
+		leftTypes, leftMergeOrd, leftRouters,
+	)
+	for _, leftDistinctProcIdx := range p.ResultRouters {
+		node := p.Processors[leftDistinctProcIdx].Node
+		distinctProcs[node] = append(distinctProcs[node], leftDistinctProcIdx)
 	}
-
-	// Connect the left and right streams to the distinct processors.
-	for side, routers := range inputResultRouters {
-		// Get the processor index offset for the current side.
-		sideOffset := side * len(nodes)
-		for bucket := 0; bucket < len(nodes); bucket++ {
-			pIdx := ProcessorIdx(distinctIdxStart + sideOffset + bucket)
-			p.MergeResultStreams(routers, bucket, inputMergeOrderings[side], pIdx, 0)
-		}
+	p.AddStageOnNodes(
+		nodes, distinctCores[1], execinfrapb.PostProcessSpec{}, eqCols,
+		rightTypes, rightMergeOrd, rightRouters,
+	)
+	for _, rightDistinctProcIdx := range p.ResultRouters {
+		node := p.Processors[rightDistinctProcIdx].Node
+		distinctProcs[node] = append(distinctProcs[node], rightDistinctProcIdx)
 	}
 
 	// Create a join stage, where the distinct processors on the same node are