You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "Kontinuation (via GitHub)" <gi...@apache.org> on 2025/04/01 12:30:46 UTC
[I] AQE Unable to Rewrite Joins as Broadcast Hash Joins Due to Existing CometBroadcastHashJoin Operator [datafusion-comet]
Kontinuation opened a new issue, #1589:
URL: https://github.com/apache/datafusion-comet/issues/1589
### Describe the bug
AQE could transform SortMergeJoin or ShuffledHashJoin to BroadcastHashJoin dynamically after discovering that one of the Exchange operator only shuffle writes small amount of data. However, this optimization does not always happen when using Comet.
TPC-H Q7 has an equi-join between `supplier` and `lineitem`. Spark could discover that `supplier` is small enough to be broadcasted after running the `Exchange` operator, and dynamically change the sort-merge-join to a broadcast hash join (see `BroadcastHashJoin Inner BuildLeft (15)`):
```
== Physical Plan ==
AdaptiveSparkPlan (99)
+- == Final Plan ==
* Sort (62)
+- AQEShuffleRead (61)
+- ShuffleQueryStage (60), Statistics(sizeInBytes=288.0 B, rowCount=4)
+- Exchange (59)
+- * HashAggregate (58)
+- AQEShuffleRead (57)
+- ShuffleQueryStage (56), Statistics(sizeInBytes=2.8 KiB, rowCount=36)
+- Exchange (55)
+- * HashAggregate (54)
+- * Project (53)
+- * BroadcastHashJoin Inner BuildRight (52)
:- * Project (49)
: +- * BroadcastHashJoin Inner BuildRight (48)
: :- * Project (42)
: : +- * SortMergeJoin Inner (41)
: : :- * Sort (33)
: : : +- AQEShuffleRead (32)
: : : +- ShuffleQueryStage (31), Statistics(sizeInBytes=667.5 MiB, rowCount=1.46E+7)
: : : +- Exchange (30)
: : : +- * Project (29)
: : : +- * SortMergeJoin Inner (28)
: : : :- * Sort (20)
: : : : +- AQEShuffleRead (19)
: : : : +- ShuffleQueryStage (18), Statistics(sizeInBytes=667.5 MiB, rowCount=1.46E+7)
: : : : +- Exchange (17)
: : : : +- * Project (16)
: : : : +- * BroadcastHashJoin Inner BuildLeft (15) <-- Transformed from SortMergeJoin by AQE
: : : : :- BroadcastQueryStage (8), Statistics(sizeInBytes=8.0 MiB, rowCount=8.00E+4)
: : : : : +- BroadcastExchange (7)
: : : : : +- AQEShuffleRead (6)
: : : : : +- ShuffleQueryStage (5), Statistics(sizeInBytes=1874.1 KiB, rowCount=8.00E+4)
: : : : : +- Exchange (4)
: : : : : +- * Filter (3)
: : : : : +- * ColumnarToRow (2)
: : : : : +- Scan parquet (1)
: : : : +- AQEShuffleRead (14)
: : : : +- ShuffleQueryStage (13), Statistics(sizeInBytes=8.1 GiB, rowCount=1.82E+8)
: : : : +- Exchange (12)
: : : : +- * Filter (11)
: : : : +- * ColumnarToRow (10)
: : : : +- Scan parquet (9)
: : : +- * Sort (27)
: : : +- AQEShuffleRead (26)
: : : +- ShuffleQueryStage (25), Statistics(sizeInBytes=3.4 GiB, rowCount=1.50E+8)
: : : +- Exchange (24)
: : : +- * Filter (23)
: : : +- * ColumnarToRow (22)
: : : +- Scan parquet (21)
: : +- * Sort (40)
: : +- AQEShuffleRead (39)
: : +- ShuffleQueryStage (38), Statistics(sizeInBytes=27.5 MiB, rowCount=1.20E+6)
: : +- Exchange (37)
: : +- * Filter (36)
: : +- * ColumnarToRow (35)
: : +- Scan parquet (34)
: +- BroadcastQueryStage (47), Statistics(sizeInBytes=1024.0 KiB, rowCount=2)
: +- BroadcastExchange (46)
: +- * Filter (45)
: +- * ColumnarToRow (44)
: +- Scan parquet (43)
+- BroadcastQueryStage (51), Statistics(sizeInBytes=1024.0 KiB, rowCount=2)
+- ReusedExchange (50)
```
The following query plan is generated for running TPC-H Q7 with Comet enabled, The CometSortMergeJoin was not transformed to CometBroadcastHashJoin:
```
== Physical Plan ==
AdaptiveSparkPlan (95)
+- == Final Plan ==
* CometColumnarToRow (58)
+- CometSort (57)
+- AQEShuffleRead (56)
+- ShuffleQueryStage (55), Statistics(sizeInBytes=288.0 B, rowCount=4)
+- CometColumnarExchange (54)
+- CometHashAggregate (53)
+- AQEShuffleRead (52)
+- ShuffleQueryStage (51), Statistics(sizeInBytes=8.5 KiB, rowCount=72)
+- CometExchange (50)
+- CometHashAggregate (49)
+- CometProject (48)
+- CometBroadcastHashJoin (47)
:- CometProject (44)
: +- CometBroadcastHashJoin (43)
: :- CometProject (38)
: : +- CometSortMergeJoin (37)
: : :- CometSort (30)
: : : +- AQEShuffleRead (29)
: : : +- ShuffleQueryStage (28), Statistics(sizeInBytes=724.0 MiB, rowCount=2.92E+7)
: : : +- CometExchange (27)
: : : +- CometProject (26)
: : : +- CometSortMergeJoin (25)
: : : :- CometSort (18)
: : : : +- AQEShuffleRead (17)
: : : : +- ShuffleQueryStage (16), Statistics(sizeInBytes=724.0 MiB, rowCount=2.92E+7)
: : : : +- CometExchange (15)
: : : : +- CometProject (14)
: : : : +- CometSortMergeJoin (13) <-- Not transformed to CometBroadcastHashJoin
: : : : :- CometSort (6)
: : : : : +- AQEShuffleRead (5)
: : : : : +- ShuffleQueryStage (4), Statistics(sizeInBytes=1281.0 KiB, rowCount=1.60E+5)
: : : : : +- CometExchange (3)
: : : : : +- CometFilter (2)
: : : : : +- CometScan parquet (1)
: : : : +- CometSort (12)
: : : : +- AQEShuffleRead (11)
: : : : +- ShuffleQueryStage (10), Statistics(sizeInBytes=8.9 GiB, rowCount=3.65E+8)
: : : : +- CometExchange (9)
: : : : +- CometFilter (8)
: : : : +- CometScan parquet (7)
: : : +- CometSort (24)
: : : +- AQEShuffleRead (23)
: : : +- ShuffleQueryStage (22), Statistics(sizeInBytes=2.2 GiB, rowCount=3.00E+8)
: : : +- CometExchange (21)
: : : +- CometFilter (20)
: : : +- CometScan parquet (19)
: : +- CometSort (36)
: : +- AQEShuffleRead (35)
: : +- ShuffleQueryStage (34), Statistics(sizeInBytes=18.7 MiB, rowCount=2.40E+6)
: : +- CometExchange (33)
: : +- CometFilter (32)
: : +- CometScan parquet (31)
: +- BroadcastQueryStage (42), Statistics(sizeInBytes=337.0 B, rowCount=2)
: +- CometBroadcastExchange (41)
: +- CometFilter (40)
: +- CometScan parquet (39)
+- BroadcastQueryStage (46), Statistics(sizeInBytes=337.0 B, rowCount=2)
+- ReusedExchange (45)
```
### Steps to reproduce
Run TPC-H Q7 using TPC-H SF=100 dataset. The benchmarking code is in https://github.com/apache/datafusion-benchmarks/tree/main/tpch.
```bash
spark-submit \
--master local[8] \
--conf spark.driver.memory=3g \
--conf spark.memory.offHeap.enabled=true \
--conf spark.memory.offHeap.size=16g \
--conf spark.jars=$COMET_JAR \
--conf spark.driver.extraClassPath=$COMET_JAR \
--conf spark.executor.extraClassPath=$COMET_JAR \
--conf spark.sql.extensions=org.apache.comet.CometSparkSessionExtensions \
--conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \
--conf spark.comet.enabled=true \
--conf spark.comet.exec.shuffle.enabled=true \
--conf spark.comet.exec.shuffle.mode=auto \
--conf spark.comet.exec.shuffle.fallbackToColumnar=true \
--conf spark.comet.exec.shuffle.compression.codec=lz4 \
--conf spark.comet.exec.replaceSortMergeJoin=false \
tpcbench.py \
--benchmark tpch \
--data /path/to/tpch/sf100_parquet \
--queries ../../tpch/queries \
--output tpc-results \
--iterations 3
```
### Expected behavior
The inner most CometSortMergeJoin gets transformed to CometBroadcastHashJoin.
### Additional context
_No response_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] AQE Unable to Rewrite Joins as Broadcast Hash Joins Due to Existing CometBroadcastHashJoin Operator [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on issue #1589:
URL: https://github.com/apache/datafusion-comet/issues/1589#issuecomment-2770064370
It would be good to propose adding `BroadcastJoinLike` to Spark, although it would be a while before we could leverage that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] AQE Unable to Rewrite Joins as Broadcast Hash Joins Due to Existing CometBroadcastHashJoin Operator [datafusion-comet]
Posted by "Kontinuation (via GitHub)" <gi...@apache.org>.
Kontinuation commented on issue #1589:
URL: https://github.com/apache/datafusion-comet/issues/1589#issuecomment-2772520039
Turns out that [CometBroadcastExchangeExec](https://github.com/apache/datafusion-comet/blob/0.7.0/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastExchangeExec.scala) does not implement `outputPartitioning` method like what [BroadcastExchangeExec](https://github.com/apache/spark/blob/v3.5.5/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala#L95) does. This makes `EnsureRequirements` consider that the CometBroadcastExchange node does not satisfy the data distribution requirement of BroadcastHashJoin. I'm working on a fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] AQE Unable to Rewrite Joins as Broadcast Hash Joins Due to Existing CometBroadcastHashJoin Operator [datafusion-comet]
Posted by "Kontinuation (via GitHub)" <gi...@apache.org>.
Kontinuation commented on issue #1589:
URL: https://github.com/apache/datafusion-comet/issues/1589#issuecomment-2769287839
Spark does not have a "BroadcastJoinLike" trait, so the AQE validation rule hardcodes concrete Broadcast join case classes. This is unfortunate, it makes it hard for Comet to fully utilize Spark AQE.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] AQE Unable to Rewrite Joins as Broadcast Hash Joins Due to Existing CometBroadcastHashJoin Operator [datafusion-comet]
Posted by "Kontinuation (via GitHub)" <gi...@apache.org>.
Kontinuation commented on issue #1589:
URL: https://github.com/apache/datafusion-comet/issues/1589#issuecomment-2771986460
Just took a deeper look and found that CometBroadcastHashJoin is not the root cause of AQE failure. I enabled plan change log and found that CometScanRule and CometExecRule have not kicked in yet, the broadcast hash join node is not a Comet operator but a Spark `BroadcastHashJoin` operator.
The immediate child of `BroadcastHashJoin` became a `BroadcastExchange` instead of `BroadcastQueryStage` after the `EnsureRequirements` transformation. This violates the constraint that `BroadcastQueryStage` must be the immediate child of `BroadcastHashJoin`.
Part of the original plan
```
BroadcastHashJoin [c_nationkey#3L], [n_nationkey#127L], Inner, BuildRight, (((n_name#49 = GERMANY) AND (n_name#128 = IRAQ)) OR ((n_name#49 = IRAQ) AND (n_name#128 = GERMANY))), false
:- Project [l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L, n_name#49]
: +- BroadcastHashJoin [s_nationkey#111L], [n_nationkey#48L], Inner, BuildRight, false
: :- Project [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L]
: : +- <omitted>
: +- BroadcastQueryStage 4
: +- CometBroadcastExchange [n_nationkey#48L, n_name#49]
: +- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
: +- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, ...
+- BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
```
was transformed to
```
BroadcastHashJoin [c_nationkey#3L], [n_nationkey#127L], Inner, BuildRight, (((n_name#49 = GERMANY) AND (n_name#128 = IRAQ)) OR ((n_name#49 = IRAQ) AND (n_name#128 = GERMANY))), false
:- Project [l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L, n_name#49]
: +- BroadcastHashJoin [s_nationkey#111L], [n_nationkey#48L], Inner, BuildRight, false
: :- Project [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L]
: : +- <omitted>
: +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]),false), [plan_id=916]
: +- BroadcastQueryStage 4
: +- CometBroadcastExchange [n_nationkey#48L, n_name#49]
: +- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
: +- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, ...
+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]),false), [plan_id=920]
+- BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
```
This transformation does not introduce a new `BroadcastExchange` node when not using Comet.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] AQE Unable to Rewrite Joins as Broadcast Hash Joins Due to Existing CometBroadcastHashJoin Operator [datafusion-comet]
Posted by "Kontinuation (via GitHub)" <gi...@apache.org>.
Kontinuation commented on issue #1589:
URL: https://github.com/apache/datafusion-comet/issues/1589#issuecomment-2769252236
We enabled logging for adaptive execution and obtained the following log:
```
Executing: -- SQLBench-H query 7 derived from TPC-H query 7 under the terms of the TPC Fair Use Policy.
-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
select
supp_nation,
cust_nation,
l_year,
sum(volume) as revenue
from
(
select
n1.n_name as supp_nation,
n2.n_name as cust_nation,
extract(year from l_shipdate) as l_year,
l_extendedprice * (1 - l_discount) as volume
from
supplier,
lineitem,
orders,
customer,
nation n1,
nation n2
where
s_suppkey = l_suppkey
and o_orderkey = l_orderkey
and c_custkey = o_custkey
and s_nationkey = n1.n_nationkey
and c_nationkey = n2.n_nationkey
and (
(n1.n_name = 'GERMANY' and n2.n_name = 'IRAQ')
or (n1.n_name = 'IRAQ' and n2.n_name = 'GERMANY')
)
and l_shipdate between date '1995-01-01' and date '1996-12-31'
) as shipping
group by
supp_nation,
cust_nation,
l_year
order by
supp_nation,
cust_nation,
l_year
25/04/01 17:51:33 DEBUG InsertAdaptiveSparkPlan: Adaptive execution enabled for plan: ObjectHashAggregate(keys=[], functions=[bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)], output=[bloomFilter#137])
+- ObjectHashAggregate(keys=[], functions=[partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)], output=[buf#147])
+- Project [n_nationkey#48L]
+- Filter (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- FileScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
25/04/01 17:51:33 DEBUG InsertAdaptiveSparkPlan: Adaptive execution enabled for plan: ObjectHashAggregate(keys=[], functions=[bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)], output=[bloomFilter#137])
+- ObjectHashAggregate(keys=[], functions=[partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)], output=[buf#147])
+- Project [n_nationkey#48L]
+- Filter (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- FileScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
25/04/01 17:51:33 DEBUG InsertAdaptiveSparkPlan: Adaptive execution enabled for plan: Sort [supp_nation#122 ASC NULLS FIRST, cust_nation#123 ASC NULLS FIRST, l_year#124 ASC NULLS FIRST], true, 0
+- HashAggregate(keys=[supp_nation#122, cust_nation#123, l_year#124], functions=[sum(volume#125)], output=[supp_nation#122, cust_nation#123, l_year#124, revenue#126])
+- HashAggregate(keys=[supp_nation#122, cust_nation#123, l_year#124], functions=[partial_sum(volume#125)], output=[supp_nation#122, cust_nation#123, l_year#124, sum#144, isEmpty#145])
+- Project [n_name#49 AS supp_nation#122, n_name#128 AS cust_nation#123, year(l_shipdate#26) AS l_year#124, (l_extendedprice#21 * (1 - l_discount#22)) AS volume#125]
+- BroadcastHashJoin [c_nationkey#3L], [n_nationkey#127L], Inner, BuildRight, (((n_name#49 = GERMANY) AND (n_name#128 = IRAQ)) OR ((n_name#49 = IRAQ) AND (n_name#128 = GERMANY))), false
:- Project [l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L, n_name#49]
: +- BroadcastHashJoin [s_nationkey#111L], [n_nationkey#48L], Inner, BuildRight, false
: :- Project [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L]
: : +- SortMergeJoin [o_custkey#57L], [c_custkey#0L], Inner
: : :- Project [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L]
: : : +- SortMergeJoin [l_orderkey#16L], [o_orderkey#56L], Inner
: : : :- Project [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26]
: : : : +- SortMergeJoin [s_suppkey#108L], [l_suppkey#18L], Inner
: : : : :- Project [s_suppkey#108L, s_nationkey#111L]
: : : : : +- Filter ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(scalar-subquery#138 [], xxhash64(s_nationkey#111L, 42)))
: : : : : : +- Aggregate [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0) AS bloomFilter#137]
: : : : : : +- Project [n_nationkey#48L]
: : : : : : +- Project [n_nationkey#48L]
: : : : : : +- Filter (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
: : : : : : +- Relation [n_nationkey#48L,n_name#49,n_regionkey#50L,n_comment#51] parquet
: : : : : +- FileScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint>
: : : : +- Project [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26]
: : : : +- Filter ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l_suppkey#18L)) AND isnotnull(l_orderkey#16L))
: : : : +- FileScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-12-31), isnotnul..., Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),...
: : : +- Project [o_orderkey#56L, o_custkey#57L]
: : : +- Filter (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L))
: : : +- FileScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint>
: : +- Project [c_custkey#0L, c_nationkey#3L]
: : +- Filter ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(scalar-subquery#141 [], xxhash64(c_nationkey#3L, 42)))
: : : +- Aggregate [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0) AS bloomFilter#137]
: : : +- Project [n_nationkey#48L]
: : : +- Project [n_nationkey#48L]
: : : +- Filter (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
: : : +- Relation [n_nationkey#48L,n_name#49,n_regionkey#50L,n_comment#51] parquet
: : +- FileScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint>
: +- Project [n_nationkey#48L, n_name#49]
: +- Filter (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
: +- FileScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- Project [n_nationkey#127L, n_name#128]
+- Filter (isnotnull(n_nationkey#127L) AND ((n_name#128 = IRAQ) OR (n_name#128 = GERMANY)))
+- FileScan parquet [n_nationkey#127L,n_name#128] Batched: true, DataFilters: [isnotnull(n_nationkey#127L), ((n_name#128 = IRAQ) OR (n_name#128 = GERMANY))], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,IRAQ),EqualTo(n_name,GERMANY))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
25/04/01 17:51:33 WARN QueryPlanSerde: Comet native execution is disabled due to: unsupported Spark partitioning: org.apache.spark.sql.catalyst.plans.physical.RangePartitioning
25/04/01 17:51:33 DEBUG BroadcastQueryStageExec: Materialize query stage BroadcastQueryStageExec: 4
25/04/01 17:51:33 DEBUG BroadcastQueryStageExec: Materialize query stage BroadcastQueryStageExec: 5
25/04/01 17:51:33 DEBUG ShuffleQueryStageExec: Materialize query stage ShuffleQueryStageExec: 0
25/04/01 17:51:33 DEBUG ShuffleQueryStageExec: Materialize query stage ShuffleQueryStageExec: 0
25/04/01 17:51:34 DEBUG AdaptiveSparkPlanExec: Final plan:
*(1) CometColumnarToRow
+- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- ShuffleQueryStage 0
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
25/04/01 17:51:34 DEBUG ShuffleQueryStageExec: Materialize query stage ShuffleQueryStageExec: 1
25/04/01 17:51:34 WARN DAGScheduler: Broadcasting large task binary with size 1074.0 KiB
25/04/01 17:51:34 DEBUG ShuffleQueryStageExec: Materialize query stage ShuffleQueryStageExec: 2
25/04/01 17:51:34 DEBUG ShuffleQueryStageExec: Materialize query stage ShuffleQueryStageExec: 3
25/04/01 17:51:34 WARN DAGScheduler: Broadcasting large task binary with size 1083.7 KiB
25/04/01 17:51:34 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 0
+- CometExchange hashpartitioning(s_suppkey#108L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=710]
+- CometFilter [s_suppkey#108L, s_nationkey#111L], ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(Subquery subquery#138, [id=#206], xxhash64(s_nationkey#111L, 42)))
: +- Subquery subquery#138, [id=#206]
: +- AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
*(1) CometColumnarToRow
+- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- ShuffleQueryStage 0
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- == Initial Plan ==
!CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=162]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- CometScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint>
25/04/01 17:51:34 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 1
+- CometExchange hashpartitioning(l_suppkey#18L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=724]
+- CometFilter [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l_suppkey#18L)) AND isnotnull(l_orderkey#16L))
+- CometScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-12-31), isnotnul..., Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),...
25/04/01 17:51:34 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 2
+- CometExchange hashpartitioning(o_orderkey#56L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=742]
+- CometFilter [o_orderkey#56L, o_custkey#57L], (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L))
+- CometScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint>
25/04/01 17:51:34 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 3
+- CometExchange hashpartitioning(c_custkey#0L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=763]
+- CometFilter [c_custkey#0L, c_nationkey#3L], ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(ReusedSubquery Subquery subquery#138, [id=#206], xxhash64(c_nationkey#3L, 42)))
: +- ReusedSubquery Subquery subquery#138, [id=#206]
+- CometScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint>
25/04/01 17:51:34 DEBUG AdaptiveSparkPlanExec: Re-optimize - Invalid broadcast query stage:
BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
25/04/01 17:51:34 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=1273.8 KiB, rowCount=8.00E+4) for plan: ShuffleQueryStage 0
+- CometExchange hashpartitioning(s_suppkey#108L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=710]
+- CometFilter [s_suppkey#108L, s_nationkey#111L], ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(Subquery subquery#138, [id=#206], xxhash64(s_nationkey#111L, 42)))
: +- Subquery subquery#138, [id=#206]
: +- AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
*(1) CometColumnarToRow
+- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- ShuffleQueryStage 0
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- == Initial Plan ==
!CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=162]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- CometScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint>
25/04/01 17:51:34 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 1
+- CometExchange hashpartitioning(l_suppkey#18L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=724]
+- CometFilter [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l_suppkey#18L)) AND isnotnull(l_orderkey#16L))
+- CometScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-12-31), isnotnul..., Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),...
25/04/01 17:51:34 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 2
+- CometExchange hashpartitioning(o_orderkey#56L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=742]
+- CometFilter [o_orderkey#56L, o_custkey#57L], (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L))
+- CometScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint>
25/04/01 17:51:34 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 3
+- CometExchange hashpartitioning(c_custkey#0L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=763]
+- CometFilter [c_custkey#0L, c_nationkey#3L], ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(ReusedSubquery Subquery subquery#138, [id=#206], xxhash64(c_nationkey#3L, 42)))
: +- ReusedSubquery Subquery subquery#138, [id=#206]
+- CometScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint>
25/04/01 17:51:34 DEBUG AdaptiveSparkPlanExec: Re-optimize - Invalid broadcast query stage:
BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
25/04/01 17:51:47 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=1273.8 KiB, rowCount=8.00E+4) for plan: ShuffleQueryStage 0
+- CometExchange hashpartitioning(s_suppkey#108L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=710]
+- CometFilter [s_suppkey#108L, s_nationkey#111L], ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(Subquery subquery#138, [id=#206], xxhash64(s_nationkey#111L, 42)))
: +- Subquery subquery#138, [id=#206]
: +- AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
*(1) CometColumnarToRow
+- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- ShuffleQueryStage 0
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- == Initial Plan ==
!CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=162]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- CometScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint>
25/04/01 17:51:47 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=8.9 GiB, rowCount=1.82E+8) for plan: ShuffleQueryStage 1
+- CometExchange hashpartitioning(l_suppkey#18L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=724]
+- CometFilter [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l_suppkey#18L)) AND isnotnull(l_orderkey#16L))
+- CometScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-12-31), isnotnul..., Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),...
25/04/01 17:51:47 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 2
+- CometExchange hashpartitioning(o_orderkey#56L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=742]
+- CometFilter [o_orderkey#56L, o_custkey#57L], (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L))
+- CometScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint>
25/04/01 17:51:47 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 3
+- CometExchange hashpartitioning(c_custkey#0L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=763]
+- CometFilter [c_custkey#0L, c_nationkey#3L], ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(ReusedSubquery Subquery subquery#138, [id=#206], xxhash64(c_nationkey#3L, 42)))
: +- ReusedSubquery Subquery subquery#138, [id=#206]
+- CometScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint>
25/04/01 17:51:47 DEBUG AdaptiveSparkPlanExec: Re-optimize - Invalid broadcast query stage:
BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
25/04/01 17:51:47 INFO ShufflePartitionsUtil: For shuffle(1, 2), advisory target size: 67108864, actual target size 67108864, minimum partition size: 1048576
25/04/01 17:51:47 DEBUG ShuffleQueryStageExec: Materialize query stage ShuffleQueryStageExec: 6
25/04/01 17:51:47 WARN DAGScheduler: Broadcasting large task binary with size 1096.6 KiB
25/04/01 17:51:49 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 6
+- CometExchange hashpartitioning(l_orderkey#16L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1185]
+- CometProject [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26]
+- CometSortMergeJoin [s_suppkey#108L], [l_suppkey#18L], Inner
:- CometSort [s_suppkey#108L, s_nationkey#111L], [s_suppkey#108L ASC NULLS FIRST]
: +- AQEShuffleRead coalesced
: +- ShuffleQueryStage 0
: +- CometExchange hashpartitioning(s_suppkey#108L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=710]
: +- CometFilter [s_suppkey#108L, s_nationkey#111L], ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(Subquery subquery#138, [id=#206], xxhash64(s_nationkey#111L, 42)))
: : +- Subquery subquery#138, [id=#206]
: : +- AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
*(1) CometColumnarToRow
+- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- ShuffleQueryStage 0
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- == Initial Plan ==
!CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=162]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
: +- CometScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint>
+- CometSort [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_suppkey#18L ASC NULLS FIRST]
+- AQEShuffleRead coalesced
+- ShuffleQueryStage 1
+- CometExchange hashpartitioning(l_suppkey#18L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=724]
+- CometFilter [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l_suppkey#18L)) AND isnotnull(l_orderkey#16L))
+- CometScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-12-31), isnotnul..., Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),...
25/04/01 17:51:49 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=2.2 GiB, rowCount=1.50E+8) for plan: ShuffleQueryStage 2
+- CometExchange hashpartitioning(o_orderkey#56L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=742]
+- CometFilter [o_orderkey#56L, o_custkey#57L], (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L))
+- CometScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint>
25/04/01 17:51:49 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 3
+- CometExchange hashpartitioning(c_custkey#0L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=763]
+- CometFilter [c_custkey#0L, c_nationkey#3L], ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(ReusedSubquery Subquery subquery#138, [id=#206], xxhash64(c_nationkey#3L, 42)))
: +- ReusedSubquery Subquery subquery#138, [id=#206]
+- CometScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint>
25/04/01 17:51:49 DEBUG AdaptiveSparkPlanExec: Re-optimize - Invalid broadcast query stage:
BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
25/04/01 17:51:49 DEBUG LogicalQueryStage: Physical stats not available for plan: ShuffleQueryStage 6
+- CometExchange hashpartitioning(l_orderkey#16L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1185]
+- CometProject [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26]
+- CometSortMergeJoin [s_suppkey#108L], [l_suppkey#18L], Inner
:- CometSort [s_suppkey#108L, s_nationkey#111L], [s_suppkey#108L ASC NULLS FIRST]
: +- AQEShuffleRead coalesced
: +- ShuffleQueryStage 0
: +- CometExchange hashpartitioning(s_suppkey#108L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=710]
: +- CometFilter [s_suppkey#108L, s_nationkey#111L], ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(Subquery subquery#138, [id=#206], xxhash64(s_nationkey#111L, 42)))
: : +- Subquery subquery#138, [id=#206]
: : +- AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
*(1) CometColumnarToRow
+- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- ShuffleQueryStage 0
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- == Initial Plan ==
!CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=162]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
: +- CometScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint>
+- CometSort [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_suppkey#18L ASC NULLS FIRST]
+- AQEShuffleRead coalesced
+- ShuffleQueryStage 1
+- CometExchange hashpartitioning(l_suppkey#18L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=724]
+- CometFilter [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l_suppkey#18L)) AND isnotnull(l_orderkey#16L))
+- CometScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-12-31), isnotnul..., Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),...
25/04/01 17:51:49 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=2.2 GiB, rowCount=1.50E+8) for plan: ShuffleQueryStage 2
+- CometExchange hashpartitioning(o_orderkey#56L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=742]
+- CometFilter [o_orderkey#56L, o_custkey#57L], (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L))
+- CometScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint>
25/04/01 17:51:49 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=18.6 MiB, rowCount=1.20E+6) for plan: ShuffleQueryStage 3
+- CometExchange hashpartitioning(c_custkey#0L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=763]
+- CometFilter [c_custkey#0L, c_nationkey#3L], ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(ReusedSubquery Subquery subquery#138, [id=#206], xxhash64(c_nationkey#3L, 42)))
: +- ReusedSubquery Subquery subquery#138, [id=#206]
+- CometScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint>
25/04/01 17:51:49 DEBUG AdaptiveSparkPlanExec: Re-optimize - Invalid broadcast query stage:
BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
25/04/01 17:51:55 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=724.0 MiB, rowCount=1.46E+7) for plan: ShuffleQueryStage 6
+- CometExchange hashpartitioning(l_orderkey#16L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1185]
+- CometProject [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26]
+- CometSortMergeJoin [s_suppkey#108L], [l_suppkey#18L], Inner
:- CometSort [s_suppkey#108L, s_nationkey#111L], [s_suppkey#108L ASC NULLS FIRST]
: +- AQEShuffleRead coalesced
: +- ShuffleQueryStage 0
: +- CometExchange hashpartitioning(s_suppkey#108L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=710]
: +- CometFilter [s_suppkey#108L, s_nationkey#111L], ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(Subquery subquery#138, [id=#206], xxhash64(s_nationkey#111L, 42)))
: : +- Subquery subquery#138, [id=#206]
: : +- AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
*(1) CometColumnarToRow
+- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- ShuffleQueryStage 0
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- == Initial Plan ==
!CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=162]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
: +- CometScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint>
+- CometSort [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_suppkey#18L ASC NULLS FIRST]
+- AQEShuffleRead coalesced
+- ShuffleQueryStage 1
+- CometExchange hashpartitioning(l_suppkey#18L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=724]
+- CometFilter [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l_suppkey#18L)) AND isnotnull(l_orderkey#16L))
+- CometScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-12-31), isnotnul..., Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),...
25/04/01 17:51:55 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=2.2 GiB, rowCount=1.50E+8) for plan: ShuffleQueryStage 2
+- CometExchange hashpartitioning(o_orderkey#56L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=742]
+- CometFilter [o_orderkey#56L, o_custkey#57L], (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L))
+- CometScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint>
25/04/01 17:51:55 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=18.6 MiB, rowCount=1.20E+6) for plan: ShuffleQueryStage 3
+- CometExchange hashpartitioning(c_custkey#0L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=763]
+- CometFilter [c_custkey#0L, c_nationkey#3L], ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(ReusedSubquery Subquery subquery#138, [id=#206], xxhash64(c_nationkey#3L, 42)))
: +- ReusedSubquery Subquery subquery#138, [id=#206]
+- CometScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint>
25/04/01 17:51:55 DEBUG AdaptiveSparkPlanExec: Re-optimize - Invalid broadcast query stage:
BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
25/04/01 17:51:55 INFO ShufflePartitionsUtil: For shuffle(5, 3), advisory target size: 67108864, actual target size 67108864, minimum partition size: 1048576
25/04/01 17:51:55 DEBUG ShuffleQueryStageExec: Materialize query stage ShuffleQueryStageExec: 7
25/04/01 17:51:55 WARN DAGScheduler: Broadcasting large task binary with size 1112.1 KiB
25/04/01 17:52:02 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=724.0 MiB, rowCount=1.46E+7) for plan: ShuffleQueryStage 7
+- CometExchange hashpartitioning(o_custkey#57L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1496]
+- CometProject [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L], [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L]
+- CometSortMergeJoin [l_orderkey#16L], [o_orderkey#56L], Inner
:- CometSort [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_orderkey#16L ASC NULLS FIRST]
: +- AQEShuffleRead coalesced
: +- ShuffleQueryStage 6
: +- CometExchange hashpartitioning(l_orderkey#16L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1185]
: +- CometProject [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26]
: +- CometSortMergeJoin [s_suppkey#108L], [l_suppkey#18L], Inner
: :- CometSort [s_suppkey#108L, s_nationkey#111L], [s_suppkey#108L ASC NULLS FIRST]
: : +- AQEShuffleRead coalesced
: : +- ShuffleQueryStage 0
: : +- CometExchange hashpartitioning(s_suppkey#108L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=710]
: : +- CometFilter [s_suppkey#108L, s_nationkey#111L], ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(Subquery subquery#138, [id=#206], xxhash64(s_nationkey#111L, 42)))
: : : +- Subquery subquery#138, [id=#206]
: : : +- AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
*(1) CometColumnarToRow
+- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- ShuffleQueryStage 0
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- == Initial Plan ==
!CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=162]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
: : +- CometScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint>
: +- CometSort [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_suppkey#18L ASC NULLS FIRST]
: +- AQEShuffleRead coalesced
: +- ShuffleQueryStage 1
: +- CometExchange hashpartitioning(l_suppkey#18L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=724]
: +- CometFilter [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l_suppkey#18L)) AND isnotnull(l_orderkey#16L))
: +- CometScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-12-31), isnotnul..., Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),...
+- CometSort [o_orderkey#56L, o_custkey#57L], [o_orderkey#56L ASC NULLS FIRST]
+- AQEShuffleRead coalesced
+- ShuffleQueryStage 2
+- CometExchange hashpartitioning(o_orderkey#56L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=742]
+- CometFilter [o_orderkey#56L, o_custkey#57L], (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L))
+- CometScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint>
25/04/01 17:52:02 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=18.6 MiB, rowCount=1.20E+6) for plan: ShuffleQueryStage 3
+- CometExchange hashpartitioning(c_custkey#0L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=763]
+- CometFilter [c_custkey#0L, c_nationkey#3L], ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(ReusedSubquery Subquery subquery#138, [id=#206], xxhash64(c_nationkey#3L, 42)))
: +- ReusedSubquery Subquery subquery#138, [id=#206]
+- CometScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint>
25/04/01 17:52:02 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=337.0 B, rowCount=2) for plan: BroadcastQueryStage 4
+- CometBroadcastExchange [n_nationkey#48L, n_name#49]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
25/04/01 17:52:02 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=337.0 B, rowCount=2) for plan: BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
25/04/01 17:52:02 DEBUG AdaptiveSparkPlanExec: Re-optimize - Invalid broadcast query stage:
BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
25/04/01 17:52:02 INFO ShufflePartitionsUtil: For shuffle(6, 4), advisory target size: 67108864, actual target size 34760648, minimum partition size: 1048576
25/04/01 17:52:02 DEBUG ShuffleQueryStageExec: Materialize query stage ShuffleQueryStageExec: 8
25/04/01 17:52:02 WARN DAGScheduler: Broadcasting large task binary with size 2.2 MiB
25/04/01 17:52:03 DEBUG LogicalQueryStage: Physical stats available as Statistics(sizeInBytes=6.0 KiB, rowCount=36) for plan: !CometHashAggregate [supp_nation#122, cust_nation#123, l_year#124, sum#144, isEmpty#145], Final, [supp_nation#122, cust_nation#123, l_year#124], [sum(volume#125)]
+- ShuffleQueryStage 8
+- CometExchange hashpartitioning(supp_nation#122, cust_nation#123, l_year#124, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1656]
+- !CometHashAggregate [supp_nation#122, cust_nation#123, l_year#124, volume#125], Partial, [supp_nation#122, cust_nation#123, l_year#124], [partial_sum(volume#125)]
+- CometProject [supp_nation#122, cust_nation#123, l_year#124, volume#125], [n_name#49 AS supp_nation#122, n_name#128 AS cust_nation#123, year(l_shipdate#26) AS l_year#124, (l_extendedprice#21 * (1 - l_discount#22)) AS volume#125]
+- CometBroadcastHashJoin [c_nationkey#3L], [n_nationkey#127L], Inner, (((n_name#49 = GERMANY) AND (n_name#128 = IRAQ)) OR ((n_name#49 = IRAQ) AND (n_name#128 = GERMANY))), BuildRight
:- CometProject [l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L, n_name#49], [l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L, n_name#49]
: +- CometBroadcastHashJoin [s_nationkey#111L], [n_nationkey#48L], Inner, BuildRight
: :- CometProject [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L], [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L]
: : +- CometSortMergeJoin [o_custkey#57L], [c_custkey#0L], Inner
: : :- CometSort [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L], [o_custkey#57L ASC NULLS FIRST]
: : : +- AQEShuffleRead coalesced
: : : +- ShuffleQueryStage 7
: : : +- CometExchange hashpartitioning(o_custkey#57L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1496]
: : : +- CometProject [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L], [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L]
: : : +- CometSortMergeJoin [l_orderkey#16L], [o_orderkey#56L], Inner
: : : :- CometSort [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_orderkey#16L ASC NULLS FIRST]
: : : : +- AQEShuffleRead coalesced
: : : : +- ShuffleQueryStage 6
: : : : +- CometExchange hashpartitioning(l_orderkey#16L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1185]
: : : : +- CometProject [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26]
: : : : +- CometSortMergeJoin [s_suppkey#108L], [l_suppkey#18L], Inner
: : : : :- CometSort [s_suppkey#108L, s_nationkey#111L], [s_suppkey#108L ASC NULLS FIRST]
: : : : : +- AQEShuffleRead coalesced
: : : : : +- ShuffleQueryStage 0
: : : : : +- CometExchange hashpartitioning(s_suppkey#108L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=710]
: : : : : +- CometFilter [s_suppkey#108L, s_nationkey#111L], ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(Subquery subquery#138, [id=#206], xxhash64(s_nationkey#111L, 42)))
: : : : : : +- Subquery subquery#138, [id=#206]
: : : : : : +- AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
*(1) CometColumnarToRow
+- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- ShuffleQueryStage 0
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- == Initial Plan ==
!CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=162]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
: : : : : +- CometScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint>
: : : : +- CometSort [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_suppkey#18L ASC NULLS FIRST]
: : : : +- AQEShuffleRead coalesced
: : : : +- ShuffleQueryStage 1
: : : : +- CometExchange hashpartitioning(l_suppkey#18L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=724]
: : : : +- CometFilter [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l_suppkey#18L)) AND isnotnull(l_orderkey#16L))
: : : : +- CometScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-12-31), isnotnul..., Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),...
: : : +- CometSort [o_orderkey#56L, o_custkey#57L], [o_orderkey#56L ASC NULLS FIRST]
: : : +- AQEShuffleRead coalesced
: : : +- ShuffleQueryStage 2
: : : +- CometExchange hashpartitioning(o_orderkey#56L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=742]
: : : +- CometFilter [o_orderkey#56L, o_custkey#57L], (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L))
: : : +- CometScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint>
: : +- CometSort [c_custkey#0L, c_nationkey#3L], [c_custkey#0L ASC NULLS FIRST]
: : +- AQEShuffleRead coalesced
: : +- ShuffleQueryStage 3
: : +- CometExchange hashpartitioning(c_custkey#0L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=763]
: : +- CometFilter [c_custkey#0L, c_nationkey#3L], ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(ReusedSubquery Subquery subquery#138, [id=#206], xxhash64(c_nationkey#3L, 42)))
: : : +- ReusedSubquery Subquery subquery#138, [id=#206]
: : +- CometScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint>
: +- BroadcastQueryStage 4
: +- CometBroadcastExchange [n_nationkey#48L, n_name#49]
: +- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
: +- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
25/04/01 17:52:03 WARN QueryPlanSerde: Comet native execution is disabled due to: unsupported Spark partitioning: org.apache.spark.sql.catalyst.plans.physical.RangePartitioning
25/04/01 17:52:03 DEBUG AdaptiveSparkPlanExec: Plan changed:
CometSort [supp_nation#122, cust_nation#123, l_year#124, revenue#126], [supp_nation#122 ASC NULLS FIRST, cust_nation#123 ASC NULLS FIRST, l_year#124 ASC NULLS FIRST] CometSort [supp_nation#122, cust_nation#123, l_year#124, revenue#126], [supp_nation#122 ASC NULLS FIRST, cust_nation#123 ASC NULLS FIRST, l_year#124 ASC NULLS FIRST]
!+- CometColumnarExchange rangepartitioning(supp_nation#122 ASC NULLS FIRST, cust_nation#123 ASC NULLS FIRST, l_year#124 ASC NULLS FIRST, 200), ENSURE_REQUIREMENTS, CometColumnarShuffle, [plan_id=1696] +- CometColumnarExchange rangepartitioning(supp_nation#122 ASC NULLS FIRST, cust_nation#123 ASC NULLS FIRST, l_year#124 ASC NULLS FIRST, 200), ENSURE_REQUIREMENTS, CometColumnarShuffle, [plan_id=1712]
+- !CometHashAggregate [supp_nation#122, cust_nation#123, l_year#124, sum#144, isEmpty#145], Final, [supp_nation#122, cust_nation#123, l_year#124], [sum(volume#125)] +- !CometHashAggregate [supp_nation#122, cust_nation#123, l_year#124, sum#144, isEmpty#145], Final, [supp_nation#122, cust_nation#123, l_year#124], [sum(volume#125)]
+- ShuffleQueryStage 8 +- ShuffleQueryStage 8
+- CometExchange hashpartitioning(supp_nation#122, cust_nation#123, l_year#124, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1656] +- CometExchange hashpartitioning(supp_nation#122, cust_nation#123, l_year#124, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1656]
+- !CometHashAggregate [supp_nation#122, cust_nation#123, l_year#124, volume#125], Partial, [supp_nation#122, cust_nation#123, l_year#124], [partial_sum(volume#125)] +- !CometHashAggregate [supp_nation#122, cust_nation#123, l_year#124, volume#125], Partial, [supp_nation#122, cust_nation#123, l_year#124], [partial_sum(volume#125)]
+- CometProject [supp_nation#122, cust_nation#123, l_year#124, volume#125], [n_name#49 AS supp_nation#122, n_name#128 AS cust_nation#123, year(l_shipdate#26) AS l_year#124, (l_extendedprice#21 * (1 - l_discount#22)) AS volume#125] +- CometProject [supp_nation#122, cust_nation#123, l_year#124, volume#125], [n_name#49 AS supp_nation#122, n_name#128 AS cust_nation#123, year(l_shipdate#26) AS l_year#124, (l_extendedprice#21 * (1 - l_discount#22)) AS volume#125]
+- CometBroadcastHashJoin [c_nationkey#3L], [n_nationkey#127L], Inner, (((n_name#49 = GERMANY) AND (n_name#128 = IRAQ)) OR ((n_name#49 = IRAQ) AND (n_name#128 = GERMANY))), BuildRight +- CometBroadcastHashJoin [c_nationkey#3L], [n_nationkey#127L], Inner, (((n_name#49 = GERMANY) AND (n_name#128 = IRAQ)) OR ((n_name#49 = IRAQ) AND (n_name#128 = GERMANY))), BuildRight
:- CometProject [l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L, n_name#49], [l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L, n_name#49] :- CometProject [l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L, n_name#49], [l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L, n_name#49]
: +- CometBroadcastHashJoin [s_nationkey#111L], [n_nationkey#48L], Inner, BuildRight : +- CometBroadcastHashJoin [s_nationkey#111L], [n_nationkey#48L], Inner, BuildRight
: :- CometProject [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L], [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L] : :- CometProject [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L], [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L]
: : +- CometSortMergeJoin [o_custkey#57L], [c_custkey#0L], Inner : : +- CometSortMergeJoin [o_custkey#57L], [c_custkey#0L], Inner
: : :- CometSort [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L], [o_custkey#57L ASC NULLS FIRST] : : :- CometSort [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L], [o_custkey#57L ASC NULLS FIRST]
: : : +- AQEShuffleRead coalesced : : : +- AQEShuffleRead coalesced
: : : +- ShuffleQueryStage 7 : : : +- ShuffleQueryStage 7
: : : +- CometExchange hashpartitioning(o_custkey#57L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1496] : : : +- CometExchange hashpartitioning(o_custkey#57L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1496]
: : : +- CometProject [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L], [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L] : : : +- CometProject [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L], [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L]
: : : +- CometSortMergeJoin [l_orderkey#16L], [o_orderkey#56L], Inner : : : +- CometSortMergeJoin [l_orderkey#16L], [o_orderkey#56L], Inner
: : : :- CometSort [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_orderkey#16L ASC NULLS FIRST] : : : :- CometSort [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_orderkey#16L ASC NULLS FIRST]
: : : : +- AQEShuffleRead coalesced : : : : +- AQEShuffleRead coalesced
: : : : +- ShuffleQueryStage 6 : : : : +- ShuffleQueryStage 6
: : : : +- CometExchange hashpartitioning(l_orderkey#16L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1185] : : : : +- CometExchange hashpartitioning(l_orderkey#16L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1185]
: : : : +- CometProject [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26] : : : : +- CometProject [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26]
: : : : +- CometSortMergeJoin [s_suppkey#108L], [l_suppkey#18L], Inner : : : : +- CometSortMergeJoin [s_suppkey#108L], [l_suppkey#18L], Inner
: : : : :- CometSort [s_suppkey#108L, s_nationkey#111L], [s_suppkey#108L ASC NULLS FIRST] : : : : :- CometSort [s_suppkey#108L, s_nationkey#111L], [s_suppkey#108L ASC NULLS FIRST]
: : : : : +- AQEShuffleRead coalesced : : : : : +- AQEShuffleRead coalesced
: : : : : +- ShuffleQueryStage 0 : : : : : +- ShuffleQueryStage 0
: : : : : +- CometExchange hashpartitioning(s_suppkey#108L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=710] : : : : : +- CometExchange hashpartitioning(s_suppkey#108L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=710]
: : : : : +- CometFilter [s_suppkey#108L, s_nationkey#111L], ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(Subquery subquery#138, [id=#206], xxhash64(s_nationkey#111L, 42))) : : : : : +- CometFilter [s_suppkey#108L, s_nationkey#111L], ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(Subquery subquery#138, [id=#206], xxhash64(s_nationkey#111L, 42)))
: : : : : : +- Subquery subquery#138, [id=#206] : : : : : : +- Subquery subquery#138, [id=#206]
: : : : : : +- AdaptiveSparkPlan isFinalPlan=true : : : : : : +- AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan == +- == Final Plan ==
*(1) CometColumnarToRow *(1) CometColumnarToRow
+- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)] +- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- ShuffleQueryStage 0 +- ShuffleQueryStage 0
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818] +- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)] +- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L] +- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))) +- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string> +- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Loc
ation: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- == Initial Plan == +- == Initial Plan ==
!CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)] !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=162] +- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=162]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)] +- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L] +- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))) +- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string> +- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location:
InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
: : : : : +- CometScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint> : : : : : +- CometScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Use
rs/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint>
: : : : +- CometSort [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_suppkey#18L ASC NULLS FIRST] : : : : +- CometSort [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_suppkey#18L ASC NULLS FIRST]
: : : : +- AQEShuffleRead coalesced : : : : +- AQEShuffleRead coalesced
: : : : +- ShuffleQueryStage 1 : : : : +- ShuffleQueryStage 1
: : : : +- CometExchange hashpartitioning(l_suppkey#18L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=724] : : : : +- CometExchange hashpartitioning(l_suppkey#18L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=724]
: : : : +- CometFilter [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l_suppkey#18L)) AND isnotnull(l_orderkey#16L)) : : : : +- CometFilter [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l
_suppkey#18L)) AND isnotnull(l_orderkey#16L))
: : : : +- CometScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-12-31), isnotnul..., Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),... : : : : +- CometScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-1
2-31), isnotnul..., Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),...
: : : +- CometSort [o_orderkey#56L, o_custkey#57L], [o_orderkey#56L ASC NULLS FIRST] : : : +- CometSort [o_orderkey#56L, o_custkey#57L], [o_orderkey#56L ASC NULLS FIRST]
: : : +- AQEShuffleRead coalesced : : : +- AQEShuffleRead coalesced
: : : +- ShuffleQueryStage 2 : : : +- ShuffleQueryStage 2
: : : +- CometExchange hashpartitioning(o_orderkey#56L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=742] : : : +- CometExchange hashpartitioning(o_orderkey#56L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=742]
: : : +- CometFilter [o_orderkey#56L, o_custkey#57L], (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L)) : : : +- CometFilter [o_orderkey#56L, o_custkey#57L], (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L))
: : : +- CometScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint> : : : +- CometScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/gith
ub/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint>
: : +- CometSort [c_custkey#0L, c_nationkey#3L], [c_custkey#0L ASC NULLS FIRST] : : +- CometSort [c_custkey#0L, c_nationkey#3L], [c_custkey#0L ASC NULLS FIRST]
: : +- AQEShuffleRead coalesced : : +- AQEShuffleRead coalesced
: : +- ShuffleQueryStage 3 : : +- ShuffleQueryStage 3
: : +- CometExchange hashpartitioning(c_custkey#0L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=763] : : +- CometExchange hashpartitioning(c_custkey#0L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=763]
: : +- CometFilter [c_custkey#0L, c_nationkey#3L], ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(ReusedSubquery Subquery subquery#138, [id=#206], xxhash64(c_nationkey#3L, 42))) : : +- CometFilter [c_custkey#0L, c_nationkey#3L], ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(ReusedSubquery Subquery subquery#138, [id=#206], xxhash64(c_nationkey#3L, 42)))
: : : +- ReusedSubquery Subquery subquery#138, [id=#206] : : : +- ReusedSubquery Subquery subquery#138, [id=#206]
: : +- CometScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint> : : +- CometScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/
tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint>
: +- BroadcastQueryStage 4 : +- BroadcastQueryStage 4
: +- CometBroadcastExchange [n_nationkey#48L, n_name#49] : +- CometBroadcastExchange [n_nationkey#48L, n_name#49]
: +- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))) : +- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
: +- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string> : +- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusi
on-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- BroadcastQueryStage 5 +- BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49] +- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
25/04/01 17:52:03 INFO ShufflePartitionsUtil: For shuffle(7), advisory target size: 67108864, actual target size 1048576, minimum partition size: 1048576
25/04/01 17:52:03 DEBUG ShuffleQueryStageExec: Materialize query stage ShuffleQueryStageExec: 9
25/04/01 17:52:03 WARN DAGScheduler: Broadcasting large task binary with size 2.2 MiB
25/04/01 17:52:03 WARN DAGScheduler: Broadcasting large task binary with size 2.2 MiB
25/04/01 17:52:03 WARN QueryPlanSerde: Comet native execution is disabled due to: unsupported Spark partitioning: org.apache.spark.sql.catalyst.plans.physical.RangePartitioning
25/04/01 17:52:03 INFO ShufflePartitionsUtil: For shuffle(8), advisory target size: 67108864, actual target size 1048576, minimum partition size: 1048576
25/04/01 17:52:03 WARN DAGScheduler: Broadcasting large task binary with size 2.2 MiB
25/04/01 17:52:03 DEBUG AdaptiveSparkPlanExec: Final plan:
*(1) CometColumnarToRow
+- CometSort [supp_nation#122, cust_nation#123, l_year#124, revenue#126], [supp_nation#122 ASC NULLS FIRST, cust_nation#123 ASC NULLS FIRST, l_year#124 ASC NULLS FIRST]
+- AQEShuffleRead coalesced
+- ShuffleQueryStage 9
+- CometColumnarExchange rangepartitioning(supp_nation#122 ASC NULLS FIRST, cust_nation#123 ASC NULLS FIRST, l_year#124 ASC NULLS FIRST, 200), ENSURE_REQUIREMENTS, CometColumnarShuffle, [plan_id=1731]
+- !CometHashAggregate [supp_nation#122, cust_nation#123, l_year#124, sum#144, isEmpty#145], Final, [supp_nation#122, cust_nation#123, l_year#124], [sum(volume#125)]
+- AQEShuffleRead coalesced
+- ShuffleQueryStage 8
+- CometExchange hashpartitioning(supp_nation#122, cust_nation#123, l_year#124, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1656]
+- !CometHashAggregate [supp_nation#122, cust_nation#123, l_year#124, volume#125], Partial, [supp_nation#122, cust_nation#123, l_year#124], [partial_sum(volume#125)]
+- CometProject [supp_nation#122, cust_nation#123, l_year#124, volume#125], [n_name#49 AS supp_nation#122, n_name#128 AS cust_nation#123, year(l_shipdate#26) AS l_year#124, (l_extendedprice#21 * (1 - l_discount#22)) AS volume#125]
+- CometBroadcastHashJoin [c_nationkey#3L], [n_nationkey#127L], Inner, (((n_name#49 = GERMANY) AND (n_name#128 = IRAQ)) OR ((n_name#49 = IRAQ) AND (n_name#128 = GERMANY))), BuildRight
:- CometProject [l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L, n_name#49], [l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L, n_name#49]
: +- CometBroadcastHashJoin [s_nationkey#111L], [n_nationkey#48L], Inner, BuildRight
: :- CometProject [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L], [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, c_nationkey#3L]
: : +- CometSortMergeJoin [o_custkey#57L], [c_custkey#0L], Inner
: : :- CometSort [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L], [o_custkey#57L ASC NULLS FIRST]
: : : +- AQEShuffleRead coalesced
: : : +- ShuffleQueryStage 7
: : : +- CometExchange hashpartitioning(o_custkey#57L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1496]
: : : +- CometProject [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L], [s_nationkey#111L, l_extendedprice#21, l_discount#22, l_shipdate#26, o_custkey#57L]
: : : +- CometSortMergeJoin [l_orderkey#16L], [o_orderkey#56L], Inner
: : : :- CometSort [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_orderkey#16L ASC NULLS FIRST]
: : : : +- AQEShuffleRead coalesced
: : : : +- ShuffleQueryStage 6
: : : : +- CometExchange hashpartitioning(l_orderkey#16L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=1185]
: : : : +- CometProject [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26], [s_nationkey#111L, l_orderkey#16L, l_extendedprice#21, l_discount#22, l_shipdate#26]
: : : : +- CometSortMergeJoin [s_suppkey#108L], [l_suppkey#18L], Inner
: : : : :- CometSort [s_suppkey#108L, s_nationkey#111L], [s_suppkey#108L ASC NULLS FIRST]
: : : : : +- AQEShuffleRead coalesced
: : : : : +- ShuffleQueryStage 0
: : : : : +- CometExchange hashpartitioning(s_suppkey#108L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=710]
: : : : : +- CometFilter [s_suppkey#108L, s_nationkey#111L], ((isnotnull(s_suppkey#108L) AND isnotnull(s_nationkey#111L)) AND might_contain(Subquery subquery#138, [id=#206], xxhash64(s_nationkey#111L, 42)))
: : : : : : +- Subquery subquery#138, [id=#206]
: : : : : : +- AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
*(1) CometColumnarToRow
+- !CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- ShuffleQueryStage 0
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=818]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- == Initial Plan ==
!CometHashAggregate [buf#147], Final, [bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometExchange SinglePartition, ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=162]
+- !CometHashAggregate [n_nationkey#48L], Partial, [partial_bloom_filter_agg(xxhash64(n_nationkey#48L, 42), 1000000, 8388608, 0, 0)]
+- CometProject [n_nationkey#48L], [n_nationkey#48L]
+- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
+- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
: : : : : +- CometScan parquet [s_suppkey#108L,s_nationkey#111L] Batched: true, DataFilters: [isnotnull(s_suppkey#108L), isnotnull(s_nationkey#111L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(s_suppkey), IsNotNull(s_nationkey)], ReadSchema: struct<s_suppkey:bigint,s_nationkey:bigint>
: : : : +- CometSort [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], [l_suppkey#18L ASC NULLS FIRST]
: : : : +- AQEShuffleRead coalesced
: : : : +- ShuffleQueryStage 1
: : : : +- CometExchange hashpartitioning(l_suppkey#18L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=724]
: : : : +- CometFilter [l_orderkey#16L, l_suppkey#18L, l_extendedprice#21, l_discount#22, l_shipdate#26], ((((isnotnull(l_shipdate#26) AND (l_shipdate#26 >= 1995-01-01)) AND (l_shipdate#26 <= 1996-12-31)) AND isnotnull(l_suppkey#18L)) AND isnotnull(l_orderkey#16L))
: : : : +- CometScan parquet [l_orderkey#16L,l_suppkey#18L,l_extendedprice#21,l_discount#22,l_shipdate#26] Batched: true, DataFilters: [isnotnull(l_shipdate#26), (l_shipdate#26 >= 1995-01-01), (l_shipdate#26 <= 1996-12-31), isnotnul..., Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(l_shipdate), GreaterThanOrEqual(l_shipdate,1995-01-01), LessThanOrEqual(l_shipdate,199..., ReadSchema: struct<l_orderkey:bigint,l_suppkey:bigint,l_extendedprice:decimal(12,2),l_discount:decimal(12,2),...
: : : +- CometSort [o_orderkey#56L, o_custkey#57L], [o_orderkey#56L ASC NULLS FIRST]
: : : +- AQEShuffleRead coalesced
: : : +- ShuffleQueryStage 2
: : : +- CometExchange hashpartitioning(o_orderkey#56L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=742]
: : : +- CometFilter [o_orderkey#56L, o_custkey#57L], (isnotnull(o_orderkey#56L) AND isnotnull(o_custkey#57L))
: : : +- CometScan parquet [o_orderkey#56L,o_custkey#57L] Batched: true, DataFilters: [isnotnull(o_orderkey#56L), isnotnull(o_custkey#57L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderkey), IsNotNull(o_custkey)], ReadSchema: struct<o_orderkey:bigint,o_custkey:bigint>
: : +- CometSort [c_custkey#0L, c_nationkey#3L], [c_custkey#0L ASC NULLS FIRST]
: : +- AQEShuffleRead coalesced
: : +- ShuffleQueryStage 3
: : +- CometExchange hashpartitioning(c_custkey#0L, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=763]
: : +- CometFilter [c_custkey#0L, c_nationkey#3L], ((isnotnull(c_custkey#0L) AND isnotnull(c_nationkey#3L)) AND might_contain(ReusedSubquery Subquery subquery#138, [id=#206], xxhash64(c_nationkey#3L, 42)))
: : : +- ReusedSubquery Subquery subquery#138, [id=#206]
: : +- CometScan parquet [c_custkey#0L,c_nationkey#3L] Batched: true, DataFilters: [isnotnull(c_custkey#0L), isnotnull(c_nationkey#3L)], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(c_custkey), IsNotNull(c_nationkey)], ReadSchema: struct<c_custkey:bigint,c_nationkey:bigint>
: +- BroadcastQueryStage 4
: +- CometBroadcastExchange [n_nationkey#48L, n_name#49]
: +- CometFilter [n_nationkey#48L, n_name#49], (isnotnull(n_nationkey#48L) AND ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ)))
: +- CometScan parquet [n_nationkey#48L,n_name#49] Batched: true, DataFilters: [isnotnull(n_nationkey#48L), ((n_name#49 = GERMANY) OR (n_name#49 = IRAQ))], Format: CometParquet, Location: InMemoryFileIndex(1 paths)[file:/Users/bopeng/workspace/github/datafusion-comet/tpch/sf100_parque..., PartitionFilters: [], PushedFilters: [IsNotNull(n_nationkey), Or(EqualTo(n_name,GERMANY),EqualTo(n_name,IRAQ))], ReadSchema: struct<n_nationkey:bigint,n_name:string>
+- BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
Query 7 returned 4 rows
Query 7 took 30.72541904449463 seconds
```
The following lines in the log shows why AQE didn't proceed:
```
25/04/01 17:51:47 DEBUG AdaptiveSparkPlanExec: Re-optimize - Invalid broadcast query stage:
BroadcastQueryStage 5
+- ReusedExchange [n_nationkey#127L, n_name#128], CometBroadcastExchange [n_nationkey#48L, n_name#49]
```
There's a [ValidateSparkPlan rule](https://github.com/apache/spark/blob/v3.5.5/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ValidateSparkPlan.scala) checking if `BroadcastQueryStage` only appears on the build side of `BroadcastHashJoinExec` or `BroadcastNestedLoopJoinExec`. It does not recognize that `CometBroadcastHashJoinExec` is actually a broadcast hash join so the validation failed and the AQE query optimization was aborted. The CometBroadcastHashJoin nodes in the initial plan disrupted the AQE optimization.
I tried running a simplified version of TPC-H Q7, which does not have CometBroadcastHashJoin in the initial plan, AQE successfully transformed the CometSortMergeJoin into CometBroadcastHashJoin:
```sql
select
l_year,
sum(volume) as revenue
from
(
select
extract(year from l_shipdate) as l_year,
l_extendedprice * (1 - l_discount) as volume
from
supplier,
lineitem
where
s_suppkey = l_suppkey
and l_shipdate between date '1995-01-01' and date '1996-12-31'
) as shipping
group by
l_year
order by
l_year;
```
The query plan is:
```
== Physical Plan ==
AdaptiveSparkPlan (38)
+- == Final Plan ==
* CometColumnarToRow (24)
+- CometSort (23)
+- AQEShuffleRead (22)
+- ShuffleQueryStage (21), Statistics(sizeInBytes=80.0 B, rowCount=2)
+- CometColumnarExchange (20)
+- CometHashAggregate (19)
+- AQEShuffleRead (18)
+- ShuffleQueryStage (17), Statistics(sizeInBytes=15.7 KiB, rowCount=100)
+- CometExchange (16)
+- CometHashAggregate (15)
+- CometProject (14)
+- CometBroadcastHashJoin (13) <-- Transformed from CometSortMergeJoin
:- BroadcastQueryStage (7), Statistics(sizeInBytes=5.3 MiB, rowCount=1.00E+6)
: +- CometBroadcastExchange (6)
: +- AQEShuffleRead (5)
: +- ShuffleQueryStage (4), Statistics(sizeInBytes=7.6 MiB, rowCount=1.00E+6)
: +- CometExchange (3)
: +- CometFilter (2)
: +- CometScan parquet (1)
+- AQEShuffleRead (12)
+- ShuffleQueryStage (11), Statistics(sizeInBytes=7.5 GiB, rowCount=1.82E+8)
+- CometExchange (10)
+- CometFilter (9)
+- CometScan parquet (8)
+- == Initial Plan ==
CometSort (37)
+- CometColumnarExchange (36)
+- CometHashAggregate (35)
+- CometExchange (34)
+- CometHashAggregate (33)
+- CometProject (32)
+- CometSortMergeJoin (31)
:- CometSort (27)
: +- CometExchange (26)
: +- CometFilter (25)
: +- CometScan parquet (1)
+- CometSort (30)
+- CometExchange (29)
+- CometFilter (28)
+- CometScan parquet (8)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] AQE Unable to Rewrite Joins as Broadcast Hash Joins Due to Existing CometBroadcastHashJoin Operator [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove closed issue #1589: AQE Unable to Rewrite Joins as Broadcast Hash Joins Due to Existing CometBroadcastHashJoin Operator
URL: https://github.com/apache/datafusion-comet/issues/1589
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] AQE Unable to Rewrite Joins as Broadcast Hash Joins Due to Existing CometBroadcastHashJoin Operator [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on issue #1589:
URL: https://github.com/apache/datafusion-comet/issues/1589#issuecomment-2769269205
Good catch, @Kontinuation! https://github.com/apache/datafusion-comet/pull/1578 has me looking at AQE wondering if there are other places where Comet isn't working with AQE where we should be. This is motivating to keep investigating.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org