[SPARK-43321][Connect] Dataset#Joinwith #40997

zhenlineo · 2023-04-28T23:23:34Z

What changes were proposed in this pull request?

Impl missing method JoinWith with Join relation operation

The JoinWith adds left and right struct type info in the Join relation proto.

Why are the changes needed?

Missing Dataset API

Does this PR introduce any user-facing change?

Yes. Added the missing Dataset#JoinWith method

How was this patch tested?

E2E tests.

hvanhovell · 2023-05-01T17:01:26Z

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala

+        .setRight(other.plan.getRoot)
+        .setJoinType(joinTypeValue)
+        .setJoinCondition(condition.expr)
+        .setLeftSchema(DataTypeProtoConverter.toConnectProtoType(this.schema))


Is there a way to avoid the RPCs here? The input schema(s) can actually change between the schema call and the actual materialization of the plan. IMO it would be better to state in the proto we want the join to produce a different result shape, instead of explicitly sending the schema.

We need to pass one info to the server: whether the top level of the encoder is a struct or not. I feel it might look more strange to pass a boolean here isTopLevelStruct?

Yeah, but passing the encoders only tells the server that a different output schema is expected, i.e. struct(left.*), struct(right.*) instead of left.*, right.*. There is a precedent in SQL called NATURAL JOIN. I would probably add a field to the Join message called leftAndRightAsStruct or something in that vain.

Not 100% sure I understand. Looked into the NATURAL JOIN, we computed the joinResultStruct directly. I do not see an easy way to do the same on the client, so I changed the code to pass struct(left.*, right.*) now.

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala

hvanhovell · 2023-05-01T20:25:13Z

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala

+        throw new IllegalArgumentException(s"Unsupported join type `joinType`: $e")
+    }
+
+    val tupleEncoder = ProductEncoder


This opens up an interesting can of Worms. The problem is basically that we don't know the encoder for a Dataframe. If you joinWith a DataFrame (Dataset[Row]) with a Dataset[E] you will end up with a Dataset[(Row, E)], the problem is that we don't know the schema for Row and we infer this from the schema in SparkResult, this does not work in the joinWith scenario...

The worm can is sealed. I added some extra logic in SparkResult to convert each UnboundRowEncoder to a new encoder inferred from the row schema.

The can of worms is only closed if the UnboundRowEncoders are only nested in Products/Rows :)... Fine for now though.

...ctor/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/SparkResult.scala

hvanhovell · 2023-06-28T00:07:11Z

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -252,6 +252,18 @@ class Dataset[T] private[sql] (
    }
  }

+  private def containsUnboudRowEncoder(enc: AgnosticEncoder[_]): Boolean = {


Can we address other nested structures in a follow-up? Granted this covers 90% of the cases. If we end up having a lot of communality with the code in SparkResult, then I'd suggest moving this into a single place.

Added SPARK-44228 to address the nested row encoders all together.

hvanhovell · 2023-06-28T00:08:36Z

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala

+    }
+
+    def isRowStruct(enc: AgnosticEncoder[_]): Boolean =
+      enc.dataType.isInstanceOf[StructType] && !enc.isInstanceOf[OptionEncoder[_]]


Can we just have a parent encoder for all struct encoders, and check for that? I feel that this is overly complex.

hvanhovell · 2023-06-28T00:13:31Z

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala

