dwsmith1983 commented on code in PR #7130:
URL: https://github.com/apache/datafusion-comet/pull/7130#discussion_r4355666293
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -1272,6 +1272,295 @@ class CometIcebergNativeSuite
}
}
+ /** Returns one message per query failing [[checkIcebergNativeScan]], so all
are reported. */
+ 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)
+
+ private def importParquetInto(
+ table: org.apache.iceberg.Table,
+ sourceTable: String,
+ stagingDir: String): Unit =
+ Class
+ .forName("org.apache.iceberg.spark.SparkTableUtil")
+ .getMethod(
+ "importSparkTable",
+ classOf[org.apache.spark.sql.SparkSession],
+ classOf[org.apache.spark.sql.catalyst.TableIdentifier],
+ classOf[org.apache.iceberg.Table],
+ classOf[String])
+ .invoke(
+ null,
+ spark,
+ new org.apache.spark.sql.catalyst.TableIdentifier(sourceTable),
+ table,
+ stagingDir)
+
+ test("issue 5540: equality deletes keyed on columns a and A") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ val t = "case_cat.db.case_pair_eq"
+ var failures = Seq.empty[String]
+ 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)")
+ // Binding either delete key by a folded name would misread one of
them.
+ commitEqualityDelete("case_cat", "db", "case_pair_eq", "a", 1,
warehouseDir)
+ commitEqualityDelete("case_cat", "db", "case_pair_eq", "A", 40,
warehouseDir)
+ 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"))
+ }
+ // `a` is ambiguous here, so name no column; Spark 4.1+ rejects the
duplicates itself.
+ 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")
+ }
+ 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: 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") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ val t = "case_cat.db.case_mapped"
+ val dataPath = s"${warehouseDir.getAbsolutePath}/case_mapped_src"
+ withCaseCatalog(warehouseDir, caseSensitive = false) {
+ spark
+ .sql("SELECT CAST(id AS INT) AS ID, CAST(id * 10 AS INT) AS Val FROM
range(1, 4)")
+ .coalesce(1)
+ .write
+ .parquet(dataPath)
+ spark.sql(s"CREATE TABLE $t (id INT, val INT) USING iceberg")
+ // Set before the import so the file's column metrics are computed
through the mapping.
+ spark.sql(
+ s"ALTER TABLE $t SET TBLPROPERTIES ('schema.name-mapping.default' =
" +
+
"""'[{"field-id":1,"names":["ID"]},{"field-id":2,"names":["Val"]}]')""")
+ val icebergTable = spark.sessionState.catalogManager
+ .catalog("case_cat")
+ .asInstanceOf[org.apache.iceberg.spark.SparkCatalog]
+ .loadTable(
+ org.apache.spark.sql.connector.catalog.Identifier.of(Array("db"),
"case_mapped"))
+ .asInstanceOf[org.apache.iceberg.spark.source.SparkTable]
+ .table()
+ spark.sql(s"CREATE TABLE parquet_case_mapped USING parquet LOCATION
'$dataPath'")
+ try {
+ importParquetInto(icebergTable, "parquet_case_mapped",
s"$warehouseDir/staging")
+ Seq(true, false).foreach { caseSensitive =>
+ withCaseCatalog(warehouseDir, caseSensitive) {
+ checkIcebergNativeScan(s"SELECT id, val FROM $t ORDER BY id")
+ checkIcebergNativeScan(s"SELECT val FROM $t WHERE id = 2")
+ }
+ }
+ } finally {
+ spark.sql("DROP TABLE IF EXISTS parquet_case_mapped")
+ spark.sql(s"DROP TABLE IF EXISTS $t")
+ }
+ }
+ }
+ }
+
+ test("issue 5540: name mapping onto nested children that differ in case") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ val name = "case_mapped_nested"
+ val t = s"case_cat.db.$name"
+ var failures = Seq.empty[String]
+ // Physical (s, x, y) names; the first matches the table exactly, as a
control.
+ Seq(("s", "x", "y"), ("S", "X", "Y"), ("s", "X", "y")).foreach { case
(sName, xName, yName) =>
+ withTempIcebergDir { warehouseDir =>
+ val dataPath = s"${warehouseDir.getAbsolutePath}/${name}_src"
+ withCaseCatalog(warehouseDir, caseSensitive = false) {
+ spark
+ .sql(s"SELECT CAST(id AS INT) AS id, named_struct('$xName',
CAST(id * 10 AS INT), " +
+ s"'$yName', CAST(id * 100 AS INT)) AS $sName FROM range(1, 4)")
+ .coalesce(1)
+ .write
+ .parquet(dataPath)
+ spark.sql(s"CREATE TABLE $t (id INT, s STRUCT<x: INT, y: INT>) USING
iceberg")
+ val icebergTable = loadIcebergTable(spark, "case_cat", "db", name)
+ .asInstanceOf[org.apache.iceberg.Table]
+ def fieldId(path: String): Int =
icebergTable.schema().findField(path).fieldId()
+ val mapping =
+ s"""[{"field-id":${fieldId("id")},"names":["id"]},""" +
+ s"""{"field-id":${fieldId("s")},"names":["$sName"],"fields":["""
+
+ s"""{"field-id":${fieldId("s.x")},"names":["$xName"]},""" +
+ s"""{"field-id":${fieldId("s.y")},"names":["$yName"]}]}]"""
+ spark.sql(
+ s"ALTER TABLE $t SET TBLPROPERTIES ('schema.name-mapping.default'
= '$mapping')")
+ icebergTable.refresh()
+ spark.sql(s"CREATE TABLE parquet_$name USING parquet LOCATION
'$dataPath'")
+ try {
+ importParquetInto(icebergTable, s"parquet_$name",
s"$warehouseDir/staging")
+ // Spark reads the mapped children, so the checks below compare
against real values.
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ assert(
+ spark.sql(s"SELECT id, s FROM $t ORDER BY id").collect().toSeq
==
+ (1 to 3).map(i => Row(i, Row(i * 10, i * 100))))
+ }
+ for {
+ caseSensitive <- Seq(false, true)
+ pruning <- Seq(true, false)
+ } {
+ withCaseCatalog(warehouseDir, caseSensitive) {
+ withSQLConf(
+ CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key ->
pruning.toString) {
+ val label = s"file $sName<$xName, $yName>,
caseSensitive=$caseSensitive, " +
+ s"nestedPruning=$pruning"
+ // Unchecked, as iceberg-rust maps top-level names only: `s`
beside `id` is
+ // NULL (#6790), and a pruned `s.y` over physical `X`, `Y`
reads `X` (#7134).
+ val skipsPrunedY = pruning && xName == "X" && yName == "Y"
+ val queries = Seq(
+ s"SELECT s FROM $t ORDER BY s.x",
+ s"SELECT s.x FROM $t ORDER BY s.x",
+ s"SELECT s.y FROM $t ORDER BY s.y",
+ s"SELECT s.y FROM $t WHERE s.x = 20")
+ .filterNot(q => skipsPrunedY && q.endsWith("ORDER BY s.y"))
+ failures ++= failingNativeScanChecks(queries).map(f =>
s"[$label]\n$f")
+ }
+ }
+ }
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS parquet_$name")
+ spark.sql(s"DROP TABLE IF EXISTS $t")
+ }
+ }
+ }
+ }
+ assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
+ }
+
+ test("issue 5540: metadata columns beside data columns under case
sensitivity") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ val t = "case_cat.db.case_meta"
+ var failures = Seq.empty[String]
+ withCaseCatalog(warehouseDir, caseSensitive = true) {
+ spark.sql(
+ s"CREATE TABLE $t (id INT, Region STRING, val INT) USING iceberg " +
+ "PARTITIONED BY (Region) TBLPROPERTIES ('format-version' = '2')")
+ spark.sql(s"INSERT INTO $t VALUES (1, 'a', 10), (2, 'a', 20), (3, 'b',
30)")
+ }
+ // Metadata columns resolve by reserved field id, so the exact name
check must skip them.
Review Comment:
It doesn't skip them. `build_projection_expressions` requires every
projected name in the batch, and iceberg-rust emits `_file`, `_pos`, `_spec_id`
and `_partition` under those names, so they pass the same way data columns do.
Reworded in 74979463a.
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -1272,6 +1272,295 @@ class CometIcebergNativeSuite
}
}
+ /** Returns one message per query failing [[checkIcebergNativeScan]], so all
are reported. */
+ 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)
+
+ private def importParquetInto(
+ table: org.apache.iceberg.Table,
+ sourceTable: String,
+ stagingDir: String): Unit =
+ Class
+ .forName("org.apache.iceberg.spark.SparkTableUtil")
+ .getMethod(
+ "importSparkTable",
+ classOf[org.apache.spark.sql.SparkSession],
+ classOf[org.apache.spark.sql.catalyst.TableIdentifier],
+ classOf[org.apache.iceberg.Table],
+ classOf[String])
+ .invoke(
+ null,
+ spark,
+ new org.apache.spark.sql.catalyst.TableIdentifier(sourceTable),
+ table,
+ stagingDir)
+
+ test("issue 5540: equality deletes keyed on columns a and A") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ withTempIcebergDir { warehouseDir =>
+ val t = "case_cat.db.case_pair_eq"
+ var failures = Seq.empty[String]
+ 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)")
+ // Binding either delete key by a folded name would misread one of
them.
+ commitEqualityDelete("case_cat", "db", "case_pair_eq", "a", 1,
warehouseDir)
+ commitEqualityDelete("case_cat", "db", "case_pair_eq", "A", 40,
warehouseDir)
+ 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"))
+ }
+ // `a` is ambiguous here, so name no column; Spark 4.1+ rejects the
duplicates itself.
+ withCaseCatalog(warehouseDir, caseSensitive = false) {
Review Comment:
Yes, and it showed up as wrong rows. On Spark 3.5 and 4.0 with
`caseSensitive=false`, `SELECT *` over a table with `a` and `A` mapped both to
`a`'s field id, so the nested-field check never looked at `A`. A field renamed
under `A` then read back as NULL from files written before the rename.
74979463a switches to `globalFieldIdMapping.get(attr.name)`, the lookup the
serde already uses, and adds a test for that table. Spark 4.1 rejects the
`SELECT *` with `COLUMN_ALREADY_EXISTS`, so on 4.1 the test covers the
case-sensitive reads: `SELECT A` falls back and `SELECT a` stays native.
--
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]
