comphead commented on code in PR #7256:
URL: https://github.com/apache/datafusion-comet/pull/7256#discussion_r4429656599
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala:
##########
@@ -81,10 +88,45 @@ object CometWindowExec extends
CometOperatorSerde[WindowExec] {
if (windowExprProto.forall(_.isDefined) &&
partitionExprs.forall(_.isDefined)
&& sortOrders.forall(_.isDefined)) {
+ // Spark plans no sort below the window when its input already reports
the order the window
+ // needs, but rows from the JVM arrive through a ScanExec that declares
none. Send the
+ // partition keys' order, the only one the native window checks, for the
planner to declare
+ // (see SortedInputExec). When the order may be false, nothing native
can declare it, so the
+ // window stays in Spark, unless the doubt is a Comet join in the same
native plan, which
+ // DataFusion sees itself. A key Comet cannot express, or a collated
string key, sends the
+ // window back to Spark too.
+ val requiredOrdering = op.requiredChildOrdering.head
+ val inputSortOrders =
+ if (op.partitionSpec.nonEmpty &&
+ SortOrder.orderingSatisfies(op.child.outputOrdering,
requiredOrdering)) {
+ // The walk that counts Comet joins in the same native plan flags
everything the other
+ // one does, so the other runs only when this one flags the order.
+ if (!CometHashJoinExec.mayReportFalseInputOrder(op.child,
countInPlan = true)) {
+ val partitionKeys = requiredOrdering.take(op.partitionSpec.length)
+ val orders = partitionKeys.flatMap { order =>
+ if (hasNonDefaultStringCollation(order.dataType)) None
+ else exprToProto(order, op.child.output)
+ }
+ if (orders.length < partitionKeys.length) {
+ withFallbackReason(op, inexpressibleInputOrderReason)
+ return None
+ }
+ orders
+ } else if (CometHashJoinExec.mayReportFalseInputOrder(op.child,
countInPlan = false)) {
+ withFallbackReason(op, uncheckedInputOrderReason)
+ return None
Review Comment:
Thanks, agreed. Falling back hands the same out-of-order rows to Spark's
window, which trusts the order as much as the native one does. 136e5cf53f
removes the fallback. When the order may be false, the window sends no order,
so the native window rejects its input with the ordering error, as on `main`
and the previous revision, instead of returning wrong rows.
Restoring the order needs either a nested loop join that keeps it or a sort
below the window. #6785 does both: it keeps a nested loop join over an ordered
streamed side in Spark (#6898), and it sorts a window's input natively when the
partition keys' order is not declared.
From reading the code, a native `WindowGroupLimit` between the join and the
window gives wrong rows on `main` already, since `PartitionedRankLimitExec`
declares the partition and order keys' order itself. Keeping the join in Spark
covers that shape too.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala:
##########
@@ -81,10 +88,45 @@ object CometWindowExec extends
CometOperatorSerde[WindowExec] {
if (windowExprProto.forall(_.isDefined) &&
partitionExprs.forall(_.isDefined)
&& sortOrders.forall(_.isDefined)) {
+ // Spark plans no sort below the window when its input already reports
the order the window
+ // needs, but rows from the JVM arrive through a ScanExec that declares
none. Send the
+ // partition keys' order, the only one the native window checks, for the
planner to declare
+ // (see SortedInputExec). When the order may be false, nothing native
can declare it, so the
+ // window stays in Spark, unless the doubt is a Comet join in the same
native plan, which
+ // DataFusion sees itself. A key Comet cannot express, or a collated
string key, sends the
+ // window back to Spark too.
+ val requiredOrdering = op.requiredChildOrdering.head
+ val inputSortOrders =
+ if (op.partitionSpec.nonEmpty &&
+ SortOrder.orderingSatisfies(op.child.outputOrdering,
requiredOrdering)) {
+ // The walk that counts Comet joins in the same native plan flags
everything the other
+ // one does, so the other runs only when this one flags the order.
+ if (!CometHashJoinExec.mayReportFalseInputOrder(op.child,
countInPlan = true)) {
+ val partitionKeys = requiredOrdering.take(op.partitionSpec.length)
Review Comment:
Thanks. 136e5cf53f checks every partition and order key for a collated
string before anything is converted, so such a window stays in Spark whatever
its input is. The native window finds partitions and peers by comparing keys
byte by byte. The check now also covers a window with no `PARTITION BY`, which
never reached the old one. The declaration still holds only the partition keys.
`a window over a cache sorted by a collated string key stays in Spark` now
also runs your shape: `dense_rank() OVER (PARTITION BY n ORDER BY k)` over a
cache sorted by an integer `n` and a `UTF8_LCASE` `k`. It checks Spark's ranks
and the fallback reason.
##########
native/core/src/execution/planner/join.rs:
##########
@@ -402,48 +403,59 @@ impl PhysicalPlanner {
Ok(Arc::new(prepared))
}
- /// `plan`, declaring `sort_orders` as the order its rows arrive in,
unless it already
- /// declares one. Spark's plan knows the order of an input that enters the
native plan from
- /// the JVM, and of a scan of sorted files, but the `ScanExec` or the file
scan the rows
- /// arrive through declares none. Only that order is declared: when an
operator in between may
- /// have moved rows, `plan` is left as it is.
+ /// `plan`, declaring `sort_orders` as the order its rows arrive in,
unless the order it
+ /// declares already satisfies them. Spark's plan knows the order of an
input that enters the
+ /// native plan from the JVM, and of a scan of sorted files, but the
`ScanExec` or the file
+ /// scan the rows arrive through declares none. Only that order is
declared: when an operator
+ /// in between may have moved rows, `plan` is left as it is.
///
- /// The serde sends an order only when the streamed side's Comet operators
report it, and it
- /// checks that nothing under them moves rows (`mayReportFalseOrder`),
while this walk asks
- /// the native operators the same thing through `maintains_input_order`.
The two agree as
- /// long as a Comet operator that reports an order keeps it natively and
says so, and a Comet
- /// leaf that reports one is a source this walk knows or declares its
order itself, as the
- /// native range does. One that does not lands here with `plan` unchanged,
and the join then
- /// moves its unmatched rows after Spark has dropped the sort above it.
Nothing here can fall
- /// back at that point, so the invariant is kept at the operator: see
"Adding a New Operator"
- /// in the contributor guide.
- fn with_declared_order(
+ /// An input that declares only part of `sort_orders`, such as a window
over another window,
+ /// gets them declared next to its own order.
+ ///
+ /// The serde sends an order only when the Comet operators of a join's
streamed side or of a
+ /// window's input report it, and it checks that nothing under them moves
rows
+ /// (`mayReportFalseOrder`), while this walk asks the native operators the
same thing through
+ /// `maintains_input_order`. The two agree as long as a Comet operator
that reports an order
+ /// keeps it natively and says so, and a Comet leaf that reports one is a
source this walk
+ /// knows or declares its order itself, as the native range does. One that
does not lands
+ /// here with `plan` unchanged, and the join then moves its unmatched rows
after Spark has
+ /// dropped the sort above it, or the window rejects its input. Nothing
here can fall back at
+ /// that point, so the invariant is kept at the operator: see "Adding a
New Operator" in the
+ /// contributor guide.
+ pub(super) fn with_declared_order(
&self,
plan: Arc<dyn ExecutionPlan>,
sort_orders: &[Expr],
) -> Result<Arc<dyn ExecutionPlan>, ExecutionError> {
- if plan.output_ordering().is_some() ||
!Self::keeps_source_order(&plan) {
+ if sort_orders.is_empty() || !Self::keeps_source_order(&plan) {
Review Comment:
Thanks. `keeps_source_order` now accepts a right outer join whose probe
declares an order. That is exactly when DataFusion keeps the join's unmatched
probe rows in place, since `HashJoinExec::execute` hands
`right.output_ordering().is_some()` to the stream. So the window's order is
declared next to the join's, and your query runs. A right outer join whose
probe declares nothing is still rejected. Fixed in 136e5cf53f.
`hash_join_declares_streamed_order_on_probe_input` in `planner.rs` now also
checks the rule for both outer join shapes. The cache test `a native window
declares its order next to the order its input declares` now runs your shape: a
left outer join with a one-row broadcast table on `k`, and `sum(v) OVER
(PARTITION BY a)` above it, where `a` is an alias of `k`. A residual condition
leaves unmatched rows among the matched ones of key 0, so the query would split
that key if the join moved them. It checks the rows, that the window sends `a`,
and that no sort is planned.
--
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]
