dwsmith1983 commented on code in PR #7130:
URL: https://github.com/apache/datafusion-comet/pull/7130#discussion_r4332840879
##########
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:
Added `importParquetInto` in b8518b8de for these two. The older call sites
look up `SparkTableUtil` before creating their source table and cancel if it's
missing, so I left them as they were.
--
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]
