You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "zhengruifeng (via GitHub)" <gi...@apache.org> on 2023/01/25 06:32:15 UTC
[GitHub] [spark] zhengruifeng opened a new pull request, #39734: [SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns in Join
zhengruifeng opened a new pull request, #39734:
URL: https://github.com/apache/spark/pull/39734
### What changes were proposed in this pull request?
PySpark's `DataFrame.__getattr__` and `DataFrame.__getitem__` invokes `jc = self._jdf.apply(name)` in JVM, which resolve the column name and attach the dataframe id via `addDataFrameIdToCol` to handle ambiguous columns , see https://github.com/apache/spark/blob/faedcd91d554a00fc76116a0c188752cf036f907/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala#L1472-L1481
But in Connect, the output of `DataFrame.__getattr__` and `DataFrame.__getitem__` is not bound to the input `DataFrame`, it is just an `UnresolvedAttribute`.
This PR aims to fix this issue by switching to the DataFrame API based implementation.
### Why are the changes needed?
for parity
### Does this PR introduce _any_ user-facing change?
yes
### How was this patch tested?
enabled doctests and added UT
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #39734: [SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns issue in Join
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on PR #39734:
URL: https://github.com/apache/spark/pull/39734#issuecomment-1406092548
sorry, this PR can't fix chained operators like `df.join(df2, df.name == df2.name, 'outer').select(df.name, df2.height)`, it need more investigation
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #39734: [SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns in Join
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #39734:
URL: https://github.com/apache/spark/pull/39734#discussion_r1086250404
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -1174,27 +1174,63 @@ class SparkConnectPlanner(val session: SparkSession) {
}
}
+ private def splitConjunctivePredicates(condition: Expression): Seq[Expression] = {
+ condition match {
+ case UnresolvedFunction(Seq("and"), Seq(cond1, cond2), _, _, _) =>
+ splitConjunctivePredicates(cond1) ++ splitConjunctivePredicates(cond2)
+ case other => other :: Nil
+ }
+ }
+
private def transformJoin(rel: proto.Join): LogicalPlan = {
assert(rel.hasLeft && rel.hasRight, "Both join sides must be present")
if (rel.hasJoinCondition && rel.getUsingColumnsCount > 0) {
throw InvalidPlanInput(
s"Using columns or join conditions cannot be set at the same time in Join")
}
- val joinCondition =
- if (rel.hasJoinCondition) Some(transformExpression(rel.getJoinCondition)) else None
+
val catalystJointype = transformJoinType(
if (rel.getJoinType != null) rel.getJoinType else proto.Join.JoinType.JOIN_TYPE_INNER)
val joinType = if (rel.getUsingColumnsCount > 0) {
UsingJoin(catalystJointype, rel.getUsingColumnsList.asScala.toSeq)
} else {
catalystJointype
}
- logical.Join(
- left = transformRelation(rel.getLeft),
- right = transformRelation(rel.getRight),
- joinType = joinType,
- condition = joinCondition,
- hint = logical.JoinHint.NONE)
+
+ if (rel.hasJoinCondition) {
+ val leftDF = Dataset.ofRows(session, transformRelation(rel.getLeft))
+ val rightDF = Dataset.ofRows(session, transformRelation(rel.getRight))
+ val joinExprs = splitConjunctivePredicates(transformExpression(rel.getJoinCondition))
+ .map {
+ case func @ UnresolvedFunction(Seq(f), Seq(l, r), _, _, _)
+ if Seq("==", "<=>").contains(f) =>
+ val l2 = l match {
+ case UnresolvedAttribute(Seq(c)) => leftDF.apply(c).expr
+ case other => other
+ }
+ val r2 = r match {
+ case UnresolvedAttribute(Seq(c)) => rightDF.apply(c).expr
+ case other => other
+ }
+ func.copy(arguments = Seq(l2, r2))
+
+ case other => other
+ }
+ .reduce(And)
+
+ leftDF
Review Comment:
we must use `DataFrame.join` here to make sure the DataFrame ID is not changed.
https://github.com/apache/spark/blob/faedcd91d554a00fc76116a0c188752cf036f907/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala#L202-L203
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #39734: [SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns issue in Join
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #39734:
URL: https://github.com/apache/spark/pull/39734#discussion_r1086264760
##########
connector/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala:
##########
@@ -260,21 +260,21 @@ class SparkConnectPlannerSuite extends SparkFunSuite with SparkConnectPlanTest {
.addArguments(unresolvedAttribute)
.build())
- val simpleJoin = proto.Relation.newBuilder
- .setJoin(
- proto.Join.newBuilder
- .setLeft(readRel)
- .setRight(readRel)
- .setJoinType(proto.Join.JoinType.JOIN_TYPE_INNER)
- .setJoinCondition(joinCondition)
- .build())
- .build()
-
- val res = transform(simpleJoin)
- assert(res.nodeName == "Join")
- assert(res != null)
+ val e0 = intercept[AnalysisException] {
Review Comment:
after this PR, Join with `JoinCondition` will be eagerly analyzed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #39734: [SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns issue in Join
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #39734:
URL: https://github.com/apache/spark/pull/39734#discussion_r1086250404
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -1174,27 +1174,63 @@ class SparkConnectPlanner(val session: SparkSession) {
}
}
+ private def splitConjunctivePredicates(condition: Expression): Seq[Expression] = {
+ condition match {
+ case UnresolvedFunction(Seq("and"), Seq(cond1, cond2), _, _, _) =>
+ splitConjunctivePredicates(cond1) ++ splitConjunctivePredicates(cond2)
+ case other => other :: Nil
+ }
+ }
+
private def transformJoin(rel: proto.Join): LogicalPlan = {
assert(rel.hasLeft && rel.hasRight, "Both join sides must be present")
if (rel.hasJoinCondition && rel.getUsingColumnsCount > 0) {
throw InvalidPlanInput(
s"Using columns or join conditions cannot be set at the same time in Join")
}
- val joinCondition =
- if (rel.hasJoinCondition) Some(transformExpression(rel.getJoinCondition)) else None
+
val catalystJointype = transformJoinType(
if (rel.getJoinType != null) rel.getJoinType else proto.Join.JoinType.JOIN_TYPE_INNER)
val joinType = if (rel.getUsingColumnsCount > 0) {
UsingJoin(catalystJointype, rel.getUsingColumnsList.asScala.toSeq)
} else {
catalystJointype
}
- logical.Join(
- left = transformRelation(rel.getLeft),
- right = transformRelation(rel.getRight),
- joinType = joinType,
- condition = joinCondition,
- hint = logical.JoinHint.NONE)
+
+ if (rel.hasJoinCondition) {
+ val leftDF = Dataset.ofRows(session, transformRelation(rel.getLeft))
+ val rightDF = Dataset.ofRows(session, transformRelation(rel.getRight))
+ val joinExprs = splitConjunctivePredicates(transformExpression(rel.getJoinCondition))
+ .map {
+ case func @ UnresolvedFunction(Seq(f), Seq(l, r), _, _, _)
+ if Seq("==", "<=>").contains(f) =>
+ val l2 = l match {
+ case UnresolvedAttribute(Seq(c)) => leftDF.apply(c).expr
+ case other => other
+ }
+ val r2 = r match {
+ case UnresolvedAttribute(Seq(c)) => rightDF.apply(c).expr
+ case other => other
+ }
+ func.copy(arguments = Seq(l2, r2))
+
+ case other => other
+ }
+ .reduce(And)
+
+ leftDF
Review Comment:
we must use `DataFrame.join` here to make sure the DataFrame ID will not be changed.
https://github.com/apache/spark/blob/faedcd91d554a00fc76116a0c188752cf036f907/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala#L202-L203
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #39734: [WIP][SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns issue in Join
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on PR #39734:
URL: https://github.com/apache/spark/pull/39734#issuecomment-1420377688
close this in favor of https://github.com/apache/spark/pull/39925
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng closed pull request #39734: [WIP][SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns issue in Join
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng closed pull request #39734: [WIP][SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns issue in Join
URL: https://github.com/apache/spark/pull/39734
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #39734: [SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns in Join
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #39734:
URL: https://github.com/apache/spark/pull/39734#discussion_r1086248363
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -1174,27 +1174,63 @@ class SparkConnectPlanner(val session: SparkSession) {
}
}
+ private def splitConjunctivePredicates(condition: Expression): Seq[Expression] = {
+ condition match {
+ case UnresolvedFunction(Seq("and"), Seq(cond1, cond2), _, _, _) =>
+ splitConjunctivePredicates(cond1) ++ splitConjunctivePredicates(cond2)
+ case other => other :: Nil
+ }
+ }
+
private def transformJoin(rel: proto.Join): LogicalPlan = {
assert(rel.hasLeft && rel.hasRight, "Both join sides must be present")
if (rel.hasJoinCondition && rel.getUsingColumnsCount > 0) {
throw InvalidPlanInput(
s"Using columns or join conditions cannot be set at the same time in Join")
}
- val joinCondition =
- if (rel.hasJoinCondition) Some(transformExpression(rel.getJoinCondition)) else None
+
val catalystJointype = transformJoinType(
if (rel.getJoinType != null) rel.getJoinType else proto.Join.JoinType.JOIN_TYPE_INNER)
val joinType = if (rel.getUsingColumnsCount > 0) {
UsingJoin(catalystJointype, rel.getUsingColumnsList.asScala.toSeq)
} else {
catalystJointype
}
- logical.Join(
- left = transformRelation(rel.getLeft),
- right = transformRelation(rel.getRight),
- joinType = joinType,
- condition = joinCondition,
- hint = logical.JoinHint.NONE)
+
+ if (rel.hasJoinCondition) {
+ val leftDF = Dataset.ofRows(session, transformRelation(rel.getLeft))
+ val rightDF = Dataset.ofRows(session, transformRelation(rel.getRight))
+ val joinExprs = splitConjunctivePredicates(transformExpression(rel.getJoinCondition))
+ .map {
+ case func @ UnresolvedFunction(Seq(f), Seq(l, r), _, _, _)
+ if Seq("==", "<=>").contains(f) =>
Review Comment:
currently, both `a == a` and `a <=> a` are taken into account
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #39734: [WIP][SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns issue in Join
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #39734:
URL: https://github.com/apache/spark/pull/39734#issuecomment-1407684513
unless we introduce expr id in connect, I don't think we can solve this problem.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #39734: [SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns in Join
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #39734:
URL: https://github.com/apache/spark/pull/39734#discussion_r1086248363
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -1174,27 +1174,63 @@ class SparkConnectPlanner(val session: SparkSession) {
}
}
+ private def splitConjunctivePredicates(condition: Expression): Seq[Expression] = {
+ condition match {
+ case UnresolvedFunction(Seq("and"), Seq(cond1, cond2), _, _, _) =>
+ splitConjunctivePredicates(cond1) ++ splitConjunctivePredicates(cond2)
+ case other => other :: Nil
+ }
+ }
+
private def transformJoin(rel: proto.Join): LogicalPlan = {
assert(rel.hasLeft && rel.hasRight, "Both join sides must be present")
if (rel.hasJoinCondition && rel.getUsingColumnsCount > 0) {
throw InvalidPlanInput(
s"Using columns or join conditions cannot be set at the same time in Join")
}
- val joinCondition =
- if (rel.hasJoinCondition) Some(transformExpression(rel.getJoinCondition)) else None
+
val catalystJointype = transformJoinType(
if (rel.getJoinType != null) rel.getJoinType else proto.Join.JoinType.JOIN_TYPE_INNER)
val joinType = if (rel.getUsingColumnsCount > 0) {
UsingJoin(catalystJointype, rel.getUsingColumnsList.asScala.toSeq)
} else {
catalystJointype
}
- logical.Join(
- left = transformRelation(rel.getLeft),
- right = transformRelation(rel.getRight),
- joinType = joinType,
- condition = joinCondition,
- hint = logical.JoinHint.NONE)
+
+ if (rel.hasJoinCondition) {
+ val leftDF = Dataset.ofRows(session, transformRelation(rel.getLeft))
+ val rightDF = Dataset.ofRows(session, transformRelation(rel.getRight))
+ val joinExprs = splitConjunctivePredicates(transformExpression(rel.getJoinCondition))
+ .map {
+ case func @ UnresolvedFunction(Seq(f), Seq(l, r), _, _, _)
+ if Seq("==", "<=>").contains(f) =>
Review Comment:
currently, only `a == a` and `a <=> a` are taken into account
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #39734: [SPARK-41812][SPARK-41823][CONNECT][PYTHON] Fix ambiguous columns issue in Join
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on PR #39734:
URL: https://github.com/apache/spark/pull/39734#issuecomment-1406012031
@HyukjinKwon @cloud-fan @grundprinzip
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org