[SPARK-48985][CONNECT] Connect Compatible Expression Constructors #47464

hvanhovell · 2024-07-24T01:25:05Z

What changes were proposed in this pull request?

There are a number of hard coded expressions in the SparkConnectPlanner. Most of these expressions are hardcoded because they are missing a proper constructor, or because they are not registered in the FunctionRegistry. The Column API has a similar problem. This PR fixes most of these exceptions.

Why are the changes needed?

Reduce the number of hard coded expressions in the SparkConnectPlanner and the Column API. This will make it significantly easier to create an implementation agnostic Column API.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing tests.

Was this patch authored or co-authored using generative AI tooling?

No.

HyukjinKwon · 2024-07-24T01:42:09Z

cc @beliefer and @zhengruifeng

hvanhovell · 2024-07-24T01:54:06Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala

@@ -519,6 +519,8 @@ object FunctionRegistry {
    expressionBuilder("mode", ModeBuilder),
    expression[HllSketchAgg]("hll_sketch_agg"),
    expression[HllUnionAgg]("hll_union_agg"),
+    expression[Product]("product"),


The changes in this file actually increase our surface area. So I'd like to have some feedback and understand if this is ok.

I remember there was some reason to not add product here @cloud-fan

#38915 (comment)

I don't have a strong opinion now, as we are going for feature parity between SQL and DataFrame APIs recently. cc @HyukjinKwon

Actually yeah. Those methods were intentionally not added to SQL expressions.

Ok, let me try something different.

I will create a separate registry for these functions, and only use them in Dataframe resolution.

Tracked in https://issues.apache.org/jira/browse/SPARK-49004.

hvanhovell · 2024-07-24T01:54:22Z

common/utils/src/main/resources/error/error-conditions.json

@@ -2637,6 +2637,26 @@
        "message" : [
          "expects %1$, %2$ and so on, but got %0$."
        ]
+      },


Not sure if this is the proper way of doing this.

zhengruifeng · 2024-07-24T05:06:24Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala

@@ -519,6 +519,8 @@ object FunctionRegistry {
    expressionBuilder("mode", ModeBuilder),
    expression[HllSketchAgg]("hll_sketch_agg"),
    expression[HllUnionAgg]("hll_union_agg"),
+    expression[Product]("product"),


I remember there was some reason to not add product here @cloud-fan

#38915 (comment)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala

dismissed

hvanhovell · 2024-07-26T00:22:11Z

connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala

-          case other => SessionWindow(timeCol, other)
-        }
-        Some(
-          Alias(sessionWindow, "session_window")(nonInheritableMetadataKeys =


@zhengruifeng Why do you explicitly remove the Dataset.DATASET_ID_KEY and Dataset.COL_POS_KEY?

I think it is to be consistent with Spark Classic at that time

spark/sql/core/src/main/scala/org/apache/spark/sql/functions.scala

Lines 5708 to 5712 in 15c98e0

def session_window(timeColumn: Column, gapDuration: String): Column = {

withExpr {

SessionWindow(timeColumn.expr, gapDuration)

}.as("session_window")

}

as is an alias for name which remove them

spark/sql/core/src/main/scala/org/apache/spark/sql/Column.scala

Lines 1186 to 1192 in 9cf6dc8

def name(alias: String): Column = withExpr {

// SPARK-33536: an alias is no longer a column reference. Therefore,

// we should not inherit the column reference related metadata in an alias

// so that it is not caught as a column reference in DetectAmbiguousSelfJoin.

Alias(expr, alias)(

nonInheritableMetadataKeys = Seq(Dataset.DATASET_ID_KEY, Dataset.COL_POS_KEY))

}

Ah ok, that makes sense.

hvanhovell · 2024-07-26T00:24:29Z

sql/core/src/main/scala/org/apache/spark/sql/functions.scala

-    }.as("session_window")
-  }
+  def session_window(timeColumn: Column, gapDuration: String): Column =
+    session_window(timeColumn, lit(gapDuration))


This is now supported because of the small change I made to ResolveTimeWindows.

hvanhovell · 2024-07-26T00:25:27Z

sql/core/src/main/scala/org/apache/spark/sql/functions.scala

