andygrove commented on issue #6499: URL: https://github.com/apache/datafusion-comet/issues/6499#issuecomment-6043295523
Thanks, the plan looks good to me. On the Murmur3 question: two `RepartitionExec`s with DataFusion's hash do co-partition with each other, so the hash only matters when one side of a join gets its partitioning from a leaf instead of an exchange. Bucketed tables are the common case. `CometNativeScanExec` reports the bucket `HashPartitioning`, and `EnsureRequirements` keeps the side that needs no exchange and shuffles only the other side to match it. If that exchange becomes a `RepartitionExec` with DataFusion's hash, partition i of the shuffled side no longer lines up with bucket i and the join returns wrong rows. Could admission decline plans where a join child's partitioning comes from a scan (bucketed `HashPartitioning`, or `KeyGroupedPartitioning` from a storage-partitioned join), or could the local exchange keep Comet's Spark-compatible hash? A join between a bucketed and a non-bucketed table on the bucket column would make a good test. Lowering each block once with `PhysicalPlanner::create_plan` has a related catch. Every DataFusion partition shares one instance of each planned expression, and that instance carries the planner's partition index. So `spark_partition_id()`, `monotonically_increasing_id()`, the `rand` and `randn` seeds, `Sample`'s seed and the partition index `JvmScalarUdfExpr` hands the dispatcher would all see partition 0. Could the `deterministic` gate and the `nativeOnly` protobuf walk in `LocalParquetPlanner` carry over to the operator-tree admission? For the first PR, the range needs an overflow guard. The local range admits ranges whose arithmetic overflows in Spark. On this head, `spark.range(Long.MinValue, Long.MaxValue, 1L << 62, 1)` returns four rows with local execution and none without, because Spark's generated code overflows. `spark.range(Long.MinValue, Long.MaxValue, 1, 1).take(3)` returns three rows against none. Main now has a native `RangeScan` (#6553) whose `mayOverflow` check keeps these ranges on Spark. Could the first PR lower `RangeScan` through `create_plan` instead of adding a second range operator, and compare the range tests against Spark's answer rather than Scala ranges? The session has a similar gap. `query_context` starts from `SessionContext::new_with_config_rt`, which registers DataFusion's default functions and none of Comet's overrides. That's harmless while `nativeOnly` keeps `ScalarFunc` out. Once admission follows `CometExecRule` trees, `concat` would resolve to DataFusion's version, which skips nulls, instead of `SparkConcat`. Could the local session come from the per-task setup (`prepare_datafusion_session_context`, or `create_execution_context` once #6695 lands)? That would also bring the spill directories and their size cap, the `spark.comet.datafusion.*` pass-through and the skip-partial-aggregation guard. Last, which budget should bound a local query? Each one gets its own 256 MiB `FairSpillPool` that Spark never sees, so concurrent local queries add up with nothing bounding the total, and the `spark.memory.offHeap.size` users already set for Comet doesn't apply. The result task is a real Spark task. Could the pool come from its `CometTaskMemoryManager`, as in the per-task path, so `spark.comet.exec.local.memoryLimit` isn't needed? -- 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]
