[SPARK-44341][SQL][PYTHON] Define the computing logic through PartitionEvaluator API and use it in WindowExec and WindowInPandasExec #41939

beliefer · 2023-07-11T11:24:11Z

What changes were proposed in this pull request?

WindowExec and WindowInPandasExec are updated to use the PartitionEvaluator API to do execution.

Why are the changes needed?

To define the computing logic and requires the caller side to explicitly list what needs to be serialized and sent to executors

Does this PR introduce any user-facing change?

'No'.
Just update the inner implementation.

How was this patch tested?

Add new test cases.

beliefer · 2023-07-12T06:02:40Z

ping @cloud-fan @HyukjinKwon @zhengruifeng @viirya cc @vinodkc
The CI failure looks unrelated.

cloud-fan · 2023-07-14T02:18:15Z

sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala

+    val windowExpression: Seq[NamedExpression],
+    val partitionSpec: Seq[Expression],
+    val orderSpec: Seq[SortOrder],
+    val conf: SQLConf,


we usually don't need to pass the SQLConf, as worker side can get it via SQLConf.get

cloud-fan · 2023-07-14T02:20:50Z

We can probably skip testing it. Overall it's just a refactor and it's probably too much to run many tests twice. We can enable it by default later so that all tests cover this code path.

viirya · 2023-07-14T04:56:38Z

sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala

+  def windowExpression: Seq[NamedExpression]
+  def partitionSpec: Seq[Expression]
+  def orderSpec: Seq[SortOrder]
+  def conf: SQLConf


Seems conf is always SQLConf.get in both implementations?

Yes. I have used SQLConf.get in both implementations.

viirya

Refactored code looks okay to me.

viirya · 2023-07-14T05:05:13Z

As tests are skipped now, maybe you can enable it to test change in CI, and revert back current config before merging.

beliefer · 2023-07-14T05:06:07Z

As tests are skipped now, maybe you can enable it to test change in CI, and revert back current config before merging.

In fact, CI passed in previous commit.

viirya · 2023-07-14T05:12:29Z

As tests are skipped now, maybe you can enable it to test change in CI, and revert back current config before merging.

In fact, CI passed in previous commit.

Oh okay. 👍

beliefer · 2023-07-14T08:46:35Z

The CI failure is unrelated.

cloud-fan · 2023-07-17T05:40:10Z

...re/src/main/scala/org/apache/spark/sql/execution/python/WindowInPandasEvaluatorFactory.scala

+  }
+
+  // Unwrap the expressions and factories from the map.
+  private val expressionsWithFrameIndex =


so all these private vals will be serialized to the executor side, right? I think it's better to define them in the evaluator class, not the factor, to reduce the data to be serialized.

cc @Hisoka-X can you also check your PRs (either merged or not)? We should avoid putting variables in the factor class, but define them in the evaluator class instead.

I haven't come across a factory that needs to define member variables (except for parameters)😂

…onEvaluator API and use it in WindowExec and WindowInPandasExec

cloud-fan · 2023-07-18T03:36:53Z

thanks, merging to master!

beliefer · 2023-07-18T03:42:55Z

@cloud-fan @viirya @Hisoka-X Thank you.

cloud-fan · 2023-07-21T03:39:11Z

sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala

