You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2023/07/12 19:22:30 UTC
[spark] branch master updated: [SPARK-44392][SQL][TESTS] Add tests for schema pruning in delta-based UPDATEs
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new db160e8a166 [SPARK-44392][SQL][TESTS] Add tests for schema pruning in delta-based UPDATEs
db160e8a166 is described below
commit db160e8a16609ee327962f08bd5b5ede76c44888
Author: aokolnychyi <ao...@apple.com>
AuthorDate: Wed Jul 12 12:22:13 2023 -0700
[SPARK-44392][SQL][TESTS] Add tests for schema pruning in delta-based UPDATEs
### What changes were proposed in this pull request?
This PR adds tests for schema pruning for delta-based UPDATEs, similar to the ones we have for MERGE.
### Why are the changes needed?
These changes are needed to verify schema pruning works as expected.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
This PR comes with tests.
Closes #41962 from aokolnychyi/spark-44392.
Authored-by: aokolnychyi <ao...@apple.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../DeltaBasedMergeIntoTableSuiteBase.scala | 17 -------
...ltaBasedUpdateAsDeleteAndInsertTableSuite.scala | 34 +-------------
.../sql/connector/DeltaBasedUpdateTableSuite.scala | 52 +---------------------
....scala => DeltaBasedUpdateTableSuiteBase.scala} | 40 +++++++++++------
.../sql/connector/RowLevelOperationSuiteBase.scala | 14 ++++++
5 files changed, 43 insertions(+), 114 deletions(-)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoTableSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoTableSuiteBase.scala
index ffc46e3757a..43497edd99f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoTableSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoTableSuiteBase.scala
@@ -18,9 +18,6 @@
package org.apache.spark.sql.connector
import org.apache.spark.sql.Row
-import org.apache.spark.sql.catalyst.types.DataTypeUtils
-import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
-import org.apache.spark.sql.types.StructType
abstract class DeltaBasedMergeIntoTableSuiteBase extends MergeIntoTableSuiteBase {
@@ -215,18 +212,4 @@ abstract class DeltaBasedMergeIntoTableSuiteBase extends MergeIntoTableSuiteBase
Row(3, 300, "china", "software"))) // insert
}
}
-
- private def executeAndCheckScan(
- query: String,
- expectedScanSchema: String): Unit = {
-
- val executedPlan = executeAndKeepPlan {
- sql(query)
- }
-
- val scan = collect(executedPlan) {
- case s: BatchScanExec => s
- }.head
- assert(DataTypeUtils.sameType(scan.schema, StructType.fromDDL(expectedScanSchema)))
- }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateAsDeleteAndInsertTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateAsDeleteAndInsertTableSuite.scala
index 51f3d49372c..363b90b45b8 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateAsDeleteAndInsertTableSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateAsDeleteAndInsertTableSuite.scala
@@ -17,9 +17,7 @@
package org.apache.spark.sql.connector
-import org.apache.spark.sql.{AnalysisException, Row}
-
-class DeltaBasedUpdateAsDeleteAndInsertTableSuite extends UpdateTableSuiteBase {
+class DeltaBasedUpdateAsDeleteAndInsertTableSuite extends DeltaBasedUpdateTableSuiteBase {
override protected lazy val extraTableProps: java.util.Map[String, String] = {
val props = new java.util.HashMap[String, String]()
@@ -27,34 +25,4 @@ class DeltaBasedUpdateAsDeleteAndInsertTableSuite extends UpdateTableSuiteBase {
props.put("split-updates", "true")
props
}
-
- test("nullable row ID attrs") {
- createAndInitTable("pk INT, salary INT, dep STRING",
- """{ "pk": 1, "salary": 300, "dep": 'hr' }
- |{ "pk": 2, "salary": 150, "dep": 'software' }
- |{ "pk": 3, "salary": 120, "dep": 'hr' }
- |""".stripMargin)
-
- checkErrorMatchPVals(
- exception = intercept[AnalysisException] {
- sql(s"UPDATE $tableNameAsString SET salary = -1 WHERE pk = 1")
- },
- errorClass = "NULLABLE_ROW_ID_ATTRIBUTES",
- parameters = Map("nullableRowIdAttrs" -> "pk#\\d+")
- )
- }
-
- test("update with assignments to row ID") {
- createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
- """{ "pk": 1, "id": 1, "dep": "hr" }
- |{ "pk": 2, "id": 2, "dep": "software" }
- |{ "pk": 3, "id": 3, "dep": "hr" }
- |""".stripMargin)
-
- sql(s"UPDATE $tableNameAsString SET pk = 10 WHERE id = 1")
-
- checkAnswer(
- sql(s"SELECT * FROM $tableNameAsString"),
- Row(10, 1, "hr") :: Row(2, 2, "software") :: Row(3, 3, "hr") :: Nil)
- }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateTableSuite.scala
index 73b6ec22a72..bf6dcfac04c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateTableSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateTableSuite.scala
@@ -17,61 +17,11 @@
package org.apache.spark.sql.connector
-import org.apache.spark.sql.{AnalysisException, Row}
-
-class DeltaBasedUpdateTableSuite extends UpdateTableSuiteBase {
+class DeltaBasedUpdateTableSuite extends DeltaBasedUpdateTableSuiteBase {
override protected lazy val extraTableProps: java.util.Map[String, String] = {
val props = new java.util.HashMap[String, String]()
props.put("supports-deltas", "true")
props
}
-
- test("nullable row ID attrs") {
- createAndInitTable("pk INT, salary INT, dep STRING",
- """{ "pk": 1, "salary": 300, "dep": 'hr' }
- |{ "pk": 2, "salary": 150, "dep": 'software' }
- |{ "pk": 3, "salary": 120, "dep": 'hr' }
- |""".stripMargin)
-
- val exception = intercept[AnalysisException] {
- sql(s"UPDATE $tableNameAsString SET salary = -1 WHERE pk = 1")
- }
- assert(exception.message.contains("Row ID attributes cannot be nullable"))
- }
-
- test("update with assignments to row ID") {
- createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
- """{ "pk": 1, "id": 1, "dep": "hr" }
- |{ "pk": 2, "id": 2, "dep": "software" }
- |{ "pk": 3, "id": 3, "dep": "hr" }
- |""".stripMargin)
-
- sql(s"UPDATE $tableNameAsString SET pk = 10 WHERE id = 1")
-
- checkAnswer(
- sql(s"SELECT * FROM $tableNameAsString"),
- Row(10, 1, "hr") :: Row(2, 2, "software") :: Row(3, 3, "hr") :: Nil)
- }
-
- test("update with nondeterministic conditions") {
- createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
- """{ "pk": 1, "id": 1, "dep": "hr" }
- |{ "pk": 2, "id": 2, "dep": "software" }
- |{ "pk": 3, "id": 3, "dep": "hr" }
- |""".stripMargin)
-
- checkError(
- exception = intercept[AnalysisException] {
- sql(s"UPDATE $tableNameAsString SET dep = 'invalid' WHERE id <= 1 AND rand() > 0.5")
- },
- errorClass = "INVALID_NON_DETERMINISTIC_EXPRESSIONS",
- parameters = Map(
- "sqlExprs" -> "\"((id <= 1) AND (rand() > 0.5))\""),
- context = ExpectedContext(
- fragment = "UPDATE cat.ns1.test_table SET dep = 'invalid' WHERE id <= 1 AND rand() > 0.5",
- start = 0,
- stop = 75)
- )
- }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateTableSuiteBase.scala
similarity index 66%
copy from sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateTableSuite.scala
copy to sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateTableSuiteBase.scala
index 73b6ec22a72..46942cac1c7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateTableSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateTableSuiteBase.scala
@@ -19,13 +19,7 @@ package org.apache.spark.sql.connector
import org.apache.spark.sql.{AnalysisException, Row}
-class DeltaBasedUpdateTableSuite extends UpdateTableSuiteBase {
-
- override protected lazy val extraTableProps: java.util.Map[String, String] = {
- val props = new java.util.HashMap[String, String]()
- props.put("supports-deltas", "true")
- props
- }
+abstract class DeltaBasedUpdateTableSuiteBase extends UpdateTableSuiteBase {
test("nullable row ID attrs") {
createAndInitTable("pk INT, salary INT, dep STRING",
@@ -34,10 +28,13 @@ class DeltaBasedUpdateTableSuite extends UpdateTableSuiteBase {
|{ "pk": 3, "salary": 120, "dep": 'hr' }
|""".stripMargin)
- val exception = intercept[AnalysisException] {
- sql(s"UPDATE $tableNameAsString SET salary = -1 WHERE pk = 1")
- }
- assert(exception.message.contains("Row ID attributes cannot be nullable"))
+ checkErrorMatchPVals(
+ exception = intercept[AnalysisException] {
+ sql(s"UPDATE $tableNameAsString SET salary = -1 WHERE pk = 1")
+ },
+ errorClass = "NULLABLE_ROW_ID_ATTRIBUTES",
+ parameters = Map("nullableRowIdAttrs" -> "pk#\\d+")
+ )
}
test("update with assignments to row ID") {
@@ -66,12 +63,29 @@ class DeltaBasedUpdateTableSuite extends UpdateTableSuiteBase {
sql(s"UPDATE $tableNameAsString SET dep = 'invalid' WHERE id <= 1 AND rand() > 0.5")
},
errorClass = "INVALID_NON_DETERMINISTIC_EXPRESSIONS",
- parameters = Map(
- "sqlExprs" -> "\"((id <= 1) AND (rand() > 0.5))\""),
+ parameters = Map("sqlExprs" -> "\"((id <= 1) AND (rand() > 0.5))\""),
context = ExpectedContext(
fragment = "UPDATE cat.ns1.test_table SET dep = 'invalid' WHERE id <= 1 AND rand() > 0.5",
start = 0,
stop = 75)
)
}
+
+ test("update with schema pruning") {
+ createAndInitTable("pk INT NOT NULL, id INT, salary INT, dep STRING",
+ """{ "pk": 1, "id": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "id": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "id": 3, "salary": 300, "dep": "hr" }
+ |""".stripMargin)
+
+ executeAndCheckScan(
+ s"UPDATE $tableNameAsString SET salary = -1, id = -1, dep = 'invalid' WHERE pk = 1",
+ // `pk` is used in the condition
+ // `_partition` is used in the requested write distribution
+ expectedScanSchema = "pk INT, _partition STRING")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, -1, -1, "invalid") :: Row(2, 2, 200, "software") :: Row(3, 3, 300, "hr") :: Nil)
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
index 1f8c20947b8..5df80c111ba 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
@@ -119,6 +119,20 @@ abstract class RowLevelOperationSuiteBase
stripAQEPlan(executedPlan)
}
+ protected def executeAndCheckScan(
+ query: String,
+ expectedScanSchema: String): Unit = {
+
+ val executedPlan = executeAndKeepPlan {
+ sql(query)
+ }
+
+ val scan = collect(executedPlan) {
+ case s: BatchScanExec => s
+ }.head
+ assert(DataTypeUtils.sameType(scan.schema, StructType.fromDDL(expectedScanSchema)))
+ }
+
protected def executeAndCheckScans(
query: String,
primaryScanSchema: String,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org