Github user tdas commented on a diff in the pull request: https://github.com/apache/spark/pull/19271#discussion_r139830482 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala --- @@ -114,6 +115,16 @@ class IncrementalExecution( stateInfo = Some(nextStatefulOperationStateInfo), batchTimestampMs = Some(offsetSeqMetadata.batchTimestampMs), eventTimeWatermark = Some(offsetSeqMetadata.batchWatermarkMs)) + + case j @ StreamingSymmetricHashJoinExec(lKeys, rKeys, _, cond, _, _, _, left, right) => + j.copy( + stateInfo = Some(nextStatefulOperationStateInfo), --- End diff -- Whatever optimization takes place, the same optimizations will occur in EVERY batch. So if aggregation is pushed below join, then all the batches will have that. What we have to guard against is cost-based optimization that can reorder things differently in different batches. That is, why I have disabled cost-based join optimization. And when adding such optimizations in the future, we have to be cautious for the streaming case and disable them. Also, this is a general concern with other stateful ops as well, not something that this PR would address.
--- --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org