sunchao commented on code in PR #6938:
URL: https://github.com/apache/datafusion-comet/pull/6938#discussion_r4266135285
##########
native/core/src/execution/planner.rs:
##########
@@ -1989,6 +1989,36 @@ impl PhysicalPlanner {
))
}
}
+ OpStruct::CrossJoin(_) => {
+ // Spark's CartesianProductExec, delegated to DataFusion's
CrossJoinExec. A pure
+ // cross product has no join keys and no condition; the Inner
join type here only
+ // satisfies the shared helper and is unused by CrossJoinExec.
CrossJoinExec emits
+ // left columns followed by right columns, matching Spark's
output order.
+ let (join_params, scans, shuffle_scans) =
self.parse_join_parameters(
+ inputs,
+ children,
+ &[],
+ &[],
+ JoinType::Inner as i32,
+ &None,
+ partition_count,
+ )?;
+
+ let left = Arc::clone(&join_params.left.native_plan);
+ let right = Arc::clone(&join_params.right.native_plan);
+
+ let cross_join = Arc::new(CrossJoinExec::new(left, right));
Review Comment:
[P2] [P2] Preserve bounded-memory execution before enabling native Cartesian
products
`CrossJoinExec::new(left, right)` collects the entire left input and
propagates a refused memory reservation without spilling. Spark’s
`UnsafeCartesianRDD` instead streams the left input and uses spillable
buffering for the right. Consequently, the single-partition guard admits
queries that previously succeeded but now fail by default. A million-row left
range crossed with a one-row right range reproduced this under a 4 MiB off-heap
pool: Spark returned the expected sum, while native execution failed at its 2
MiB fair share. Could this preserve spillable execution, or remain opt-in until
it can, with a regression covering this case?
Evidence: Executed a disposable `Review6938Suite` against exact-head JVM
sources and CI native artifact 11841794909 on Spark 4.1.3/JDK 21.
Configuration: `spark.memory.offHeap.enabled=true`,
`spark.memory.offHeap.size=4m`, `spark.comet.convert.range.enabled=true`, both
broadcast thresholds `-1`, and `spark.sql.adaptive.enabled=false`. Query:
`SELECT sum(l.id) FROM range(0, 1000000, 1, 1) l CROSS JOIN range(0, 1, 1, 1)
r`. With `spark.comet.exec.cartesianProduct.enabled=false`, the executed plan
contained Spark `CartesianProduct` and returned `499999500000`. With the flag
enabled, the plan contained `CometCartesianProduct` and failed with `Resources
exhausted: Additional allocation failed for CrossJoinExec`, `can spill: false`,
and `Failed to acquire 65632 bytes ... already holds 2034592 bytes ... fair
limit is 2097152 bytes`. DataFusion 55.2.0 `load_left_input` confirms
unconditional accumulation through `reservation.try_grow(batch_size)?`.
Validation log: `/tmp/pr6938-current-val
idation/repro-localhost.log`.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
