diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/SubqueryPlanner.java b/core/trino-main/src/main/java/io/trino/sql/planner/SubqueryPlanner.java
index 4d9656b37094..3ed9583276f3 100644
--- a/core/trino-main/src/main/java/io/trino/sql/planner/SubqueryPlanner.java
+++ b/core/trino-main/src/main/java/io/trino/sql/planner/SubqueryPlanner.java
@@ -268,7 +268,7 @@ private PlanBuilder planScalarSubquery(PlanBuilder subPlan, Cluster<SubqueryExpr
                 subPlan,
                 root,
                 scalarSubquery.getQuery(),
-                CorrelatedJoinNode.Type.INNER,
+                CorrelatedJoinNode.Type.LEFT,
                 TRUE_LITERAL,
                 mapAll(cluster, subPlan.getScope(), column));
     }
diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/TransformCorrelatedScalarSubquery.java b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/TransformCorrelatedScalarSubquery.java
index 42de46b02c05..79d00184f2c0 100644
--- a/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/TransformCorrelatedScalarSubquery.java
+++ b/core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/TransformCorrelatedScalarSubquery.java
@@ -42,6 +42,7 @@
 import static io.trino.sql.planner.LogicalPlanner.failFunction;
 import static io.trino.sql.planner.optimizations.PlanNodeSearcher.searchFrom;
 import static io.trino.sql.planner.optimizations.QueryCardinalityUtil.extractCardinality;
+import static io.trino.sql.planner.plan.CorrelatedJoinNode.Type.INNER;
 import static io.trino.sql.planner.plan.CorrelatedJoinNode.Type.LEFT;
 import static io.trino.sql.planner.plan.Patterns.CorrelatedJoin.correlation;
 import static io.trino.sql.planner.plan.Patterns.CorrelatedJoin.filter;
@@ -123,7 +124,7 @@ public Result apply(CorrelatedJoinNode correlatedJoinNode, Captures captures, Co
                     correlatedJoinNode.getInput(),
                     rewrittenSubquery,
                     correlatedJoinNode.getCorrelation(),
-                    producesSingleRow ? correlatedJoinNode.getType() : LEFT,
+                    producesSingleRow ? INNER : correlatedJoinNode.getType(),
                     correlatedJoinNode.getFilter(),
                     correlatedJoinNode.getOriginSubquery()));
         }
diff --git a/core/trino-main/src/test/java/io/trino/sql/query/TestCorrelatedAggregation.java b/core/trino-main/src/test/java/io/trino/sql/query/TestCorrelatedAggregation.java
index f25dcb53af9b..c3dc4fae2a49 100644
--- a/core/trino-main/src/test/java/io/trino/sql/query/TestCorrelatedAggregation.java
+++ b/core/trino-main/src/test/java/io/trino/sql/query/TestCorrelatedAggregation.java
@@ -482,4 +482,34 @@ public void testChecksum()
                 "ON TRUE"))
                 .matches("VALUES (1, null), (2, x'd0f70cebd131ec61')");
     }
+
+    @Test
+    public void testCorrelatedSubqueryWithGroupedAggregation()
+    {
+        assertThat(assertions.query("WITH" +
+                "    t(k, v) AS (VALUES ('A', 1), ('B', NULL), ('C', 2), ('D', 3)), " +
+                "    u(k, v) AS (VALUES (1, 10), (1, 20), (2, 30)) " +
+                "SELECT" +
+                "    k," +
+                "    (" +
+                "        SELECT max(v) FROM u WHERE t.v = u.k GROUP BY k" +
+                "     ) AS cols " +
+                "FROM t"))
+                .matches("VALUES ('A', 20), ('B', NULL), ('C', 30), ('D', NULL)");
+    }
+
+    @Test
+    public void testCorrelatedSubqueryWithGlobalAggregation()
+    {
+        assertThat(assertions.query("WITH" +
+                "    t(k, v) AS (VALUES ('A', 1), ('B', NULL), ('C', 2), ('D', 3)), " +
+                "    u(k, v) AS (VALUES (1, 10), (1, 20), (2, 30)) " +
+                "SELECT" +
+                "    k," +
+                "    (" +
+                "        SELECT max(v) FROM u WHERE t.v = u.k" +
+                "     ) AS cols " +
+                "FROM t"))
+                .matches("VALUES ('A', 20), ('B', NULL), ('C', 30), ('D', NULL)");
+    }
 }
diff --git a/core/trino-main/src/test/java/io/trino/sql/query/TestSubqueries.java b/core/trino-main/src/test/java/io/trino/sql/query/TestSubqueries.java
index e6a933517cf9..e3e31117fda2 100644
--- a/core/trino-main/src/test/java/io/trino/sql/query/TestSubqueries.java
+++ b/core/trino-main/src/test/java/io/trino/sql/query/TestSubqueries.java
@@ -46,7 +46,7 @@
 import static io.trino.sql.planner.plan.AggregationNode.Step.FINAL;
 import static io.trino.sql.planner.plan.AggregationNode.Step.PARTIAL;
 import static io.trino.sql.planner.plan.AggregationNode.Step.SINGLE;
-import static io.trino.sql.planner.plan.JoinNode.Type.INNER;
+import static io.trino.sql.planner.plan.JoinNode.Type.LEFT;
 import static io.trino.testing.TestingHandles.TEST_CATALOG_NAME;
 import static io.trino.testing.TestingSession.testSessionBuilder;
 import static java.lang.String.format;