@@ -5743,7 +5728,7 @@ object functions {
   * @since 3.2.0
   */
  def session_window(timeColumn: Column, gapDuration: Column): Column =
-    Column.fn("session_window", timeColumn, gapDuration).as("session_window")


Alias is not needed. This is added by the ResolveTimeWindows rule.

hvanhovell · 2024-07-26T00:29:16Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTimeWindows.scala

-          case Literal(interval: CalendarInterval, CalendarIntervalType) =>
-            interval == null || interval.months + interval.days + interval.microseconds <= 0
-          case _ => true
+        val filterByTimeRange = if (gapDuration.foldable) {


This is a bit better than pattern matching on a literal since it supports more complex expressions, in particular Cast(Literal("<some gap duration>r", StringType), CalendarIntervalType). Normally you'd expect the constant folding to take care of this, but this happens during optimization, and this rule is part of analysis; hence the need for manual constant folding.

hvanhovell · 2024-07-28T02:16:42Z

Merging.

### What changes were proposed in this pull request? There are a number of hard coded expressions in the SparkConnectPlanner. Most of these expressions are hardcoded because they are missing a proper constructor, or because they are not registered in the FunctionRegistry. The Column API has a similar problem. This PR fixes most of these exceptions. ### Why are the changes needed? Reduce the number of hard coded expressions in the SparkConnectPlanner and the Column API. This will make it significantly easier to create an implementation agnostic Column API. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing tests. ### Was this patch authored or co-authored using generative AI tooling? No. Closes apache#47464 from hvanhovell/SPARK-48985. Authored-by: Herman van Hovell <herman@databricks.com> Signed-off-by: Herman van Hovell <herman@databricks.com>

Connect Compatible Expression Constructors

517010d

github-actions bot added SQL CONNECT labels Jul 24, 2024

Fix bucket

65545d7

Add CaseWhen

739e80e

hvanhovell requested review from zhengruifeng, MaxGekk, cloud-fan and HyukjinKwon and removed request for zhengruifeng July 24, 2024 01:47

hvanhovell commented Jul 24, 2024

View reviewed changes

HyukjinKwon previously approved these changes Jul 24, 2024

View reviewed changes

zhengruifeng reviewed Jul 24, 2024

View reviewed changes

hvanhovell added 3 commits July 25, 2024 11:07

Wip Wip Wip

25324ec

Do not expose non-public sql functions

06b5c6f

Merge remote-tracking branch 'apache/master' into SPARK-48985

a44ef5e

hvanhovell commented Jul 26, 2024

View reviewed changes

hvanhovell added 5 commits July 25, 2024 20:40

Remove unneeded IN constructor

6ba0d89

Update errors conditions

dbcdf91

Touch ups

70a9a30

Remove unneeded SessionWindow code

36e6f76

Unused imports

452a106

HyukjinKwon approved these changes Jul 28, 2024

View reviewed changes

asfgit closed this in 80223bb Jul 28, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-48985][CONNECT] Connect Compatible Expression Constructors #47464

[SPARK-48985][CONNECT] Connect Compatible Expression Constructors #47464

hvanhovell commented Jul 24, 2024

HyukjinKwon commented Jul 24, 2024

hvanhovell Jul 24, 2024

zhengruifeng Jul 24, 2024

cloud-fan Jul 24, 2024

HyukjinKwon Jul 25, 2024

hvanhovell Jul 25, 2024

hvanhovell Jul 25, 2024

hvanhovell Jul 26, 2024

hvanhovell Jul 24, 2024

zhengruifeng Jul 24, 2024

hvanhovell Jul 26, 2024 •

edited

Loading

zhengruifeng Jul 26, 2024

hvanhovell Jul 27, 2024

hvanhovell Jul 26, 2024

hvanhovell Jul 26, 2024

hvanhovell Jul 26, 2024 •

edited

Loading

hvanhovell commented Jul 28, 2024

	def session_window(timeColumn: Column, gapDuration: String): Column = {
	withExpr {
	SessionWindow(timeColumn.expr, gapDuration)
	}.as("session_window")
	}

	def name(alias: String): Column = withExpr {
	// SPARK-33536: an alias is no longer a column reference. Therefore,
	// we should not inherit the column reference related metadata in an alias
	// so that it is not caught as a column reference in DetectAmbiguousSelfJoin.
	Alias(expr, alias)(
	nonInheritableMetadataKeys = Seq(Dataset.DATASET_ID_KEY, Dataset.COL_POS_KEY))
	}

[SPARK-48985][CONNECT] Connect Compatible Expression Constructors #47464

[SPARK-48985][CONNECT] Connect Compatible Expression Constructors #47464

Conversation

hvanhovell commented Jul 24, 2024

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?

Was this patch authored or co-authored using generative AI tooling?

HyukjinKwon commented Jul 24, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hvanhovell Jul 26, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hvanhovell Jul 26, 2024 • edited Loading

Choose a reason for hiding this comment

hvanhovell commented Jul 28, 2024

hvanhovell Jul 26, 2024 •

edited

Loading

hvanhovell Jul 26, 2024 •

edited

Loading