sunchao commented on code in PR #5729:
URL: https://github.com/apache/datafusion-comet/pull/5729#discussion_r3942777915
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -1033,6 +1033,124 @@ class CometIcebergWriteActionSuite
}
}
+ // iceberg-java renders a `timestamptz` partition value with
+ // `DateTimeUtil.microsToIsoTimestamptz` and a `binary` one with base64;
iceberg-rust's
+ // `partition_to_path` renders the first as a `chrono` `DateTime<Utc>`
(space separator, " UTC"
+ // suffix) -- and outright panics on a pre-1970 value with a sub-second
part, since
+ // `microseconds_to_datetimetz` casts a negative remainder to `u32`
+ // (apache/datafusion-comet#5694) -- and the second as uppercase hex. Comet
renders the path
+ // itself; this pins the result against the layout iceberg-java's own writer
produces.
+ test("native acceleration: timestamptz and binary partition paths match
iceberg-java") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ // A UTC session zone makes the stored micros of each literal exact, so
the expected
+ // directory names below are not a function of the machine's zone.
+ withSQLConf("spark.sql.session.timeZone" -> "UTC") {
+ Seq("ts_path_native", "ts_path_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (id INT, ts TIMESTAMP, bin BINARY)
+ USING iceberg PARTITIONED BY (ts, bin)
+ """)
+ }
+ // Pre-epoch with a sub-second part (the panic case), pre-epoch on a
whole second, the
+ // epoch itself, and a post-epoch microsecond value.
+ val values =
+ "(1, TIMESTAMP '1969-12-31 23:59:58.5', X'0001FF'), " +
+ "(2, TIMESTAMP '1969-12-31 23:59:58', X'00'), " +
+ "(3, TIMESTAMP '1970-01-01 00:00:00', X''), " +
+ "(4, TIMESTAMP '2024-04-01 19:25:00.123456', X'FF')"
+
+ assertNativeWriteEngages("ts_path_native", Seq(1, 2, 3, 4)) {
+ spark.sql(s"INSERT INTO $catalog.$ns.ts_path_native VALUES $values")
+ }
+ spark.sql(s"INSERT INTO $catalog.$ns.ts_path_jvm VALUES $values")
+
+ val nativeDirs = partitionDirs(warehouseDir, "ts_path_native")
+ assert(nativeDirs == partitionDirs(warehouseDir, "ts_path_jvm"),
s"native: $nativeDirs")
Review Comment:
### Correctness
[P2] Make timestamp-path parity conditional on the Iceberg formatter version
This equality fails in the supported Spark 3.4 profile, which pins Iceberg
1.5.2. Its `TransformUtil.humanTimestampWithZone` uses
`OffsetDateTime.toString()`, so the JVM writes `1969-12-31T23:59:58.500Z` and
`1970-01-01T00:00Z`, while this patch writes `1969-12-31T23:59:58.5+00:00` and
`1970-01-01T00:00:00+00:00`. The [Spark 3.4 scan
job](https://github.com/apache/datafusion-comet/actions/runs/33998881999/job/101397610298)
fails at this assertion with exactly those sets. Make the byte-for-byte JVM
comparison version-aware, while retaining the pre-epoch write and readback
checks on 3.4, and qualify the corresponding documentation claim. The older
layout difference does not itself indicate incorrect stored values.
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -1033,6 +1033,124 @@ class CometIcebergWriteActionSuite
}
}
+ // iceberg-java renders a `timestamptz` partition value with
+ // `DateTimeUtil.microsToIsoTimestamptz` and a `binary` one with base64;
iceberg-rust's
+ // `partition_to_path` renders the first as a `chrono` `DateTime<Utc>`
(space separator, " UTC"
+ // suffix) -- and outright panics on a pre-1970 value with a sub-second
part, since
+ // `microseconds_to_datetimetz` casts a negative remainder to `u32`
+ // (apache/datafusion-comet#5694) -- and the second as uppercase hex. Comet
renders the path
+ // itself; this pins the result against the layout iceberg-java's own writer
produces.
+ test("native acceleration: timestamptz and binary partition paths match
iceberg-java") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ // A UTC session zone makes the stored micros of each literal exact, so
the expected
+ // directory names below are not a function of the machine's zone.
+ withSQLConf("spark.sql.session.timeZone" -> "UTC") {
+ Seq("ts_path_native", "ts_path_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (id INT, ts TIMESTAMP, bin BINARY)
+ USING iceberg PARTITIONED BY (ts, bin)
+ """)
+ }
+ // Pre-epoch with a sub-second part (the panic case), pre-epoch on a
whole second, the
+ // epoch itself, and a post-epoch microsecond value.
+ val values =
+ "(1, TIMESTAMP '1969-12-31 23:59:58.5', X'0001FF'), " +
+ "(2, TIMESTAMP '1969-12-31 23:59:58', X'00'), " +
+ "(3, TIMESTAMP '1970-01-01 00:00:00', X''), " +
+ "(4, TIMESTAMP '2024-04-01 19:25:00.123456', X'FF')"
+
+ assertNativeWriteEngages("ts_path_native", Seq(1, 2, 3, 4)) {
+ spark.sql(s"INSERT INTO $catalog.$ns.ts_path_native VALUES $values")
+ }
+ spark.sql(s"INSERT INTO $catalog.$ns.ts_path_jvm VALUES $values")
+
+ val nativeDirs = partitionDirs(warehouseDir, "ts_path_native")
+ assert(nativeDirs == partitionDirs(warehouseDir, "ts_path_jvm"),
s"native: $nativeDirs")
+ assert(
+
nativeDirs.contains("ts=1969-12-31T23%3A59%3A58.5%2B00%3A00/bin=AAH%2F"),
+ s"native: $nativeDirs")
+ assert(
+ nativeDirs.contains("ts=1970-01-01T00%3A00%3A00%2B00%3A00/bin="),
+ s"native: $nativeDirs")
+ assert(
+
nativeDirs.contains("ts=2024-04-01T19%3A25%3A00.123456%2B00%3A00/bin=%2Fw%3D%3D"),
+ s"native: $nativeDirs")
+
+ // Values round-trip through both readers regardless of how the path
was spelled.
+ Seq("true", "false").foreach { cometEnabled =>
+ withSQLConf(CometConf.COMET_ENABLED.key -> cometEnabled) {
+ val rows =
+ spark.sql(s"SELECT id, ts FROM $catalog.$ns.ts_path_native ORDER
BY id").collect()
+ assert(
+ rows.toSeq ==
+ spark
+ .sql(s"SELECT id, ts FROM $catalog.$ns.ts_path_jvm ORDER BY
id")
+ .collect()
+ .toSeq,
+ s"comet=$cometEnabled: $rows")
+ }
+ }
+ }
+ }
+ }
+
+ // iceberg-java's `UpdatePartitionSpec` keeps a dropped partition field in a
format-version-1
+ // spec as a `void` transform so its field id survives, and
`PartitionSpec#isUnpartitioned` is
+ // "every field is void", not "no fields". The next write therefore runs
through the
+ // unpartitioned writer -- which stamps an empty partition struct -- against
a spec whose fields
+ // are non-empty, and iceberg-rust's `ManifestWriter` used to `zip_eq` the
two and panic across
+ // the JNI boundary (apache/datafusion-comet#5691). Dropping the source
column afterwards then
+ // broke the manifest's `partition_type` resolution as well
(apache/datafusion-comet#5693).
+ test("native acceleration: writes after a V1 partition field is dropped
match iceberg-java") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ Seq("evolved_native", "evolved_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (id INT, region STRING)
+ USING iceberg TBLPROPERTIES ('format-version'='1')
+ """)
+ }
+
+ def evolve(table: String, write: (String, Seq[Int]) => Unit): Unit = {
+ write("(1, 'us')", Seq(1))
+ spark.sql(s"ALTER TABLE $catalog.$ns.$table ADD PARTITION FIELD region
AS region_part")
+ write("(2, 'eu')", Seq(1, 2))
+ spark.sql(s"ALTER TABLE $catalog.$ns.$table DROP PARTITION FIELD
region_part")
+ // The spec is now void-only: the #5691 shape.
+ write("(3, 'ap')", Seq(1, 2, 3))
+ spark.sql(s"ALTER TABLE $catalog.$ns.$table DROP COLUMN region")
+ // The void field's source column is gone: the #5693 shape.
+ write("(4)", Seq(1, 2, 3, 4))
Review Comment:
### Correctness
[P2] Separate the dropped-source regression from runtimes that cannot commit
it
The unconditional final stage fails in every Spark 3.4, 3.5 and 4.0 scan
job. Their pinned Iceberg 1.5.2/1.8.1 runtimes fail resolving the dropped
source in `PartitionSpec.partitionType`; 1.10.0 reaches commit but fails in
`PartitionSpec.javaClasses` because the transform result type is null. The
[Spark 4.0
failure](https://github.com/apache/datafusion-comet/actions/runs/33998881999/job/101397610203)
traces through `SnapshotSummary` and `IcebergCommitExec`, so normalizing the
native transport manifest does not avoid this JVM failure. Iceberg 1.11 handles
the missing source, and the new test passes in the Spark 4.1 job. Split the
source-column-drop scenario behind the supported Iceberg-version condition
while keeping the preceding void-only writes covered on older profiles;
otherwise this regression test leaves four scan jobs failing.
--
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]