@@ -204,7 +204,7 @@ public void testCorrelatedSubqueriesWithTopN()
                 "SELECT (SELECT t.a FROM (VALUES 1, 2, 3) t(a) WHERE t.a = t2.b ORDER BY a LIMIT 1) FROM (VALUES 1.0, 2.0) t2(b)",
                 "VALUES 1, 2",
                 output(
-                        join(INNER, builder -> builder
+                        join(LEFT, builder -> builder
                                 .equiCriteria("cast_b", "cast_a")
                                 .left(
                                         project(
@@ -228,7 +228,7 @@ public void testCorrelatedSubqueriesWithTopN()
                 "SELECT (SELECT t.a FROM (VALUES 1, 2, 3, 4, 5) t(a) WHERE t.a = t2.b * t2.c - 1 ORDER BY a LIMIT 1) FROM (VALUES (1, 2), (2, 3)) t2(b, c)",
                 "VALUES 1, 5",
                 output(
-                        join(INNER, builder -> builder
+                        join(LEFT, builder -> builder
                                 .equiCriteria("expr", "a")
                                 .left(
                                         project(
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/partitioned/q09.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/partitioned/q09.plan.txt
index 4466a2d88feb..cbcd78f04da6 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/partitioned/q09.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/partitioned/q09.plan.txt
@@ -1,93 +1,122 @@
-cross join:
-    cross join:
-        cross join:
-            cross join:
-                cross join:
-                    cross join:
-                        cross join:
-                            cross join:
-                                cross join:
-                                    cross join:
-                                        cross join:
-                                            cross join:
-                                                cross join:
-                                                    cross join:
-                                                        cross join:
-                                                            final aggregation over ()
-                                                                local exchange (GATHER, SINGLE, [])
-                                                                    remote exchange (GATHER, SINGLE, [])
-                                                                        partial aggregation over ()
-                                                                            scan store_sales
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    scan reason
-                                                        final aggregation over ()
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    partial aggregation over ()
-                                                                        scan store_sales
+join (LEFT, REPLICATED):
+    join (LEFT, REPLICATED):
+        join (LEFT, REPLICATED):
+            join (LEFT, REPLICATED):
+                join (LEFT, REPLICATED):
+                    join (LEFT, REPLICATED):
+                        join (LEFT, REPLICATED):
+                            join (RIGHT, PARTITIONED):
+                                remote exchange (REPARTITION, HASH, [])
+                                    final aggregation over ()
+                                        local exchange (GATHER, SINGLE, [])
+                                            remote exchange (GATHER, SINGLE, [])
+                                                partial aggregation over ()
+                                                    scan store_sales
+                                local exchange (GATHER, SINGLE, [])
+                                    join (RIGHT, PARTITIONED):
+                                        remote exchange (REPARTITION, HASH, [])
+                                            final aggregation over ()
+                                                local exchange (GATHER, SINGLE, [])
+                                                    remote exchange (GATHER, SINGLE, [])
+                                                        partial aggregation over ()
+                                                            scan store_sales
+                                        local exchange (GATHER, SINGLE, [])
+                                            join (RIGHT, PARTITIONED):
+                                                remote exchange (REPARTITION, HASH, [])
                                                     final aggregation over ()
                                                         local exchange (GATHER, SINGLE, [])
                                                             remote exchange (GATHER, SINGLE, [])
                                                                 partial aggregation over ()
                                                                     scan store_sales
-                                                final aggregation over ()
-                                                    local exchange (GATHER, SINGLE, [])
-                                                        remote exchange (GATHER, SINGLE, [])
-                                                            partial aggregation over ()
-                                                                scan store_sales
-                                            final aggregation over ()
                                                 local exchange (GATHER, SINGLE, [])
-                                                    remote exchange (GATHER, SINGLE, [])
-                                                        partial aggregation over ()
-                                                            scan store_sales
-                                        final aggregation over ()
-                                            local exchange (GATHER, SINGLE, [])
-                                                remote exchange (GATHER, SINGLE, [])
-                                                    partial aggregation over ()
-                                                        scan store_sales
+                                                    join (RIGHT, PARTITIONED):
+                                                        remote exchange (REPARTITION, HASH, [])
+                                                            final aggregation over ()
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                        partial aggregation over ()
+                                                                            scan store_sales
+                                                        local exchange (GATHER, SINGLE, [])
+                                                            join (RIGHT, PARTITIONED):
+                                                                remote exchange (REPARTITION, HASH, [])
+                                                                    final aggregation over ()
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                partial aggregation over ()
+                                                                                    scan store_sales
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    join (RIGHT, PARTITIONED):
+                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                            final aggregation over ()
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                        partial aggregation over ()
+                                                                                            scan store_sales
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            join (RIGHT, PARTITIONED):
+                                                                                remote exchange (REPARTITION, HASH, [])
+                                                                                    final aggregation over ()
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                                partial aggregation over ()
+                                                                                                    scan store_sales
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    join (RIGHT, PARTITIONED):
+                                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                                            final aggregation over ()
+                                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                                        partial aggregation over ()
+                                                                                                            scan store_sales
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (REPARTITION, HASH, [])
+                                                                                                scan reason
+                            local exchange (GATHER, SINGLE, [])
+                                remote exchange (REPLICATE, BROADCAST, [])
                                     final aggregation over ()
                                         local exchange (GATHER, SINGLE, [])
                                             remote exchange (GATHER, SINGLE, [])
                                                 partial aggregation over ()
                                                     scan store_sales
+                        local exchange (GATHER, SINGLE, [])
+                            remote exchange (REPLICATE, BROADCAST, [])
                                 final aggregation over ()
                                     local exchange (GATHER, SINGLE, [])
                                         remote exchange (GATHER, SINGLE, [])
                                             partial aggregation over ()
                                                 scan store_sales
+                    local exchange (GATHER, SINGLE, [])
+                        remote exchange (REPLICATE, BROADCAST, [])
                             final aggregation over ()
                                 local exchange (GATHER, SINGLE, [])
                                     remote exchange (GATHER, SINGLE, [])
                                         partial aggregation over ()
                                             scan store_sales
+                local exchange (GATHER, SINGLE, [])
+                    remote exchange (REPLICATE, BROADCAST, [])
                         final aggregation over ()
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (GATHER, SINGLE, [])
                                     partial aggregation over ()
                                         scan store_sales
+            local exchange (GATHER, SINGLE, [])
+                remote exchange (REPLICATE, BROADCAST, [])
                     final aggregation over ()
                         local exchange (GATHER, SINGLE, [])
                             remote exchange (GATHER, SINGLE, [])
                                 partial aggregation over ()
                                     scan store_sales
+        local exchange (GATHER, SINGLE, [])
+            remote exchange (REPLICATE, BROADCAST, [])
                 final aggregation over ()
                     local exchange (GATHER, SINGLE, [])
                         remote exchange (GATHER, SINGLE, [])
                             partial aggregation over ()
                                 scan store_sales
+    local exchange (GATHER, SINGLE, [])
+        remote exchange (REPLICATE, BROADCAST, [])
             final aggregation over ()
                 local exchange (GATHER, SINGLE, [])
                     remote exchange (GATHER, SINGLE, [])
                         partial aggregation over ()
                             scan store_sales
-        final aggregation over ()
-            local exchange (GATHER, SINGLE, [])
-                remote exchange (GATHER, SINGLE, [])
-                    partial aggregation over ()
-                        scan store_sales
-    final aggregation over ()
-        local exchange (GATHER, SINGLE, [])
-            remote exchange (GATHER, SINGLE, [])
-                partial aggregation over ()
-                    scan store_sales
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/partitioned/q54.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/partitioned/q54.plan.txt
index a38e81074509..d524385f72f4 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/partitioned/q54.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/partitioned/q54.plan.txt
@@ -8,8 +8,8 @@ local exchange (GATHER, SINGLE, [])
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
                                     partial aggregation over (ss_customer_sk)
-                                        cross join:
-                                            cross join:
+                                        join (LEFT, REPLICATED):
+                                            join (LEFT, REPLICATED):
                                                 join (INNER, REPLICATED):
                                                     join (INNER, REPLICATED):
                                                         dynamic filter (["ss_customer_sk", "ss_sold_date_sk"])
@@ -48,8 +48,7 @@ local exchange (GATHER, SINGLE, [])
                                                                                                             scan date_dim
                                                     local exchange (GATHER, SINGLE, [])
                                                         remote exchange (REPLICATE, BROADCAST, [])
-                                                            dynamic filter (["d_month_seq_26"])
-                                                                scan date_dim
+                                                            scan date_dim
                                                 local exchange (GATHER, SINGLE, [])
                                                     remote exchange (REPLICATE, BROADCAST, [])
                                                         local exchange (GATHER, SINGLE, [])
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/unpartitioned/q09.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/unpartitioned/q09.plan.txt
index 4466a2d88feb..cbcd78f04da6 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/unpartitioned/q09.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/unpartitioned/q09.plan.txt
@@ -1,93 +1,122 @@
-cross join:
-    cross join:
-        cross join:
-            cross join:
-                cross join:
-                    cross join:
-                        cross join:
-                            cross join:
-                                cross join:
-                                    cross join:
-                                        cross join:
-                                            cross join:
-                                                cross join:
-                                                    cross join:
-                                                        cross join:
-                                                            final aggregation over ()
-                                                                local exchange (GATHER, SINGLE, [])
-                                                                    remote exchange (GATHER, SINGLE, [])
-                                                                        partial aggregation over ()
-                                                                            scan store_sales
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    scan reason
-                                                        final aggregation over ()
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    partial aggregation over ()
-                                                                        scan store_sales
+join (LEFT, REPLICATED):
+    join (LEFT, REPLICATED):
+        join (LEFT, REPLICATED):
+            join (LEFT, REPLICATED):
+                join (LEFT, REPLICATED):
+                    join (LEFT, REPLICATED):
+                        join (LEFT, REPLICATED):
+                            join (RIGHT, PARTITIONED):
+                                remote exchange (REPARTITION, HASH, [])
+                                    final aggregation over ()
+                                        local exchange (GATHER, SINGLE, [])
+                                            remote exchange (GATHER, SINGLE, [])
+                                                partial aggregation over ()
+                                                    scan store_sales
+                                local exchange (GATHER, SINGLE, [])
+                                    join (RIGHT, PARTITIONED):
+                                        remote exchange (REPARTITION, HASH, [])
+                                            final aggregation over ()
+                                                local exchange (GATHER, SINGLE, [])
+                                                    remote exchange (GATHER, SINGLE, [])
+                                                        partial aggregation over ()
+                                                            scan store_sales
+                                        local exchange (GATHER, SINGLE, [])
+                                            join (RIGHT, PARTITIONED):
+                                                remote exchange (REPARTITION, HASH, [])
                                                     final aggregation over ()
                                                         local exchange (GATHER, SINGLE, [])
                                                             remote exchange (GATHER, SINGLE, [])
                                                                 partial aggregation over ()
                                                                     scan store_sales
-                                                final aggregation over ()
-                                                    local exchange (GATHER, SINGLE, [])
-                                                        remote exchange (GATHER, SINGLE, [])
-                                                            partial aggregation over ()
-                                                                scan store_sales
-                                            final aggregation over ()
                                                 local exchange (GATHER, SINGLE, [])
-                                                    remote exchange (GATHER, SINGLE, [])
-                                                        partial aggregation over ()
-                                                            scan store_sales
-                                        final aggregation over ()
-                                            local exchange (GATHER, SINGLE, [])
-                                                remote exchange (GATHER, SINGLE, [])
-                                                    partial aggregation over ()
-                                                        scan store_sales
+                                                    join (RIGHT, PARTITIONED):
+                                                        remote exchange (REPARTITION, HASH, [])
+                                                            final aggregation over ()
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                        partial aggregation over ()
+                                                                            scan store_sales
+                                                        local exchange (GATHER, SINGLE, [])
+                                                            join (RIGHT, PARTITIONED):
+                                                                remote exchange (REPARTITION, HASH, [])
+                                                                    final aggregation over ()
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                partial aggregation over ()
+                                                                                    scan store_sales
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    join (RIGHT, PARTITIONED):
+                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                            final aggregation over ()
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                        partial aggregation over ()
+                                                                                            scan store_sales
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            join (RIGHT, PARTITIONED):
+                                                                                remote exchange (REPARTITION, HASH, [])
+                                                                                    final aggregation over ()
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                                partial aggregation over ()
+                                                                                                    scan store_sales
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    join (RIGHT, PARTITIONED):
+                                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                                            final aggregation over ()
+                                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                                        partial aggregation over ()
+                                                                                                            scan store_sales
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (REPARTITION, HASH, [])
+                                                                                                scan reason
+                            local exchange (GATHER, SINGLE, [])
+                                remote exchange (REPLICATE, BROADCAST, [])
                                     final aggregation over ()
                                         local exchange (GATHER, SINGLE, [])
                                             remote exchange (GATHER, SINGLE, [])
                                                 partial aggregation over ()
                                                     scan store_sales
+                        local exchange (GATHER, SINGLE, [])
+                            remote exchange (REPLICATE, BROADCAST, [])
                                 final aggregation over ()
                                     local exchange (GATHER, SINGLE, [])
                                         remote exchange (GATHER, SINGLE, [])
                                             partial aggregation over ()
                                                 scan store_sales
+                    local exchange (GATHER, SINGLE, [])
+                        remote exchange (REPLICATE, BROADCAST, [])
                             final aggregation over ()
                                 local exchange (GATHER, SINGLE, [])
                                     remote exchange (GATHER, SINGLE, [])
                                         partial aggregation over ()
                                             scan store_sales
+                local exchange (GATHER, SINGLE, [])
+                    remote exchange (REPLICATE, BROADCAST, [])
                         final aggregation over ()
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (GATHER, SINGLE, [])
                                     partial aggregation over ()
                                         scan store_sales
+            local exchange (GATHER, SINGLE, [])
+                remote exchange (REPLICATE, BROADCAST, [])
                     final aggregation over ()
                         local exchange (GATHER, SINGLE, [])
                             remote exchange (GATHER, SINGLE, [])
                                 partial aggregation over ()
                                     scan store_sales
+        local exchange (GATHER, SINGLE, [])
+            remote exchange (REPLICATE, BROADCAST, [])
                 final aggregation over ()
                     local exchange (GATHER, SINGLE, [])
                         remote exchange (GATHER, SINGLE, [])
                             partial aggregation over ()
                                 scan store_sales
+    local exchange (GATHER, SINGLE, [])
+        remote exchange (REPLICATE, BROADCAST, [])
             final aggregation over ()
                 local exchange (GATHER, SINGLE, [])
                     remote exchange (GATHER, SINGLE, [])
                         partial aggregation over ()
                             scan store_sales
-        final aggregation over ()
-            local exchange (GATHER, SINGLE, [])
-                remote exchange (GATHER, SINGLE, [])
-                    partial aggregation over ()
-                        scan store_sales
-    final aggregation over ()
-        local exchange (GATHER, SINGLE, [])
-            remote exchange (GATHER, SINGLE, [])
-                partial aggregation over ()
-                    scan store_sales
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/unpartitioned/q54.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/unpartitioned/q54.plan.txt
index 6b99ed23c254..623a7d64ea2c 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/unpartitioned/q54.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/hive/unpartitioned/q54.plan.txt
@@ -8,8 +8,8 @@ local exchange (GATHER, SINGLE, [])
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
                                     partial aggregation over (ss_customer_sk)
-                                        cross join:
-                                            cross join:
+                                        join (LEFT, REPLICATED):
+                                            join (LEFT, REPLICATED):
                                                 join (INNER, REPLICATED):
                                                     join (INNER, REPLICATED):
                                                         dynamic filter (["ss_customer_sk", "ss_sold_date_sk"])
@@ -48,8 +48,7 @@ local exchange (GATHER, SINGLE, [])
                                                                                                             scan date_dim
                                                     local exchange (GATHER, SINGLE, [])
                                                         remote exchange (REPLICATE, BROADCAST, [])
-                                                            dynamic filter (["d_month_seq_24"])
-                                                                scan date_dim
+                                                            scan date_dim
                                                 local exchange (GATHER, SINGLE, [])
                                                     remote exchange (REPLICATE, BROADCAST, [])
                                                         local exchange (GATHER, SINGLE, [])
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/partitioned/q09.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/partitioned/q09.plan.txt
index 4466a2d88feb..cbcd78f04da6 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/partitioned/q09.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/partitioned/q09.plan.txt
@@ -1,93 +1,122 @@
-cross join:
-    cross join:
-        cross join:
-            cross join:
-                cross join:
-                    cross join:
-                        cross join:
-                            cross join:
-                                cross join:
-                                    cross join:
-                                        cross join:
-                                            cross join:
-                                                cross join:
-                                                    cross join:
-                                                        cross join:
-                                                            final aggregation over ()
-                                                                local exchange (GATHER, SINGLE, [])
-                                                                    remote exchange (GATHER, SINGLE, [])
-                                                                        partial aggregation over ()
-                                                                            scan store_sales
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    scan reason
-                                                        final aggregation over ()
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    partial aggregation over ()
-                                                                        scan store_sales
+join (LEFT, REPLICATED):
+    join (LEFT, REPLICATED):
+        join (LEFT, REPLICATED):
+            join (LEFT, REPLICATED):
+                join (LEFT, REPLICATED):
+                    join (LEFT, REPLICATED):
+                        join (LEFT, REPLICATED):
+                            join (RIGHT, PARTITIONED):
+                                remote exchange (REPARTITION, HASH, [])
+                                    final aggregation over ()
+                                        local exchange (GATHER, SINGLE, [])
+                                            remote exchange (GATHER, SINGLE, [])
+                                                partial aggregation over ()
+                                                    scan store_sales
+                                local exchange (GATHER, SINGLE, [])
+                                    join (RIGHT, PARTITIONED):
+                                        remote exchange (REPARTITION, HASH, [])
+                                            final aggregation over ()
+                                                local exchange (GATHER, SINGLE, [])
+                                                    remote exchange (GATHER, SINGLE, [])
+                                                        partial aggregation over ()
+                                                            scan store_sales
+                                        local exchange (GATHER, SINGLE, [])
+                                            join (RIGHT, PARTITIONED):
+                                                remote exchange (REPARTITION, HASH, [])
                                                     final aggregation over ()
                                                         local exchange (GATHER, SINGLE, [])
                                                             remote exchange (GATHER, SINGLE, [])
                                                                 partial aggregation over ()
                                                                     scan store_sales
-                                                final aggregation over ()
-                                                    local exchange (GATHER, SINGLE, [])
-                                                        remote exchange (GATHER, SINGLE, [])
-                                                            partial aggregation over ()
-                                                                scan store_sales
-                                            final aggregation over ()
                                                 local exchange (GATHER, SINGLE, [])
-                                                    remote exchange (GATHER, SINGLE, [])
-                                                        partial aggregation over ()
-                                                            scan store_sales
-                                        final aggregation over ()
-                                            local exchange (GATHER, SINGLE, [])
-                                                remote exchange (GATHER, SINGLE, [])
-                                                    partial aggregation over ()
-                                                        scan store_sales
+                                                    join (RIGHT, PARTITIONED):
+                                                        remote exchange (REPARTITION, HASH, [])
+                                                            final aggregation over ()
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                        partial aggregation over ()
+                                                                            scan store_sales
+                                                        local exchange (GATHER, SINGLE, [])
+                                                            join (RIGHT, PARTITIONED):
+                                                                remote exchange (REPARTITION, HASH, [])
+                                                                    final aggregation over ()
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                partial aggregation over ()
+                                                                                    scan store_sales
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    join (RIGHT, PARTITIONED):
+                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                            final aggregation over ()
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                        partial aggregation over ()
+                                                                                            scan store_sales
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            join (RIGHT, PARTITIONED):
+                                                                                remote exchange (REPARTITION, HASH, [])
+                                                                                    final aggregation over ()
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                                partial aggregation over ()
+                                                                                                    scan store_sales
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    join (RIGHT, PARTITIONED):
+                                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                                            final aggregation over ()
+                                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                                        partial aggregation over ()
+                                                                                                            scan store_sales
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (REPARTITION, HASH, [])
+                                                                                                scan reason
+                            local exchange (GATHER, SINGLE, [])
+                                remote exchange (REPLICATE, BROADCAST, [])
                                     final aggregation over ()
                                         local exchange (GATHER, SINGLE, [])
                                             remote exchange (GATHER, SINGLE, [])
                                                 partial aggregation over ()
                                                     scan store_sales
+                        local exchange (GATHER, SINGLE, [])
+                            remote exchange (REPLICATE, BROADCAST, [])
                                 final aggregation over ()
                                     local exchange (GATHER, SINGLE, [])
                                         remote exchange (GATHER, SINGLE, [])
                                             partial aggregation over ()
                                                 scan store_sales
+                    local exchange (GATHER, SINGLE, [])
+                        remote exchange (REPLICATE, BROADCAST, [])
                             final aggregation over ()
                                 local exchange (GATHER, SINGLE, [])
                                     remote exchange (GATHER, SINGLE, [])
                                         partial aggregation over ()
                                             scan store_sales
+                local exchange (GATHER, SINGLE, [])
+                    remote exchange (REPLICATE, BROADCAST, [])
                         final aggregation over ()
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (GATHER, SINGLE, [])
                                     partial aggregation over ()
                                         scan store_sales
+            local exchange (GATHER, SINGLE, [])
+                remote exchange (REPLICATE, BROADCAST, [])
                     final aggregation over ()
                         local exchange (GATHER, SINGLE, [])
                             remote exchange (GATHER, SINGLE, [])
                                 partial aggregation over ()
                                     scan store_sales
+        local exchange (GATHER, SINGLE, [])
+            remote exchange (REPLICATE, BROADCAST, [])
                 final aggregation over ()
                     local exchange (GATHER, SINGLE, [])
                         remote exchange (GATHER, SINGLE, [])
                             partial aggregation over ()
                                 scan store_sales
+    local exchange (GATHER, SINGLE, [])
+        remote exchange (REPLICATE, BROADCAST, [])
             final aggregation over ()
                 local exchange (GATHER, SINGLE, [])
                     remote exchange (GATHER, SINGLE, [])
                         partial aggregation over ()
                             scan store_sales
-        final aggregation over ()
-            local exchange (GATHER, SINGLE, [])
-                remote exchange (GATHER, SINGLE, [])
-                    partial aggregation over ()
-                        scan store_sales
-    final aggregation over ()
-        local exchange (GATHER, SINGLE, [])
-            remote exchange (GATHER, SINGLE, [])
-                partial aggregation over ()
-                    scan store_sales
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/partitioned/q54.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/partitioned/q54.plan.txt
index 5806dc346635..851fa255cf39 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/partitioned/q54.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/partitioned/q54.plan.txt
@@ -8,8 +8,8 @@ local exchange (GATHER, SINGLE, [])
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
                                     partial aggregation over (ss_customer_sk)
-                                        cross join:
-                                            cross join:
+                                        join (LEFT, REPLICATED):
+                                            join (LEFT, REPLICATED):
                                                 join (INNER, REPLICATED):
                                                     join (INNER, REPLICATED):
                                                         dynamic filter (["ss_customer_sk", "ss_sold_date_sk"])
@@ -48,8 +48,7 @@ local exchange (GATHER, SINGLE, [])
                                                                                                             scan date_dim
                                                     local exchange (GATHER, SINGLE, [])
                                                         remote exchange (REPLICATE, BROADCAST, [])
-                                                            dynamic filter (["d_month_seq_17"])
-                                                                scan date_dim
+                                                            scan date_dim
                                                 local exchange (GATHER, SINGLE, [])
                                                     remote exchange (REPLICATE, BROADCAST, [])
                                                         local exchange (GATHER, SINGLE, [])
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/unpartitioned/q09.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/unpartitioned/q09.plan.txt
index 4466a2d88feb..cbcd78f04da6 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/unpartitioned/q09.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/unpartitioned/q09.plan.txt
@@ -1,93 +1,122 @@
-cross join:
-    cross join:
-        cross join:
-            cross join:
-                cross join:
-                    cross join:
-                        cross join:
-                            cross join:
-                                cross join:
-                                    cross join:
-                                        cross join:
-                                            cross join:
-                                                cross join:
-                                                    cross join:
-                                                        cross join:
-                                                            final aggregation over ()
-                                                                local exchange (GATHER, SINGLE, [])
-                                                                    remote exchange (GATHER, SINGLE, [])
-                                                                        partial aggregation over ()
-                                                                            scan store_sales
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    scan reason
-                                                        final aggregation over ()
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    partial aggregation over ()
-                                                                        scan store_sales
+join (LEFT, REPLICATED):
+    join (LEFT, REPLICATED):
+        join (LEFT, REPLICATED):
+            join (LEFT, REPLICATED):
+                join (LEFT, REPLICATED):
+                    join (LEFT, REPLICATED):
+                        join (LEFT, REPLICATED):
+                            join (RIGHT, PARTITIONED):
+                                remote exchange (REPARTITION, HASH, [])
+                                    final aggregation over ()
+                                        local exchange (GATHER, SINGLE, [])
+                                            remote exchange (GATHER, SINGLE, [])
+                                                partial aggregation over ()
+                                                    scan store_sales
+                                local exchange (GATHER, SINGLE, [])
+                                    join (RIGHT, PARTITIONED):
+                                        remote exchange (REPARTITION, HASH, [])
+                                            final aggregation over ()
+                                                local exchange (GATHER, SINGLE, [])
+                                                    remote exchange (GATHER, SINGLE, [])
+                                                        partial aggregation over ()
+                                                            scan store_sales
+                                        local exchange (GATHER, SINGLE, [])
+                                            join (RIGHT, PARTITIONED):
+                                                remote exchange (REPARTITION, HASH, [])
                                                     final aggregation over ()
                                                         local exchange (GATHER, SINGLE, [])
                                                             remote exchange (GATHER, SINGLE, [])
                                                                 partial aggregation over ()
                                                                     scan store_sales
-                                                final aggregation over ()
-                                                    local exchange (GATHER, SINGLE, [])
-                                                        remote exchange (GATHER, SINGLE, [])
-                                                            partial aggregation over ()
-                                                                scan store_sales
-                                            final aggregation over ()
                                                 local exchange (GATHER, SINGLE, [])
-                                                    remote exchange (GATHER, SINGLE, [])
-                                                        partial aggregation over ()
-                                                            scan store_sales
-                                        final aggregation over ()
-                                            local exchange (GATHER, SINGLE, [])
-                                                remote exchange (GATHER, SINGLE, [])
-                                                    partial aggregation over ()
-                                                        scan store_sales
+                                                    join (RIGHT, PARTITIONED):
+                                                        remote exchange (REPARTITION, HASH, [])
+                                                            final aggregation over ()
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                        partial aggregation over ()
+                                                                            scan store_sales
+                                                        local exchange (GATHER, SINGLE, [])
+                                                            join (RIGHT, PARTITIONED):
+                                                                remote exchange (REPARTITION, HASH, [])
+                                                                    final aggregation over ()
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                partial aggregation over ()
+                                                                                    scan store_sales
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    join (RIGHT, PARTITIONED):
+                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                            final aggregation over ()
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                        partial aggregation over ()
+                                                                                            scan store_sales
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            join (RIGHT, PARTITIONED):
+                                                                                remote exchange (REPARTITION, HASH, [])
+                                                                                    final aggregation over ()
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                                partial aggregation over ()
+                                                                                                    scan store_sales
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    join (RIGHT, PARTITIONED):
+                                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                                            final aggregation over ()
+                                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                                        partial aggregation over ()
+                                                                                                            scan store_sales
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (REPARTITION, HASH, [])
+                                                                                                scan reason
+                            local exchange (GATHER, SINGLE, [])
+                                remote exchange (REPLICATE, BROADCAST, [])
                                     final aggregation over ()
                                         local exchange (GATHER, SINGLE, [])
                                             remote exchange (GATHER, SINGLE, [])
                                                 partial aggregation over ()
                                                     scan store_sales
+                        local exchange (GATHER, SINGLE, [])
+                            remote exchange (REPLICATE, BROADCAST, [])
                                 final aggregation over ()
                                     local exchange (GATHER, SINGLE, [])
                                         remote exchange (GATHER, SINGLE, [])
                                             partial aggregation over ()
                                                 scan store_sales
+                    local exchange (GATHER, SINGLE, [])
+                        remote exchange (REPLICATE, BROADCAST, [])
                             final aggregation over ()
                                 local exchange (GATHER, SINGLE, [])
                                     remote exchange (GATHER, SINGLE, [])
                                         partial aggregation over ()
                                             scan store_sales
+                local exchange (GATHER, SINGLE, [])
+                    remote exchange (REPLICATE, BROADCAST, [])
                         final aggregation over ()
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (GATHER, SINGLE, [])
                                     partial aggregation over ()
                                         scan store_sales
+            local exchange (GATHER, SINGLE, [])
+                remote exchange (REPLICATE, BROADCAST, [])
                     final aggregation over ()
                         local exchange (GATHER, SINGLE, [])
                             remote exchange (GATHER, SINGLE, [])
                                 partial aggregation over ()
                                     scan store_sales
+        local exchange (GATHER, SINGLE, [])
+            remote exchange (REPLICATE, BROADCAST, [])
                 final aggregation over ()
                     local exchange (GATHER, SINGLE, [])
                         remote exchange (GATHER, SINGLE, [])
                             partial aggregation over ()
                                 scan store_sales
+    local exchange (GATHER, SINGLE, [])
+        remote exchange (REPLICATE, BROADCAST, [])
             final aggregation over ()
                 local exchange (GATHER, SINGLE, [])
                     remote exchange (GATHER, SINGLE, [])
                         partial aggregation over ()
                             scan store_sales
-        final aggregation over ()
-            local exchange (GATHER, SINGLE, [])
-                remote exchange (GATHER, SINGLE, [])
-                    partial aggregation over ()
-                        scan store_sales
-    final aggregation over ()
-        local exchange (GATHER, SINGLE, [])
-            remote exchange (GATHER, SINGLE, [])
-                partial aggregation over ()
-                    scan store_sales
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/unpartitioned/q54.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/unpartitioned/q54.plan.txt
index 5806dc346635..851fa255cf39 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/unpartitioned/q54.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/orc/unpartitioned/q54.plan.txt
@@ -8,8 +8,8 @@ local exchange (GATHER, SINGLE, [])
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
                                     partial aggregation over (ss_customer_sk)
-                                        cross join:
-                                            cross join:
+                                        join (LEFT, REPLICATED):
+                                            join (LEFT, REPLICATED):
                                                 join (INNER, REPLICATED):
                                                     join (INNER, REPLICATED):
                                                         dynamic filter (["ss_customer_sk", "ss_sold_date_sk"])
@@ -48,8 +48,7 @@ local exchange (GATHER, SINGLE, [])
                                                                                                             scan date_dim
                                                     local exchange (GATHER, SINGLE, [])
                                                         remote exchange (REPLICATE, BROADCAST, [])
-                                                            dynamic filter (["d_month_seq_17"])
-                                                                scan date_dim
+                                                            scan date_dim
                                                 local exchange (GATHER, SINGLE, [])
                                                     remote exchange (REPLICATE, BROADCAST, [])
                                                         local exchange (GATHER, SINGLE, [])
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/partitioned/q09.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/partitioned/q09.plan.txt
index 4466a2d88feb..cbcd78f04da6 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/partitioned/q09.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/partitioned/q09.plan.txt
@@ -1,93 +1,122 @@
-cross join:
-    cross join:
-        cross join:
-            cross join:
-                cross join:
-                    cross join:
-                        cross join:
-                            cross join:
-                                cross join:
-                                    cross join:
-                                        cross join:
-                                            cross join:
-                                                cross join:
-                                                    cross join:
-                                                        cross join:
-                                                            final aggregation over ()
-                                                                local exchange (GATHER, SINGLE, [])
-                                                                    remote exchange (GATHER, SINGLE, [])
-                                                                        partial aggregation over ()
-                                                                            scan store_sales
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    scan reason
-                                                        final aggregation over ()
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    partial aggregation over ()
-                                                                        scan store_sales
+join (LEFT, REPLICATED):
+    join (LEFT, REPLICATED):
+        join (LEFT, REPLICATED):
+            join (LEFT, REPLICATED):
+                join (LEFT, REPLICATED):
+                    join (LEFT, REPLICATED):
+                        join (LEFT, REPLICATED):
+                            join (RIGHT, PARTITIONED):
+                                remote exchange (REPARTITION, HASH, [])
+                                    final aggregation over ()
+                                        local exchange (GATHER, SINGLE, [])
+                                            remote exchange (GATHER, SINGLE, [])
+                                                partial aggregation over ()
+                                                    scan store_sales
+                                local exchange (GATHER, SINGLE, [])
+                                    join (RIGHT, PARTITIONED):
+                                        remote exchange (REPARTITION, HASH, [])
+                                            final aggregation over ()
+                                                local exchange (GATHER, SINGLE, [])
+                                                    remote exchange (GATHER, SINGLE, [])
+                                                        partial aggregation over ()
+                                                            scan store_sales
+                                        local exchange (GATHER, SINGLE, [])
+                                            join (RIGHT, PARTITIONED):
+                                                remote exchange (REPARTITION, HASH, [])
                                                     final aggregation over ()
                                                         local exchange (GATHER, SINGLE, [])
                                                             remote exchange (GATHER, SINGLE, [])
                                                                 partial aggregation over ()
                                                                     scan store_sales
-                                                final aggregation over ()
-                                                    local exchange (GATHER, SINGLE, [])
-                                                        remote exchange (GATHER, SINGLE, [])
-                                                            partial aggregation over ()
-                                                                scan store_sales
-                                            final aggregation over ()
                                                 local exchange (GATHER, SINGLE, [])
-                                                    remote exchange (GATHER, SINGLE, [])
-                                                        partial aggregation over ()
-                                                            scan store_sales
-                                        final aggregation over ()
-                                            local exchange (GATHER, SINGLE, [])
-                                                remote exchange (GATHER, SINGLE, [])
-                                                    partial aggregation over ()
-                                                        scan store_sales
+                                                    join (RIGHT, PARTITIONED):
+                                                        remote exchange (REPARTITION, HASH, [])
+                                                            final aggregation over ()
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                        partial aggregation over ()
+                                                                            scan store_sales
+                                                        local exchange (GATHER, SINGLE, [])
+                                                            join (RIGHT, PARTITIONED):
+                                                                remote exchange (REPARTITION, HASH, [])
+                                                                    final aggregation over ()
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                partial aggregation over ()
+                                                                                    scan store_sales
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    join (RIGHT, PARTITIONED):
+                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                            final aggregation over ()
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                        partial aggregation over ()
+                                                                                            scan store_sales
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            join (RIGHT, PARTITIONED):
+                                                                                remote exchange (REPARTITION, HASH, [])
+                                                                                    final aggregation over ()
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                                partial aggregation over ()
+                                                                                                    scan store_sales
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    join (RIGHT, PARTITIONED):
+                                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                                            final aggregation over ()
+                                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                                        partial aggregation over ()
+                                                                                                            scan store_sales
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (REPARTITION, HASH, [])
+                                                                                                scan reason
+                            local exchange (GATHER, SINGLE, [])
+                                remote exchange (REPLICATE, BROADCAST, [])
                                     final aggregation over ()
                                         local exchange (GATHER, SINGLE, [])
                                             remote exchange (GATHER, SINGLE, [])
                                                 partial aggregation over ()
                                                     scan store_sales
+                        local exchange (GATHER, SINGLE, [])
+                            remote exchange (REPLICATE, BROADCAST, [])
                                 final aggregation over ()
                                     local exchange (GATHER, SINGLE, [])
                                         remote exchange (GATHER, SINGLE, [])
                                             partial aggregation over ()
                                                 scan store_sales
+                    local exchange (GATHER, SINGLE, [])
+                        remote exchange (REPLICATE, BROADCAST, [])
                             final aggregation over ()
                                 local exchange (GATHER, SINGLE, [])
                                     remote exchange (GATHER, SINGLE, [])
                                         partial aggregation over ()
                                             scan store_sales
+                local exchange (GATHER, SINGLE, [])
+                    remote exchange (REPLICATE, BROADCAST, [])
                         final aggregation over ()
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (GATHER, SINGLE, [])
                                     partial aggregation over ()
                                         scan store_sales
+            local exchange (GATHER, SINGLE, [])
+                remote exchange (REPLICATE, BROADCAST, [])
                     final aggregation over ()
                         local exchange (GATHER, SINGLE, [])
                             remote exchange (GATHER, SINGLE, [])
                                 partial aggregation over ()
                                     scan store_sales
+        local exchange (GATHER, SINGLE, [])
+            remote exchange (REPLICATE, BROADCAST, [])
                 final aggregation over ()
                     local exchange (GATHER, SINGLE, [])
                         remote exchange (GATHER, SINGLE, [])
                             partial aggregation over ()
                                 scan store_sales
+    local exchange (GATHER, SINGLE, [])
+        remote exchange (REPLICATE, BROADCAST, [])
             final aggregation over ()
                 local exchange (GATHER, SINGLE, [])
                     remote exchange (GATHER, SINGLE, [])
                         partial aggregation over ()
                             scan store_sales
-        final aggregation over ()
-            local exchange (GATHER, SINGLE, [])
-                remote exchange (GATHER, SINGLE, [])
-                    partial aggregation over ()
-                        scan store_sales
-    final aggregation over ()
-        local exchange (GATHER, SINGLE, [])
-            remote exchange (GATHER, SINGLE, [])
-                partial aggregation over ()
-                    scan store_sales
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/partitioned/q54.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/partitioned/q54.plan.txt
index 5806dc346635..851fa255cf39 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/partitioned/q54.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/partitioned/q54.plan.txt
@@ -8,8 +8,8 @@ local exchange (GATHER, SINGLE, [])
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
                                     partial aggregation over (ss_customer_sk)
-                                        cross join:
-                                            cross join:
+                                        join (LEFT, REPLICATED):
+                                            join (LEFT, REPLICATED):
                                                 join (INNER, REPLICATED):
                                                     join (INNER, REPLICATED):
                                                         dynamic filter (["ss_customer_sk", "ss_sold_date_sk"])
@@ -48,8 +48,7 @@ local exchange (GATHER, SINGLE, [])
                                                                                                             scan date_dim
                                                     local exchange (GATHER, SINGLE, [])
                                                         remote exchange (REPLICATE, BROADCAST, [])
-                                                            dynamic filter (["d_month_seq_17"])
-                                                                scan date_dim
+                                                            scan date_dim
                                                 local exchange (GATHER, SINGLE, [])
                                                     remote exchange (REPLICATE, BROADCAST, [])
                                                         local exchange (GATHER, SINGLE, [])
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/unpartitioned/q09.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/unpartitioned/q09.plan.txt
index 5660bb85f04e..cbcd78f04da6 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/unpartitioned/q09.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/unpartitioned/q09.plan.txt
@@ -1,93 +1,122 @@
-cross join:
-    cross join:
-        cross join:
-            cross join:
-                cross join:
-                    cross join:
-                        cross join:
-                            cross join:
-                                cross join:
-                                    cross join:
-                                        cross join:
-                                            cross join:
-                                                cross join:
-                                                    cross join:
-                                                        final aggregation over ()
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    partial aggregation over ()
-                                                                        scan store_sales
-                                                        cross join:
-                                                            final aggregation over ()
-                                                                local exchange (GATHER, SINGLE, [])
-                                                                    remote exchange (GATHER, SINGLE, [])
-                                                                        partial aggregation over ()
-                                                                            scan store_sales
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    scan reason
+join (LEFT, REPLICATED):
+    join (LEFT, REPLICATED):
+        join (LEFT, REPLICATED):
+            join (LEFT, REPLICATED):
+                join (LEFT, REPLICATED):
+                    join (LEFT, REPLICATED):
+                        join (LEFT, REPLICATED):
+                            join (RIGHT, PARTITIONED):
+                                remote exchange (REPARTITION, HASH, [])
+                                    final aggregation over ()
+                                        local exchange (GATHER, SINGLE, [])
+                                            remote exchange (GATHER, SINGLE, [])
+                                                partial aggregation over ()
+                                                    scan store_sales
+                                local exchange (GATHER, SINGLE, [])
+                                    join (RIGHT, PARTITIONED):
+                                        remote exchange (REPARTITION, HASH, [])
+                                            final aggregation over ()
+                                                local exchange (GATHER, SINGLE, [])
+                                                    remote exchange (GATHER, SINGLE, [])
+                                                        partial aggregation over ()
+                                                            scan store_sales
+                                        local exchange (GATHER, SINGLE, [])
+                                            join (RIGHT, PARTITIONED):
+                                                remote exchange (REPARTITION, HASH, [])
                                                     final aggregation over ()
                                                         local exchange (GATHER, SINGLE, [])
                                                             remote exchange (GATHER, SINGLE, [])
                                                                 partial aggregation over ()
                                                                     scan store_sales
-                                                final aggregation over ()
-                                                    local exchange (GATHER, SINGLE, [])
-                                                        remote exchange (GATHER, SINGLE, [])
-                                                            partial aggregation over ()
-                                                                scan store_sales
-                                            final aggregation over ()
                                                 local exchange (GATHER, SINGLE, [])
-                                                    remote exchange (GATHER, SINGLE, [])
-                                                        partial aggregation over ()
-                                                            scan store_sales
-                                        final aggregation over ()
-                                            local exchange (GATHER, SINGLE, [])
-                                                remote exchange (GATHER, SINGLE, [])
-                                                    partial aggregation over ()
-                                                        scan store_sales
+                                                    join (RIGHT, PARTITIONED):
+                                                        remote exchange (REPARTITION, HASH, [])
+                                                            final aggregation over ()
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                        partial aggregation over ()
+                                                                            scan store_sales
+                                                        local exchange (GATHER, SINGLE, [])
+                                                            join (RIGHT, PARTITIONED):
+                                                                remote exchange (REPARTITION, HASH, [])
+                                                                    final aggregation over ()
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                partial aggregation over ()
+                                                                                    scan store_sales
+                                                                local exchange (GATHER, SINGLE, [])
+                                                                    join (RIGHT, PARTITIONED):
+                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                            final aggregation over ()
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                        partial aggregation over ()
+                                                                                            scan store_sales
+                                                                        local exchange (GATHER, SINGLE, [])
+                                                                            join (RIGHT, PARTITIONED):
+                                                                                remote exchange (REPARTITION, HASH, [])
+                                                                                    final aggregation over ()
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (GATHER, SINGLE, [])
+                                                                                                partial aggregation over ()
+                                                                                                    scan store_sales
+                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                    join (RIGHT, PARTITIONED):
+                                                                                        remote exchange (REPARTITION, HASH, [])
+                                                                                            final aggregation over ()
+                                                                                                local exchange (GATHER, SINGLE, [])
+                                                                                                    remote exchange (GATHER, SINGLE, [])
+                                                                                                        partial aggregation over ()
+                                                                                                            scan store_sales
+                                                                                        local exchange (GATHER, SINGLE, [])
+                                                                                            remote exchange (REPARTITION, HASH, [])
+                                                                                                scan reason
+                            local exchange (GATHER, SINGLE, [])
+                                remote exchange (REPLICATE, BROADCAST, [])
                                     final aggregation over ()
                                         local exchange (GATHER, SINGLE, [])
                                             remote exchange (GATHER, SINGLE, [])
                                                 partial aggregation over ()
                                                     scan store_sales
+                        local exchange (GATHER, SINGLE, [])
+                            remote exchange (REPLICATE, BROADCAST, [])
                                 final aggregation over ()
                                     local exchange (GATHER, SINGLE, [])
                                         remote exchange (GATHER, SINGLE, [])
                                             partial aggregation over ()
                                                 scan store_sales
+                    local exchange (GATHER, SINGLE, [])
+                        remote exchange (REPLICATE, BROADCAST, [])
                             final aggregation over ()
                                 local exchange (GATHER, SINGLE, [])
                                     remote exchange (GATHER, SINGLE, [])
                                         partial aggregation over ()
                                             scan store_sales
+                local exchange (GATHER, SINGLE, [])
+                    remote exchange (REPLICATE, BROADCAST, [])
                         final aggregation over ()
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (GATHER, SINGLE, [])
                                     partial aggregation over ()
                                         scan store_sales
+            local exchange (GATHER, SINGLE, [])
+                remote exchange (REPLICATE, BROADCAST, [])
                     final aggregation over ()
                         local exchange (GATHER, SINGLE, [])
                             remote exchange (GATHER, SINGLE, [])
                                 partial aggregation over ()
                                     scan store_sales
+        local exchange (GATHER, SINGLE, [])
+            remote exchange (REPLICATE, BROADCAST, [])
                 final aggregation over ()
                     local exchange (GATHER, SINGLE, [])
                         remote exchange (GATHER, SINGLE, [])
                             partial aggregation over ()
                                 scan store_sales
+    local exchange (GATHER, SINGLE, [])
+        remote exchange (REPLICATE, BROADCAST, [])
             final aggregation over ()
                 local exchange (GATHER, SINGLE, [])
                     remote exchange (GATHER, SINGLE, [])
                         partial aggregation over ()
                             scan store_sales
-        final aggregation over ()
-            local exchange (GATHER, SINGLE, [])
-                remote exchange (GATHER, SINGLE, [])
-                    partial aggregation over ()
-                        scan store_sales
-    final aggregation over ()
-        local exchange (GATHER, SINGLE, [])
-            remote exchange (GATHER, SINGLE, [])
-                partial aggregation over ()
-                    scan store_sales
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/unpartitioned/q54.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/unpartitioned/q54.plan.txt
index 5806dc346635..851fa255cf39 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/unpartitioned/q54.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg/parquet/unpartitioned/q54.plan.txt
@@ -8,8 +8,8 @@ local exchange (GATHER, SINGLE, [])
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (REPARTITION, HASH, ["ss_customer_sk"])
                                     partial aggregation over (ss_customer_sk)
-                                        cross join:
-                                            cross join:
+                                        join (LEFT, REPLICATED):
+                                            join (LEFT, REPLICATED):
                                                 join (INNER, REPLICATED):
                                                     join (INNER, REPLICATED):
                                                         dynamic filter (["ss_customer_sk", "ss_sold_date_sk"])
@@ -48,8 +48,7 @@ local exchange (GATHER, SINGLE, [])
                                                                                                             scan date_dim
                                                     local exchange (GATHER, SINGLE, [])
                                                         remote exchange (REPLICATE, BROADCAST, [])
-                                                            dynamic filter (["d_month_seq_17"])
-                                                                scan date_dim
+                                                            scan date_dim
                                                 local exchange (GATHER, SINGLE, [])
                                                     remote exchange (REPLICATE, BROADCAST, [])
                                                         local exchange (GATHER, SINGLE, [])
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg_small_files/parquet/unpartitioned/q09.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg_small_files/parquet/unpartitioned/q09.plan.txt
index 4466a2d88feb..510c370f94d9 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg_small_files/parquet/unpartitioned/q09.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg_small_files/parquet/unpartitioned/q09.plan.txt
@@ -1,93 +1,122 @@
-cross join:
-    cross join:
-        cross join:
-            cross join:
-                cross join:
-                    cross join:
-                        cross join:
-                            cross join:
-                                cross join:
-                                    cross join:
-                                        cross join:
-                                            cross join:
-                                                cross join:
-                                                    cross join:
-                                                        cross join:
+join (LEFT, REPLICATED):
+    join (LEFT, REPLICATED):
+        join (LEFT, REPLICATED):
+            join (LEFT, REPLICATED):
+                join (LEFT, REPLICATED):
+                    join (LEFT, REPLICATED):
+                        join (LEFT, REPLICATED):
+                            join (LEFT, REPLICATED):
+                                join (LEFT, REPLICATED):
+                                    join (LEFT, REPLICATED):
+                                        join (LEFT, REPLICATED):
+                                            join (LEFT, REPLICATED):
+                                                join (LEFT, REPLICATED):
+                                                    join (LEFT, REPLICATED):
+                                                        join (RIGHT, PARTITIONED):
+                                                            remote exchange (REPARTITION, HASH, [])
+                                                                final aggregation over ()
+                                                                    local exchange (GATHER, SINGLE, [])
+                                                                        remote exchange (GATHER, SINGLE, [])
+                                                                            partial aggregation over ()
+                                                                                scan store_sales
+                                                            local exchange (GATHER, SINGLE, [])
+                                                                remote exchange (REPARTITION, HASH, [])
+                                                                    scan reason
+                                                        local exchange (GATHER, SINGLE, [])
+                                                            remote exchange (REPLICATE, BROADCAST, [])
+                                                                final aggregation over ()
+                                                                    local exchange (GATHER, SINGLE, [])
+                                                                        remote exchange (GATHER, SINGLE, [])
+                                                                            partial aggregation over ()
+                                                                                scan store_sales
+                                                    local exchange (GATHER, SINGLE, [])
+                                                        remote exchange (REPLICATE, BROADCAST, [])
                                                             final aggregation over ()
                                                                 local exchange (GATHER, SINGLE, [])
                                                                     remote exchange (GATHER, SINGLE, [])
                                                                         partial aggregation over ()
                                                                             scan store_sales
-                                                            local exchange (GATHER, SINGLE, [])
-                                                                remote exchange (GATHER, SINGLE, [])
-                                                                    scan reason
+                                                local exchange (GATHER, SINGLE, [])
+                                                    remote exchange (REPLICATE, BROADCAST, [])
                                                         final aggregation over ()
                                                             local exchange (GATHER, SINGLE, [])
                                                                 remote exchange (GATHER, SINGLE, [])
                                                                     partial aggregation over ()
                                                                         scan store_sales
+                                            local exchange (GATHER, SINGLE, [])
+                                                remote exchange (REPLICATE, BROADCAST, [])
                                                     final aggregation over ()
                                                         local exchange (GATHER, SINGLE, [])
                                                             remote exchange (GATHER, SINGLE, [])
                                                                 partial aggregation over ()
                                                                     scan store_sales
+                                        local exchange (GATHER, SINGLE, [])
+                                            remote exchange (REPLICATE, BROADCAST, [])
                                                 final aggregation over ()
                                                     local exchange (GATHER, SINGLE, [])
                                                         remote exchange (GATHER, SINGLE, [])
                                                             partial aggregation over ()
                                                                 scan store_sales
+                                    local exchange (GATHER, SINGLE, [])
+                                        remote exchange (REPLICATE, BROADCAST, [])
                                             final aggregation over ()
                                                 local exchange (GATHER, SINGLE, [])
                                                     remote exchange (GATHER, SINGLE, [])
                                                         partial aggregation over ()
                                                             scan store_sales
+                                local exchange (GATHER, SINGLE, [])
+                                    remote exchange (REPLICATE, BROADCAST, [])
                                         final aggregation over ()
                                             local exchange (GATHER, SINGLE, [])
                                                 remote exchange (GATHER, SINGLE, [])
                                                     partial aggregation over ()
                                                         scan store_sales
+                            local exchange (GATHER, SINGLE, [])
+                                remote exchange (REPLICATE, BROADCAST, [])
                                     final aggregation over ()
                                         local exchange (GATHER, SINGLE, [])
                                             remote exchange (GATHER, SINGLE, [])
                                                 partial aggregation over ()
                                                     scan store_sales
+                        local exchange (GATHER, SINGLE, [])
+                            remote exchange (REPLICATE, BROADCAST, [])
                                 final aggregation over ()
                                     local exchange (GATHER, SINGLE, [])
                                         remote exchange (GATHER, SINGLE, [])
                                             partial aggregation over ()
                                                 scan store_sales
+                    local exchange (GATHER, SINGLE, [])
+                        remote exchange (REPLICATE, BROADCAST, [])
                             final aggregation over ()
                                 local exchange (GATHER, SINGLE, [])
                                     remote exchange (GATHER, SINGLE, [])
                                         partial aggregation over ()
                                             scan store_sales
+                local exchange (GATHER, SINGLE, [])
+                    remote exchange (REPLICATE, BROADCAST, [])
                         final aggregation over ()
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (GATHER, SINGLE, [])
                                     partial aggregation over ()
                                         scan store_sales
+            local exchange (GATHER, SINGLE, [])
+                remote exchange (REPLICATE, BROADCAST, [])
                     final aggregation over ()
                         local exchange (GATHER, SINGLE, [])
                             remote exchange (GATHER, SINGLE, [])
                                 partial aggregation over ()
                                     scan store_sales
+        local exchange (GATHER, SINGLE, [])
+            remote exchange (REPLICATE, BROADCAST, [])
                 final aggregation over ()
                     local exchange (GATHER, SINGLE, [])
                         remote exchange (GATHER, SINGLE, [])
                             partial aggregation over ()
                                 scan store_sales
+    local exchange (GATHER, SINGLE, [])
+        remote exchange (REPLICATE, BROADCAST, [])
             final aggregation over ()
                 local exchange (GATHER, SINGLE, [])
                     remote exchange (GATHER, SINGLE, [])
                         partial aggregation over ()
                             scan store_sales
-        final aggregation over ()
-            local exchange (GATHER, SINGLE, [])
-                remote exchange (GATHER, SINGLE, [])
-                    partial aggregation over ()
-                        scan store_sales
-    final aggregation over ()
-        local exchange (GATHER, SINGLE, [])
-            remote exchange (GATHER, SINGLE, [])
-                partial aggregation over ()
-                    scan store_sales
diff --git a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg_small_files/parquet/unpartitioned/q54.plan.txt b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg_small_files/parquet/unpartitioned/q54.plan.txt
index f578993e3014..b49678e02521 100644
--- a/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg_small_files/parquet/unpartitioned/q54.plan.txt
+++ b/testing/trino-tests/src/test/resources/sql/presto/tpcds/iceberg_small_files/parquet/unpartitioned/q54.plan.txt
@@ -8,8 +8,8 @@ local exchange (GATHER, SINGLE, [])
                             local exchange (GATHER, SINGLE, [])
                                 remote exchange (REPARTITION, HASH, ["customer_sk"])
                                     partial aggregation over (customer_sk)
-                                        cross join:
-                                            cross join:
+                                        join (LEFT, REPLICATED):
+                                            join (LEFT, REPLICATED):
                                                 join (INNER, REPLICATED):
                                                     join (INNER, REPLICATED):
                                                         join (INNER, PARTITIONED):
@@ -50,8 +50,7 @@ local exchange (GATHER, SINGLE, [])
                                                                 scan store
                                                     local exchange (GATHER, SINGLE, [])
                                                         remote exchange (REPLICATE, BROADCAST, [])
-                                                            dynamic filter (["d_month_seq_17"])
-                                                                scan date_dim
+                                                            scan date_dim
                                                 local exchange (GATHER, SINGLE, [])
                                                     remote exchange (REPLICATE, BROADCAST, [])
                                                         local exchange (GATHER, SINGLE, [])