+      true
+    } else {
+      enc match {
+        case ProductEncoder(clsTag, fields) if ProductEncoder.isTuple(clsTag) =>


In theory any 'struct' encoder can contain an UnboundRowEncoder. This only deals with the stuff we can currently create in scala client. That is fine, but then we should document this limitation somewhere.

hvanhovell · 2023-06-28T00:15:17Z

...ctor/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/SparkResult.scala

+      case ProductEncoder(clsTag, fields) if ProductEncoder.isTuple(clsTag) =>
+        // Recursively continue updating the tuple product encoder
+        val schema = dataType.asInstanceOf[StructType]
+        assert(fields.length == schema.fields.length)


I don't think this needs to hold. I think fields.length <= schema.fields.length needs to hold.

hvanhovell · 2023-06-28T00:16:25Z

connector/connect/common/src/main/protobuf/spark/connect/relations.proto

+
+  // Optional. set to true if the left is a struct that can be flattened to a row. Only used by
+  // JoinWith where the left is a dataset.
+  optional bool is_left_row_struct = 6;


Can we just make this one flag?

hvanhovell · 2023-06-28T00:18:10Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/object.scala

@@ -719,3 +720,99 @@ case class CoGroup(
  override protected def withNewChildrenInternal(
      newLeft: LogicalPlan, newRight: LogicalPlan): CoGroup = copy(left = newLeft, right = newRight)
 }
+
+object JoinWith {


Can you file an follow-up so we can move this into the analyzer?

Added SPARK-44225

hvanhovell · 2023-07-06T21:16:01Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/AgnosticEncoder.scala

-      fields: Seq[EncoderField])
-    extends AgnosticEncoder[K] {
+  // Contains a sequence of fields. The fields can be flattened to columns in a row.
+  trait FieldsEncoder[K] extends AgnosticEncoder[K] {


StructEncoder?

hvanhovell · 2023-07-06T21:29:58Z

connector/connect/common/src/main/protobuf/spark/connect/relations.proto

+  // (Optional) Only used by joinWith. Set the left and right join data types.
+  optional JoinDataType join_data_type = 6;
+
+  message JoinDataType {


This is probably the right thing to do for the wrong reason. The problem that you seem to be solving is dealing with structs, you need to know when you should wrap the input in a struct. However this could easily be solved by looking at the left & right input schemas (if more than 1 column make this a struct).

The problem that you are actually solving is that when you have applied a single value encoder to a dataset with multiple columns. The single encoder will bind to the first column. JoinWith should not project the remaining columns. This is not something we can infer from the serverside schema.

hvanhovell · 2023-07-06T21:40:03Z

connector/connect/common/src/main/protobuf/spark/connect/relations.proto

+
+  message JoinDataType {
+    // If the left data type is a struct that can be flatten to a row.
+    bool is_left_flattenable_to_row = 1;


i'd just name this is_left_struct or is_left_single_value. Flattening is a detail we don't really care about.

The same applies to the right side.

hvanhovell

LGTM

hvanhovell · 2023-07-06T21:41:04Z

Merging. Please address the naming issues in a follow-up.

…Client joinWith ### What changes were proposed in this pull request? Directly use `struct` in the names to directly call out the underlying encoder is composed using a struct encoder. ### Why are the changes needed? Addressing PR review feedbacks of #40997 ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests Closes #41971 from zhenlineo/followup-joinwith. Authored-by: Zhen Li <zhenlineo@users.noreply.github.com> Signed-off-by: Kent Yao <yao@apache.org>

### What changes were proposed in this pull request? Impl missing method JoinWith with Join relation operation The JoinWith adds `left` and `right` struct type info in the Join relation proto. ### Why are the changes needed? Missing Dataset API ### Does this PR introduce _any_ user-facing change? Yes. Added the missing Dataset#JoinWith method ### How was this patch tested? E2E tests. Closes apache#40997 from zhenlineo/joinwith. Authored-by: Zhen Li <zhenlineo@users.noreply.github.com> Signed-off-by: Herman van Hovell <herman@databricks.com>

…Client joinWith ### What changes were proposed in this pull request? Directly use `struct` in the names to directly call out the underlying encoder is composed using a struct encoder. ### Why are the changes needed? Addressing PR review feedbacks of apache#40997 ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests Closes apache#41971 from zhenlineo/followup-joinwith. Authored-by: Zhen Li <zhenlineo@users.noreply.github.com> Signed-off-by: Kent Yao <yao@apache.org>

github-actions bot added CONNECT CORE PYTHON SQL labels Apr 28, 2023

zhenlineo force-pushed the joinwith branch 4 times, most recently from 1e92fe0 to 68b75d3 Compare May 1, 2023 15:47

hvanhovell reviewed May 1, 2023

View reviewed changes

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala Outdated Show resolved Hide resolved

hvanhovell reviewed May 1, 2023

View reviewed changes

zhenlineo force-pushed the joinwith branch 5 times, most recently from 98f47ab to 0e2079f Compare May 2, 2023 18:46

hvanhovell reviewed May 3, 2023

View reviewed changes

...ctor/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/SparkResult.scala Outdated Show resolved Hide resolved

zhenlineo force-pushed the joinwith branch 5 times, most recently from 578422b to 5496428 Compare May 11, 2023 14:36

zhenlineo force-pushed the joinwith branch from 5496428 to 4859dcc Compare June 5, 2023 17:35

zhenlineo force-pushed the joinwith branch from 4859dcc to 819f22d Compare June 27, 2023 20:53

hvanhovell reviewed Jun 28, 2023

View reviewed changes

zhenlineo force-pushed the joinwith branch 5 times, most recently from 34c5a67 to fa23dfc Compare June 29, 2023 21:37

zhenlineo added 9 commits July 5, 2023 13:18

JoinWith

317318b

WIP

b5d3b5f

Add recursive check for product encoder in SparkResult

263080d

WIP: How to send isNonNullableStruct info to the server via schema?

da0202e

Use a boolean directly

224368b

fix format

a6c3218

address feedbacks

2585ff8

Abstract the FieldsEncoder

8491f33

fix python genereated code conflicts

b7d7515

zhenlineo force-pushed the joinwith branch from fa23dfc to b7d7515 Compare July 5, 2023 20:37

Cleanup after rebase

df19299

zhenlineo force-pushed the joinwith branch from 1b117ab to df19299 Compare July 5, 2023 20:45

hvanhovell reviewed Jul 6, 2023

View reviewed changes

hvanhovell approved these changes Jul 6, 2023

View reviewed changes

hvanhovell closed this in 9c7978d Jul 6, 2023

zhenlineo mentioned this pull request Jul 12, 2023

[SPARK-43321][Connect][Followup] Better names for APIs used in Scala Client joinWith #41971

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-43321][Connect] Dataset#Joinwith #40997

[SPARK-43321][Connect] Dataset#Joinwith #40997

zhenlineo commented Apr 28, 2023 •

edited

hvanhovell May 1, 2023

zhenlineo May 1, 2023

hvanhovell May 1, 2023

zhenlineo May 2, 2023 •

edited

hvanhovell May 1, 2023

zhenlineo May 2, 2023 •

edited

hvanhovell Jun 28, 2023

hvanhovell Jun 28, 2023

zhenlineo Jun 28, 2023

hvanhovell Jun 28, 2023

hvanhovell Jun 28, 2023 •

edited

hvanhovell Jun 28, 2023

hvanhovell Jun 28, 2023 •

edited

hvanhovell Jun 28, 2023

zhenlineo Jun 28, 2023

hvanhovell Jul 6, 2023

hvanhovell Jul 6, 2023

hvanhovell Jul 6, 2023

hvanhovell left a comment

hvanhovell commented Jul 6, 2023

[SPARK-43321][Connect] Dataset#Joinwith #40997

[SPARK-43321][Connect] Dataset#Joinwith #40997

Conversation

zhenlineo commented Apr 28, 2023 • edited

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?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhenlineo May 2, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhenlineo May 2, 2023 • edited

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 Jun 28, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hvanhovell Jun 28, 2023 • edited

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 left a comment

Choose a reason for hiding this comment

hvanhovell commented Jul 6, 2023

zhenlineo commented Apr 28, 2023 •

edited

zhenlineo May 2, 2023 •

edited

zhenlineo May 2, 2023 •

edited

hvanhovell Jun 28, 2023 •

edited

hvanhovell Jun 28, 2023 •

edited