[GitHub] [spark] cloud-fan commented on a change in pull request #28975: [SPARK-32148][SS] Fix stream-stream join issue on missing to copy reused unsafe row
cloud-fan commented on a change in pull request #28975:
URL: https://github.com/apache/spark/pull/28975#discussion_r451357986
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManager.scala
##
@@ -451,10 +451,25 @@ class SymmetricHashJoinStateManager(
}
private trait KeyWithIndexToValueRowConverter {
+/** Defines the schema of the value row (the value side of K-V in state
store). */
def valueAttributes: Seq[Attribute]
+/**
+ * Convert the value row to (actual value, match) pair.
+ *
+ * NOTE: implementations should ensure the result row is NOT reused during
execution, as
+ * caller may use the value to store without copy().
Review comment:
SGTM. I was referring to
https://github.com/apache/spark/pull/28975#discussion_r450016555
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #28975: [SPARK-32148][SS] Fix stream-stream join issue on missing to copy reused unsafe row
cloud-fan commented on a change in pull request #28975:
URL: https://github.com/apache/spark/pull/28975#discussion_r451332628
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManager.scala
##
@@ -451,10 +451,25 @@ class SymmetricHashJoinStateManager(
}
private trait KeyWithIndexToValueRowConverter {
+/** Defines the schema of the value row (the value side of K-V in state
store). */
def valueAttributes: Seq[Attribute]
+/**
+ * Convert the value row to (actual value, match) pair.
+ *
+ * NOTE: implementations should ensure the result row is NOT reused during
execution, as
+ * caller may use the value to store without copy().
Review comment:
we need to update it. It's not because of storing, but the caller side
updates it.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #28975: [SPARK-32148][SS] Fix stream-stream join issue on missing to copy reused unsafe row
cloud-fan commented on a change in pull request #28975:
URL: https://github.com/apache/spark/pull/28975#discussion_r451332628
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManager.scala
##
@@ -451,10 +451,25 @@ class SymmetricHashJoinStateManager(
}
private trait KeyWithIndexToValueRowConverter {
+/** Defines the schema of the value row (the value side of K-V in state
store). */
def valueAttributes: Seq[Attribute]
+/**
+ * Convert the value row to (actual value, match) pair.
+ *
+ * NOTE: implementations should ensure the result row is NOT reused during
execution, as
+ * caller may use the value to store without copy().
Review comment:
we need to update the comment. It's not because of storing, but the
caller side updates the row.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #28975: [SPARK-32148][SS] Fix stream-stream join issue on missing to copy reused unsafe row
cloud-fan commented on a change in pull request #28975: URL: https://github.com/apache/spark/pull/28975#discussion_r450086968 ## File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManager.scala ## @@ -259,6 +269,9 @@ class SymmetricHashJoinStateManager( return null } +// Make a copy on value row, as below cleanup logic may update the value row silently. +currentValue = currentValue.copy(value = currentValue.value.copy()) Review comment: After seeing the new changes, I think the first version looks better. The caller sides is nested and we still have unnecessary copies for v1 format. What do you think? @viirya 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #28975: [SPARK-32148][SS] Fix stream-stream join issue on missing to copy reused unsafe row
cloud-fan commented on a change in pull request #28975: URL: https://github.com/apache/spark/pull/28975#discussion_r450086968 ## File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManager.scala ## @@ -259,6 +269,9 @@ class SymmetricHashJoinStateManager( return null } +// Make a copy on value row, as below cleanup logic may update the value row silently. +currentValue = currentValue.copy(value = currentValue.value.copy()) Review comment: After seeing the new changes, I think the first version looks better. The caller sides is nested and we still have necessary copies for v1 format. What do you think? @viirya 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #28975: [SPARK-32148][SS] Fix stream-stream join issue on missing to copy reused unsafe row
cloud-fan commented on a change in pull request #28975: URL: https://github.com/apache/spark/pull/28975#discussion_r450016555 ## File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManager.scala ## @@ -259,6 +269,9 @@ class SymmetricHashJoinStateManager( return null } +// Make a copy on value row, as below cleanup logic may update the value row silently. +currentValue = currentValue.copy(value = currentValue.value.copy()) Review comment: so this is the only place to do copy? 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #28975: [SPARK-32148][SS] Fix stream-stream join issue on missing to copy reused unsafe row
cloud-fan commented on a change in pull request #28975:
URL: https://github.com/apache/spark/pull/28975#discussion_r449458290
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManager.scala
##
@@ -451,10 +451,25 @@ class SymmetricHashJoinStateManager(
}
private trait KeyWithIndexToValueRowConverter {
+/** Defines the schema of the value row (the value side of K-V in state
store). */
def valueAttributes: Seq[Attribute]
+/**
+ * Convert the value row to (actual value, match) pair.
+ *
+ * NOTE: implementations should ensure the result row is NOT reused during
execution, as
+ * caller may use the value to store without copy().
Review comment:
I don't have a strong opinion here. Just for reference: The SMJ copies
the input row on-demand: only copy the rows when we need to buffer them. See
`SortMergeJoinExec.bufferMatchingRows`. It's only a few places we need to
buffer the rows, so it's OK to do it in SMJ. I'm not sure about streaming join
though.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #28975: [SPARK-32148][SS] Fix stream-stream join issue on missing to copy reused unsafe row
cloud-fan commented on a change in pull request #28975:
URL: https://github.com/apache/spark/pull/28975#discussion_r449123103
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManager.scala
##
@@ -451,10 +451,25 @@ class SymmetricHashJoinStateManager(
}
private trait KeyWithIndexToValueRowConverter {
+/** Defines the schema of the value row (the value side of K-V in state
store). */
def valueAttributes: Seq[Attribute]
+/**
+ * Convert the value row to (actual value, match) pair.
+ *
+ * NOTE: implementations should ensure the result row is NOT reused during
execution, as
+ * caller may use the value to store without copy().
Review comment:
I don't know the details, but how likely the caller side will store the
values?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #28975: [SPARK-32148][SS] Fix stream-stream join issue on missing to copy reused unsafe row
cloud-fan commented on a change in pull request #28975: URL: https://github.com/apache/spark/pull/28975#discussion_r449076851 ## File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingSymmetricHashJoinExec.scala ## @@ -295,6 +295,10 @@ case class StreamingSymmetricHashJoinExec( postJoinFilter(joinedRow.withLeft(leftKeyValue.value).withRight(rightValue)) } } + +// NOTE: we need to make sure `outerOutputIter` is evaluated "after" exhausting all of +// elements in `innerOutputIter`, because evaluation of `innerOutputIter` may update +// the match flag which the logic for outer join is relying on. Review comment: Just to clarify: this comment is not related to the bug and just to document an existing assumption? 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
