zhangfengcdt commented on code in PR #6851:
URL: https://github.com/apache/datafusion-comet/pull/6851#discussion_r4244539655
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -2172,51 +2172,74 @@ class CometIcebergWriteActionSuite
}
}
- // https://github.com/apache/datafusion-comet/issues/6138. iceberg-rust
compares float partition
- // values with an equality that treats -0.0 and 0.0 as one value, so its
writers filed both under
- // whichever arrived first, and a read that pruned on the other value lost
rows. iceberg-java
- // keeps the two apart, so these writes fall back until iceberg-rust does
too.
- test("signed-zero float and double identity partitions match iceberg-java") {
+ // https://github.com/apache/datafusion-comet/issues/6138. iceberg-rust
compared float partition
+ // values with an equality that treated -0.0 and 0.0 as one value, so its
writers filed both
+ // under whichever arrived first, and a read that pruned on the other value
lost rows. Since
+ // apache/iceberg-rust#3327 it keeps them apart and puts every NaN in one
partition, as
+ // iceberg-java does. One shuffle partition puts every row in one task, so
the values meet in one
+ // writer. The fanout writer gets them interleaved. The clustered writer
gets each partition's
+ // rows together: Spark's sort ties -0.0 with 0.0, so a zero that repeats
could come back
+ // unclustered, which iceberg-java rejects too.
Review Comment:
Could we pin this with assertFailsLikeSpark so both writers are shown to
fail with the same exception here? Before this PR, the native writer merged the
zeros silently, so it might be worth a test that it now fails the same way.
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala:
##########
@@ -1500,92 +1500,13 @@ class CometIcebergWriteDetectionSuite extends
CometTestBase with CometIcebergTes
}
}
- test("fall-back: identity partition on a float or double column") {
- // iceberg-rust groups float partition values with an equality that treats
-0.0 and 0.0 as one
- // value, where iceberg-java keeps them apart (#6138).
- withDetectionCatalog { _ =>
- Seq("float" -> "FLOAT", "double" -> "DOUBLE").foreach { case (typeName,
sqlType) =>
- val table = s"part_$typeName"
- spark.sql(s"""
- CREATE TABLE $catalog.$ns.$table (id INT, v $sqlType)
- USING iceberg PARTITIONED BY (v)
- """)
- val writeExec = captureWriteExec(table, allowWriteFailure = false) {
- spark.sql(s"INSERT INTO $catalog.$ns.$table VALUES (1, CAST(1.5 AS
$sqlType))")
- }
- assertUnsupportedContains(writeExec, table, "partition field v",
typeName, "-0.0")
- }
- }
- }
-
- test("fall-back: identity partition on a nested double field") {
- // The source of a partition field can be nested inside a struct.
`Schema.findField` resolves
- // a nested id too, so the rule must not fail open for it.
- withDetectionCatalog { _ =>
- spark.sql(s"""
- CREATE TABLE $catalog.$ns.part_nested (id INT, s STRUCT<v: DOUBLE>)
- USING iceberg PARTITIONED BY (s.v)
- """)
- val writeExec = captureWriteExec("part_nested", allowWriteFailure =
false) {
- spark.sql(s"INSERT INTO $catalog.$ns.part_nested VALUES (1,
named_struct('v', 1.5D))")
- }
- assertUnsupportedContains(writeExec, "part_nested", "partition field
s.v", "double", "-0.0")
- }
- }
-
- test("fall-back: double identity partition beside a dropped partition
field") {
- // A format-version-1 spec keeps a dropped partition field as a `void`
transform, and that
- // field's source column can be dropped afterwards. The surviving double
field must still be
- // found, whatever the dropped one does to the spec's partition type.
- withDetectionCatalog { dir =>
- createTable(
- dir,
- "part_dropped",
- partitionSpec = "PARTITIONED BY (region, amount)",
- properties = Some("'format-version'='1'"))
- // Loaded afresh for each change: the insert in between commits through
another handle.
- def table: org.apache.iceberg.Table =
- loadIcebergTable(spark, catalog, ns, "part_dropped")
- .asInstanceOf[org.apache.iceberg.Table]
- table.updateSpec().removeField("region").commit()
- spark.sql(s"REFRESH TABLE $catalog.$ns.part_dropped")
- assertUnsupportedContains("part_dropped", "partition field amount",
"double", "-0.0")
-
- // Iceberg before 1.11 cannot plan a write once the `void` field's
source column is gone.
- // On 1.11 iceberg-java plans it but cannot build a partition key for a
spec that mixes that
- // field with a live one, so the write itself fails on either path; only
the gate's decision
- // is checked.
- if (icebergVersionAtLeast(1, 11)) {
- table.updateSchema().deleteColumn("region").commit()
- spark.sql(s"REFRESH TABLE $catalog.$ns.part_dropped")
- val writeExec = captureWriteExec("part_dropped", allowWriteFailure =
true) {
- spark.sql(s"INSERT INTO $catalog.$ns.part_dropped VALUES (2, 2.0)")
- }
- assertUnsupportedContains(
- writeExec,
- "part_dropped",
- "partition field amount",
- "double",
- "-0.0")
- }
- }
- }
-
- test("Compatible when a dropped double partition field remains as void") {
- // The `void` field only ever holds null, so there are no signed zeros to
keep apart.
+ test("Compatible when partitioned by identity on a double column") {
Review Comment:
Thanks for lifting this, and the signed-zero test is really thorough.
--
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]