+import org.apache.spark.sql.types.{CalendarIntervalType, DateType, DayTimeIntervalType, DecimalType, IntegerType, TimestampNTZType, TimestampType, YearMonthIntervalType}
+import org.apache.spark.util.collection.Utils
+
+trait WindowEvaluatorFactoryBase {


with hindsight, shall we move it to a new file? Its two sub-classes are not both in this file, so it's a bit weird to put it with one of the sub-class.

OK. Let's move it.

…torFactoryBase to a single file ### What changes were proposed in this pull request? #41939 defined the computing logic through PartitionEvaluator API and use it in `WindowExec` and `WindowInPandasExec`. According to the comment #41939 (comment), this PR want move the base trait `WindowEvaluatorFactoryBase` to a single file. ### Why are the changes needed? Improve the code. ### Does this PR introduce _any_ user-facing change? 'No'. Just update inner implementation. ### How was this patch tested? N/A Closes #42106 from beliefer/SPARK-44341_followup. Authored-by: Jiaan Geng <beliefer@163.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

vinodkc · 2023-07-27T17:27:48Z

sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowExec.scala

+    } else {
+      child.execute().mapPartitions { iter =>
+        val evaluator = evaluatorFactory.createEvaluator()
+        evaluator.eval(0, iter)


@beliefer, Can you please raise a follow-up PR to handle the partition index as this #42185

Thank you for the reminder. I will take a look!

… correctly for WindowGroupLimitExec,WindowExec and WindowInPandasExec ### What changes were proposed in this pull request? This is a followup of #41899 and #41939, to set the partition index correctly even if it's not used for now. It also contains a few code cleanup. ### Why are the changes needed? future-proof ### Does this PR introduce _any_ user-facing change? 'No'. ### How was this patch tested? existing tests Closes #42208 from beliefer/SPARK-44340_followup. Authored-by: Jiaan Geng <beliefer@163.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

…onEvaluator API and use it in WindowExec and WindowInPandasExec ### What changes were proposed in this pull request? `WindowExec` and `WindowInPandasExec` are updated to use the `PartitionEvaluator` API to do execution. ### Why are the changes needed? To define the computing logic and requires the caller side to explicitly list what needs to be serialized and sent to executors ### Does this PR introduce _any_ user-facing change? 'No'. Just update the inner implementation. ### How was this patch tested? Add new test cases. Closes apache#41939 from beliefer/SPARK-44341. Authored-by: Jiaan Geng <beliefer@163.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

…torFactoryBase to a single file ### What changes were proposed in this pull request? apache#41939 defined the computing logic through PartitionEvaluator API and use it in `WindowExec` and `WindowInPandasExec`. According to the comment apache#41939 (comment), this PR want move the base trait `WindowEvaluatorFactoryBase` to a single file. ### Why are the changes needed? Improve the code. ### Does this PR introduce _any_ user-facing change? 'No'. Just update inner implementation. ### How was this patch tested? N/A Closes apache#42106 from beliefer/SPARK-44341_followup. Authored-by: Jiaan Geng <beliefer@163.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

… correctly for WindowGroupLimitExec,WindowExec and WindowInPandasExec ### What changes were proposed in this pull request? This is a followup of apache#41899 and apache#41939, to set the partition index correctly even if it's not used for now. It also contains a few code cleanup. ### Why are the changes needed? future-proof ### Does this PR introduce _any_ user-facing change? 'No'. ### How was this patch tested? existing tests Closes apache#42208 from beliefer/SPARK-44340_followup. Authored-by: Jiaan Geng <beliefer@163.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

github-actions bot added SQL CORE PYTHON PANDAS API ON SPARK labels Jul 11, 2023

beliefer force-pushed the SPARK-44341 branch from 184fbd2 to cc0ab1c Compare July 13, 2023 04:49

cloud-fan reviewed Jul 14, 2023

View reviewed changes

github-actions bot removed CORE PANDAS API ON SPARK labels Jul 14, 2023

viirya reviewed Jul 14, 2023

View reviewed changes

viirya approved these changes Jul 14, 2023

View reviewed changes

beliefer force-pushed the SPARK-44341 branch from 332d408 to 68ef2b2 Compare July 16, 2023 07:14

cloud-fan reviewed Jul 17, 2023

View reviewed changes

beliefer added 5 commits July 17, 2023 20:12

[SPARK-44341][SQL][PYTHON] Define the computing logic through Partiti…

075c53d

…onEvaluator API and use it in WindowExec and WindowInPandasExec

Update code

a5f59e5

Update code

f465e84

Update code

8a42a38

Update code

ad88985

beliefer force-pushed the SPARK-44341 branch from d4430ed to ad88985 Compare July 17, 2023 12:19

cloud-fan approved these changes Jul 18, 2023

View reviewed changes

cloud-fan closed this in 95dc829 Jul 18, 2023

cloud-fan reviewed Jul 21, 2023

View reviewed changes

beliefer mentioned this pull request Jul 21, 2023

[SPARK-44341][SQL][PYTHON][FOLLOWUP] Move the base trait WindowEvaluatorFactoryBase to a single file #42106

Closed

vinodkc reviewed Jul 27, 2023

View reviewed changes

beliefer mentioned this pull request Jul 28, 2023

[SPARK-44340][SPARK-44341][SQL][PYTHON][FOLLOWUP] Set partition index correctly for WindowGroupLimitExec,WindowExec and WindowInPandasExec #42208

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-44341][SQL][PYTHON] Define the computing logic through PartitionEvaluator API and use it in WindowExec and WindowInPandasExec #41939

[SPARK-44341][SQL][PYTHON] Define the computing logic through PartitionEvaluator API and use it in WindowExec and WindowInPandasExec #41939

beliefer commented Jul 11, 2023 •

edited

Loading

beliefer commented Jul 12, 2023 •

edited

Loading

cloud-fan Jul 14, 2023

cloud-fan commented Jul 14, 2023

viirya Jul 14, 2023

beliefer Jul 14, 2023

viirya left a comment

viirya commented Jul 14, 2023

beliefer commented Jul 14, 2023

viirya commented Jul 14, 2023

beliefer commented Jul 14, 2023

cloud-fan Jul 17, 2023

cloud-fan Jul 17, 2023

Hisoka-X Jul 17, 2023

cloud-fan commented Jul 18, 2023

beliefer commented Jul 18, 2023

cloud-fan Jul 21, 2023

beliefer Jul 21, 2023

vinodkc Jul 27, 2023 •

edited

Loading

beliefer Jul 28, 2023

[SPARK-44341][SQL][PYTHON] Define the computing logic through PartitionEvaluator API and use it in WindowExec and WindowInPandasExec #41939

[SPARK-44341][SQL][PYTHON] Define the computing logic through PartitionEvaluator API and use it in WindowExec and WindowInPandasExec #41939

Conversation

beliefer commented Jul 11, 2023 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

beliefer commented Jul 12, 2023 • edited Loading

Choose a reason for hiding this comment

cloud-fan commented Jul 14, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya left a comment

Choose a reason for hiding this comment

viirya commented Jul 14, 2023

beliefer commented Jul 14, 2023

viirya commented Jul 14, 2023

beliefer commented Jul 14, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Jul 18, 2023

beliefer commented Jul 18, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vinodkc Jul 27, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

beliefer commented Jul 11, 2023 •

edited

Loading

beliefer commented Jul 12, 2023 •

edited

Loading

vinodkc Jul 27, 2023 •

edited

Loading