You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2020/08/26 02:03:51 UTC
[GitHub] [spark] maropu commented on a change in pull request #29475: [SPARK-32659][SQL] Fix the data issue of inserted DPP on non-atomic type
maropu commented on a change in pull request #29475:
URL: https://github.com/apache/spark/pull/29475#discussion_r476972100
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala
##########
@@ -114,9 +114,10 @@ case class InSubqueryExec(
child: Expression,
plan: BaseSubqueryExec,
exprId: ExprId,
- private var resultBroadcast: Broadcast[Array[Any]] = null) extends ExecSubqueryExpression {
+ private var resultBroadcast: Broadcast[Set[Any]] = null) extends ExecSubqueryExpression {
- @transient private var result: Array[Any] = _
+ @transient private var result: Set[Any] = _
+ @transient private lazy val inSet: InSet = InSet(child, result)
Review comment:
nit: `val inSet: InSet =` -> `val inSet =`
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
##########
@@ -1304,6 +1304,55 @@ abstract class DynamicPartitionPruningSuiteBase
)
}
}
+
+ test("SPARK-32659: Fix the data issue of inserted DPP on non-atomic type") {
+ withSQLConf(
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "2", // Make sure insert DPP
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false") {
+ withTable("df1", "df2") {
+ spark.range(1000)
+ .select(col("id"), col("id").as("k"))
+ .write
+ .partitionBy("k")
+ .format(tableFormat)
+ .mode("overwrite")
+ .saveAsTable("df1")
+
+ spark.range(100)
+ .select(col("id"), col("id").as("k"))
+ .write
+ .partitionBy("k")
+ .format(tableFormat)
+ .mode("overwrite")
+ .saveAsTable("df2")
+
+ Seq(CodegenObjectFactoryMode.NO_CODEGEN,
+ CodegenObjectFactoryMode.CODEGEN_ONLY).foreach { mode =>
+ Seq(true, false).foreach { pruning =>
+ withSQLConf(
+ SQLConf.CODEGEN_FACTORY_MODE.key -> s"${mode.toString}",
Review comment:
nit: `s"${mode.toString}"` -> `mode.toString`
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
##########
@@ -1304,6 +1304,55 @@ abstract class DynamicPartitionPruningSuiteBase
)
}
}
+
+ test("SPARK-32659: Fix the data issue of inserted DPP on non-atomic type") {
+ withSQLConf(
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "2", // Make sure insert DPP
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false") {
+ withTable("df1", "df2") {
+ spark.range(1000)
+ .select(col("id"), col("id").as("k"))
+ .write
+ .partitionBy("k")
+ .format(tableFormat)
+ .mode("overwrite")
+ .saveAsTable("df1")
+
+ spark.range(100)
+ .select(col("id"), col("id").as("k"))
+ .write
+ .partitionBy("k")
+ .format(tableFormat)
+ .mode("overwrite")
+ .saveAsTable("df2")
+
+ Seq(CodegenObjectFactoryMode.NO_CODEGEN,
+ CodegenObjectFactoryMode.CODEGEN_ONLY).foreach { mode =>
+ Seq(true, false).foreach { pruning =>
+ withSQLConf(
+ SQLConf.CODEGEN_FACTORY_MODE.key -> s"${mode.toString}",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> s"${pruning}") {
Review comment:
nit: `s"$pruning"`
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
##########
@@ -1304,6 +1304,55 @@ abstract class DynamicPartitionPruningSuiteBase
)
}
}
+
+ test("SPARK-32659: Fix the data issue of inserted DPP on non-atomic type") {
+ withSQLConf(
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "2", // Make sure insert DPP
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false") {
+ withTable("df1", "df2") {
+ spark.range(1000)
+ .select(col("id"), col("id").as("k"))
+ .write
+ .partitionBy("k")
+ .format(tableFormat)
+ .mode("overwrite")
+ .saveAsTable("df1")
+
+ spark.range(100)
+ .select(col("id"), col("id").as("k"))
+ .write
+ .partitionBy("k")
+ .format(tableFormat)
+ .mode("overwrite")
+ .saveAsTable("df2")
+
+ Seq(CodegenObjectFactoryMode.NO_CODEGEN,
+ CodegenObjectFactoryMode.CODEGEN_ONLY).foreach { mode =>
+ Seq(true, false).foreach { pruning =>
+ withSQLConf(
+ SQLConf.CODEGEN_FACTORY_MODE.key -> s"${mode.toString}",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> s"${pruning}") {
+ val df = sql(
+ """
+ |SELECT df1.id, df2.k
+ |FROM df1
+ | JOIN df2
+ | ON struct(df1.k) = struct(df2.k)
Review comment:
Does this issue only happen in struct cases? Looks like the title `non-atomic type` suggests the other non-atomic cases can happen. If so, I think its better to add tests for them, too.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org