comphead commented on code in PR #7130:
URL: https://github.com/apache/datafusion-comet/pull/7130#discussion_r4332092370
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -1272,6 +1272,374 @@ class CometIcebergNativeSuite
}
}
+ /**
+ * Runs each query through [[checkIcebergNativeScan]] and returns one
message per query that
+ * failed, so every query is reported rather than only the first.
+ */
+ private def failingNativeScanChecks(queries: Seq[String]): Seq[String] =
+ queries.flatMap { q =>
+ try {
+ checkIcebergNativeScan(q)
+ None
+ } catch {
+ case scala.util.control.NonFatal(e) =>
+ val plan =
+ try spark.sql(q).queryExecution.executedPlan.toString
+ catch { case scala.util.control.NonFatal(pe) => s"<plan failed:
${pe.getMessage}>" }
+ Some(s"QUERY: $q\nERROR: ${e.getMessage}\nPLAN:\n$plan")
+ }
+ }
+
+ private def withCaseCatalog(warehouseDir: File, caseSensitive: Boolean)(f:
=> Unit): Unit =
+ withSQLConf(
+ "spark.sql.catalog.case_cat" -> "org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.case_cat.type" -> "hadoop",
+ "spark.sql.catalog.case_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true")(f)
+
+ test("issue 5540: case-sensitive read of columns a and A") {
Review Comment:
These three tests (`case-sensitive read of columns a and A`, `case-sensitive
filters on columns a and A` and the case-insensitive control at line 1373)
create plain tables and compare queries, with no equality-delete API calls.
`sql-tests/iceberg` already has fixtures that use `-- Config:` for catalog
settings, so a fixture with `-- Config: spark.sql.caseSensitive=true` and `(a
INT, A INT)` rows could replace about 70 lines of Scala and the
`failingNativeScanChecks` helper. Fixture `query` statements also report each
failure separately. I have not confirmed that the config line is applied before
the fixture's `CREATE TABLE`, though `withFixtureConfigs` appears to wrap the
whole file. The equality-delete tests need `commitEqualityDelete` and would
stay in Scala.
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -1272,6 +1272,374 @@ class CometIcebergNativeSuite
}
}
+ /**
+ * Runs each query through [[checkIcebergNativeScan]] and returns one
message per query that
+ * failed, so every query is reported rather than only the first.
+ */
+ private def failingNativeScanChecks(queries: Seq[String]): Seq[String] =
+ queries.flatMap { q =>
+ try {
+ checkIcebergNativeScan(q)
+ None
+ } catch {
+ case scala.util.control.NonFatal(e) =>
+ val plan =
+ try spark.sql(q).queryExecution.executedPlan.toString
+ catch { case scala.util.control.NonFatal(pe) => s"<plan failed:
${pe.getMessage}>" }
+ Some(s"QUERY: $q\nERROR: ${e.getMessage}\nPLAN:\n$plan")
+ }
+ }
+
+ private def withCaseCatalog(warehouseDir: File, caseSensitive: Boolean)(f:
=> Unit): Unit =
+ withSQLConf(
+ "spark.sql.catalog.case_cat" -> "org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.case_cat.type" -> "hadoop",
+ "spark.sql.catalog.case_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true")(f)
+
+ test("issue 5540: case-sensitive read of columns a and A") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ withCaseCatalog(warehouseDir, caseSensitive = true) {
+ val t = "case_cat.db.case_pair"
+ spark.sql(s"CREATE TABLE $t (a INT, A INT) USING iceberg")
+ spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40)")
+ val failures = failingNativeScanChecks(
+ Seq(
+ s"SELECT a, A FROM $t ORDER BY a",
+ s"SELECT A, a FROM $t ORDER BY a",
+ s"SELECT A FROM $t ORDER BY A",
+ s"SELECT a FROM $t ORDER BY a",
+ s"SELECT * FROM $t ORDER BY a"))
+ spark.sql(s"DROP TABLE $t")
+ assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
+ }
+ }
+ }
+
+ test("issue 5540: case-sensitive filters on columns a and A") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ withCaseCatalog(warehouseDir, caseSensitive = true) {
+ val t = "case_cat.db.case_pair_filter"
+ spark.sql(s"CREATE TABLE $t (a INT, A INT) USING iceberg")
+ spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40)")
+ // a spans [1, 3] and A spans [20, 40]: a filter on one column bound
to the other would
+ // prune the whole row group or keep the wrong row.
+ val failures = failingNativeScanChecks(
+ Seq(
+ s"SELECT a FROM $t WHERE A = 40",
+ s"SELECT A FROM $t WHERE a = 3",
+ s"SELECT A FROM $t WHERE A = 40",
+ s"SELECT a FROM $t WHERE a = 3",
+ s"SELECT a, A FROM $t WHERE A = 40",
+ s"SELECT a, A FROM $t WHERE a = 3",
+ s"SELECT a FROM $t WHERE A > 30",
+ s"SELECT a FROM $t WHERE A IN (20)",
+ s"SELECT A FROM $t WHERE a < 2"))
+ spark.sql(s"DROP TABLE $t")
+ assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
+ }
+ }
+ }
+
+ test("issue 5540: case-sensitive equality deletes keyed on a and on A") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ withCaseCatalog(warehouseDir, caseSensitive = true) {
+ val t = "case_cat.db.case_pair_eq"
+ spark.sql(
+ s"CREATE TABLE $t (a INT, A INT) USING iceberg TBLPROPERTIES
('format-version' = '2')")
+ spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40), (5, 60)")
+ // One delete keyed on `a` and one keyed on `A`: binding either key by
a folded name
+ // misreads one of them whichever field the fold resolves to. Only (5,
60) survives.
+ commitEqualityDelete("case_cat", "db", "case_pair_eq", "a", 1,
warehouseDir)
+ commitEqualityDelete("case_cat", "db", "case_pair_eq", "A", 40,
warehouseDir)
+ val failures = failingNativeScanChecks(
+ Seq(
+ s"SELECT a, A FROM $t ORDER BY a",
+ s"SELECT A FROM $t ORDER BY A",
+ s"SELECT a FROM $t ORDER BY a",
+ s"SELECT count(*) FROM $t"))
+ spark.sql(s"DROP TABLE $t")
+ assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
+ }
+ }
+ }
+
+ test("issue 5540: case-insensitive control on mixed-case column names") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ withCaseCatalog(warehouseDir, caseSensitive = false) {
+ val t = "case_cat.db.mixed_case"
+ spark.sql(
+ s"CREATE TABLE $t (Id INT, Val INT) USING iceberg " +
+ "TBLPROPERTIES ('format-version' = '2')")
+ spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40), (5, 60)")
+ commitEqualityDelete("case_cat", "db", "mixed_case", "Id", 1,
warehouseDir)
+ val failures = failingNativeScanChecks(
+ Seq(
+ s"SELECT id, VAL FROM $t ORDER BY ID",
+ s"SELECT ID FROM $t WHERE val = 40",
+ s"SELECT * FROM $t WHERE ID > 2 ORDER BY id"))
+ spark.sql(s"DROP TABLE $t")
+ assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
+ }
+ }
+ }
+
+ test("issue 5540: case-insensitive read of columns a and A with equality
deletes") {
Review Comment:
This test and `case-sensitive equality deletes keyed on a and on A` (line
1349) create the same `(a INT, A INT)` table, insert the same three rows and
commit the same two equality deletes on `a` and `A`. Only the read-time setting
differs. One test could build the table once, run the named-column queries
under `true`, then run `SELECT *` and `count(*)` under `false`, including the
Spark 4.1 `COLUMN_ALREADY_EXISTS` branch. That drops a table setup and about 25
lines.
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -1272,6 +1272,374 @@ class CometIcebergNativeSuite
}
}
+ /**
+ * Runs each query through [[checkIcebergNativeScan]] and returns one
message per query that
+ * failed, so every query is reported rather than only the first.
+ */
+ private def failingNativeScanChecks(queries: Seq[String]): Seq[String] =
+ queries.flatMap { q =>
+ try {
+ checkIcebergNativeScan(q)
+ None
+ } catch {
+ case scala.util.control.NonFatal(e) =>
+ val plan =
+ try spark.sql(q).queryExecution.executedPlan.toString
+ catch { case scala.util.control.NonFatal(pe) => s"<plan failed:
${pe.getMessage}>" }
+ Some(s"QUERY: $q\nERROR: ${e.getMessage}\nPLAN:\n$plan")
+ }
+ }
+
+ private def withCaseCatalog(warehouseDir: File, caseSensitive: Boolean)(f:
=> Unit): Unit =
+ withSQLConf(
+ "spark.sql.catalog.case_cat" -> "org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.case_cat.type" -> "hadoop",
+ "spark.sql.catalog.case_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true")(f)
+
+ test("issue 5540: case-sensitive read of columns a and A") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ withCaseCatalog(warehouseDir, caseSensitive = true) {
+ val t = "case_cat.db.case_pair"
+ spark.sql(s"CREATE TABLE $t (a INT, A INT) USING iceberg")
+ spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40)")
+ val failures = failingNativeScanChecks(
+ Seq(
+ s"SELECT a, A FROM $t ORDER BY a",
+ s"SELECT A, a FROM $t ORDER BY a",
+ s"SELECT A FROM $t ORDER BY A",
+ s"SELECT a FROM $t ORDER BY a",
+ s"SELECT * FROM $t ORDER BY a"))
+ spark.sql(s"DROP TABLE $t")
+ assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
+ }
+ }
+ }
+
+ test("issue 5540: case-sensitive filters on columns a and A") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ withCaseCatalog(warehouseDir, caseSensitive = true) {
+ val t = "case_cat.db.case_pair_filter"
+ spark.sql(s"CREATE TABLE $t (a INT, A INT) USING iceberg")
+ spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40)")
+ // a spans [1, 3] and A spans [20, 40]: a filter on one column bound
to the other would
+ // prune the whole row group or keep the wrong row.
+ val failures = failingNativeScanChecks(
+ Seq(
+ s"SELECT a FROM $t WHERE A = 40",
+ s"SELECT A FROM $t WHERE a = 3",
+ s"SELECT A FROM $t WHERE A = 40",
+ s"SELECT a FROM $t WHERE a = 3",
+ s"SELECT a, A FROM $t WHERE A = 40",
+ s"SELECT a, A FROM $t WHERE a = 3",
+ s"SELECT a FROM $t WHERE A > 30",
+ s"SELECT a FROM $t WHERE A IN (20)",
+ s"SELECT A FROM $t WHERE a < 2"))
+ spark.sql(s"DROP TABLE $t")
+ assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
+ }
+ }
+ }
+
+ test("issue 5540: case-sensitive equality deletes keyed on a and on A") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ withCaseCatalog(warehouseDir, caseSensitive = true) {
+ val t = "case_cat.db.case_pair_eq"
+ spark.sql(
+ s"CREATE TABLE $t (a INT, A INT) USING iceberg TBLPROPERTIES
('format-version' = '2')")
+ spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40), (5, 60)")
+ // One delete keyed on `a` and one keyed on `A`: binding either key by
a folded name
+ // misreads one of them whichever field the fold resolves to. Only (5,
60) survives.
+ commitEqualityDelete("case_cat", "db", "case_pair_eq", "a", 1,
warehouseDir)
+ commitEqualityDelete("case_cat", "db", "case_pair_eq", "A", 40,
warehouseDir)
+ val failures = failingNativeScanChecks(
+ Seq(
+ s"SELECT a, A FROM $t ORDER BY a",
+ s"SELECT A FROM $t ORDER BY A",
+ s"SELECT a FROM $t ORDER BY a",
+ s"SELECT count(*) FROM $t"))
+ spark.sql(s"DROP TABLE $t")
+ assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
+ }
+ }
+ }
+
+ test("issue 5540: case-insensitive control on mixed-case column names") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ withCaseCatalog(warehouseDir, caseSensitive = false) {
+ val t = "case_cat.db.mixed_case"
+ spark.sql(
+ s"CREATE TABLE $t (Id INT, Val INT) USING iceberg " +
+ "TBLPROPERTIES ('format-version' = '2')")
+ spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40), (5, 60)")
+ commitEqualityDelete("case_cat", "db", "mixed_case", "Id", 1,
warehouseDir)
+ val failures = failingNativeScanChecks(
+ Seq(
+ s"SELECT id, VAL FROM $t ORDER BY ID",
+ s"SELECT ID FROM $t WHERE val = 40",
+ s"SELECT * FROM $t WHERE ID > 2 ORDER BY id"))
+ spark.sql(s"DROP TABLE $t")
+ assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
+ }
+ }
+ }
+
+ test("issue 5540: case-insensitive read of columns a and A with equality
deletes") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ val t = "case_cat.db.case_pair_ci"
+ withCaseCatalog(warehouseDir, caseSensitive = true) {
+ spark.sql(
+ s"CREATE TABLE $t (a INT, A INT) USING iceberg TBLPROPERTIES
('format-version' = '2')")
+ spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40), (5, 60)")
+ commitEqualityDelete("case_cat", "db", "case_pair_ci", "a", 1,
warehouseDir)
+ commitEqualityDelete("case_cat", "db", "case_pair_ci", "A", 40,
warehouseDir)
+ }
+ // `a` is ambiguous here, so the queries name no column: only (5, 60)
survives. Spark 4.1+
+ // rejects the duplicate columns itself before any scan runs.
+ withCaseCatalog(warehouseDir, caseSensitive = false) {
+ Seq(s"SELECT * FROM $t", s"SELECT count(*) FROM $t").foreach { query =>
+ if (isSpark41Plus) {
+ val e =
intercept[org.apache.spark.sql.AnalysisException](spark.sql(query).collect())
+ assert(e.getMessage.contains("COLUMN_ALREADY_EXISTS"),
e.getMessage)
+ } else {
+ checkIcebergNativeScan(query)
+ }
+ }
+ }
+ withCaseCatalog(warehouseDir, caseSensitive = true) {
+ spark.sql(s"DROP TABLE $t")
+ }
+ }
+ }
+
+ test("issue 5540: equality delete survives a case-only column rename") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ val t = "case_cat.db.case_rename"
+ withCaseCatalog(warehouseDir, caseSensitive = true) {
+ spark.sql(
+ s"CREATE TABLE $t (id INT, val INT) USING iceberg " +
+ "TBLPROPERTIES ('format-version' = '2')")
+ spark.sql(s"INSERT INTO $t VALUES (1, 10), (2, 20), (3, 30)")
+ commitEqualityDelete("case_cat", "db", "case_rename", "id", 2,
warehouseDir)
+ spark.sql(s"ALTER TABLE $t RENAME COLUMN id TO ID")
+ }
+ Seq(true, false).foreach { caseSensitive =>
+ withCaseCatalog(warehouseDir, caseSensitive) {
+ checkIcebergNativeScan(s"SELECT ID, val FROM $t ORDER BY ID")
+ checkIcebergNativeScan(s"SELECT count(*) FROM $t")
+ }
+ }
+ withCaseCatalog(warehouseDir, caseSensitive = true) {
+ spark.sql(s"DROP TABLE $t")
+ }
+ }
+ }
+
+ test("issue 5540: name mapping onto physical columns that differ in case") {
Review Comment:
This test and `name mapping onto nested children that differ in case` (line
1501) repeat the same reflective `SparkTableUtil.importSparkTable` call of
about 15 lines, and the suite already calls `importSparkTable` through
reflection in other places. A small private helper such as
`importParquetInto(table, sourceTableName, stagingDir)` would remove the two
new copies.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
