SEPURI-SAI-KRISHNA opened a new pull request, #59583:
URL: https://github.com/apache/spark/pull/59583
### What changes were proposed in this pull request?
Before a write, the values of `FLOAT` and `DOUBLE` dynamic partition columns
are now normalized as grouping keys are (`NormalizeFloatingNumbers.normalize`):
-0.0 becomes 0.0 and every NaN becomes the same NaN. This happens in the
projection that already converts empty strings to null in string partition
columns, so it applies with planned writes (`V1Writes`) and without them
(`FileFormatWriter`). `V1WritesUtils.convertEmptyToNull` is renamed to
`normalizePartitionValues` to match.
Static partition columns are left as they are. Their directory is named by
the partition spec, and an `INSERT OVERWRITE ... PARTITION (p = '-0.0')` must
replace the directory it named, not write to another one.
`AliasAwareQueryOutputOrdering.strip` now looks through the normalization,
as it already does through `Empty2Null`. The normalization keeps the values in
order, so rows already sorted by a partition column are not sorted again.
### Why are the changes needed?
A partitioned write fails when a floating-point partition column holds both
0.0 and -0.0 in one task, and nothing is written. With the default
configuration, on 3.5.3, 4.0.3 and 4.2.0 (official binaries) and on master:
```sql
CREATE TABLE t (id INT, d DOUBLE) USING parquet PARTITIONED BY (d);
INSERT INTO t SELECT /*+ COALESCE(1) */ * FROM VALUES
(1, 0.0D), (2, -0.0D), (3, 0.0D), (4, 1.5D) AS v(id, d);
-- [TASK_WRITE_FAILED] Task failed while writing rows to file:/.../t.
-- Caused by: org.apache.hadoop.fs.FileAlreadyExistsException: File already
exists:
--
file:/.../t/_temporary/0/_temporary/attempt_.../d=0.0/part-00000-<uuid>.c000.snappy.parquet
SELECT count(*) FROM t; -- 0
```
The same happens with ORC, CSV and JSON, `INSERT OVERWRITE`, CTAS,
`df.write.partitionBy`, `FLOAT`, other partition columns before it, and Hive
tables (with and without `spark.sql.hive.convertMetastoreParquet`).
The write sorts the rows of a task by the dynamic partition columns, where
-0.0 = 0.0 and all NaNs are equal, so the zeros can stay interleaved: 0.0,
-0.0, 0.0. `DynamicPartitionDataSingleWriter` starts a new directory whenever
the bytes of the partition values change, so it writes `d=0.0`, then `d=-0.0`,
then `d=0.0` again, and its second file in `d=0.0` has the same name as the
first. NaNs with different bits, e.g. `NaN` and `-NaN`, fail the same way on
`d=NaN`. With `spark.sql.maxConcurrentOutputFileWriters`, the writers are keyed
by those bytes, so NaNs with different bits open the same `d=NaN` file, and the
zeros fail once the writers fall back to sorting.
When the zeros are in different tasks the write succeeds but creates both
`d=0.0` and `d=-0.0`, so where a row ends up depends on how the rows were split
into tasks. The file stream sink names every file with a random UUID, so it
does not fail, but it writes both `d=0.0` and `d=-0.0` even within one task.
Spark already normalizes these values in grouping, join and window partition
keys (SPARK-26021) and in array set operations (SPARK-54918). Dynamic partition
values are grouping keys too. SPARK-26012 fixed the same failure for null and
empty strings, which both go to the default partition, with the `Empty2Null`
conversion that this PR extends.
### Does this PR introduce _any_ user-facing change?
Yes. A write whose `FLOAT` or `DOUBLE` dynamic partition column holds 0.0
and -0.0, or NaNs with different bits, in one task now succeeds, where it
failed before. -0.0 is now written to the `0.0` partition, where it could go to
a `-0.0` partition before, depending on how the rows were split into tasks.
With `spark.sql.sources.partitionOverwriteMode=dynamic`, a dynamic -0.0 now
overwrites the `0.0` partition. Static partition values are written as before.
### How was this patch tested?
New tests:
- `V1WriteCommandSuite`, "v1 write with 0.0, -0.0 and NaN partition column
values": writes 0.0, -0.0, `NaN` and `-NaN` interleaved in one task, for
`DOUBLE` and `FLOAT`, with and without planned writes, and with
`maxConcurrentOutputFileWriters` 0 (sorted), 1 (concurrent writer falling back
to sorting) and 4 (concurrent writers only). It checks that the directories are
`p=0.0`, `p=1.0` and `p=NaN` and reads the rows back. It also checks that
overwriting a static partition `p = '-0.0'` twice leaves only the rows of the
second write.
- `V1WriteCommandSuite`, "v1 write with already sorted plan - floating-point
partition column": `INSERT ... ORDER BY` a `DOUBLE` partition column plans a
single sort.
- `V1WriteHiveCommandSuite`: an insert into a Hive table partitioned by a
static `STRING` column and a dynamic `DOUBLE` column, with and without
converting the metastore table.
The first test fails without this change. Each part of the change was also
undone on its own, and each variant fails one of the new tests: no
normalization in `V1Writes`, no normalization in `FileFormatWriter`, static
columns normalized in either of them (the second overwrite appends), and
`strip` not looking through the normalization (a second sort is planned).
These suites pass: `V1WriteCommandSuite`, `PartitionedWriteSuite`,
`InsertSuite`, `FileFormatWriterSuite`, `BucketedWriteWithoutHiveSupportSuite`,
`ProjectedOrderingAndPartitioningSuite`, `DataFrameReaderWriterSuite`,
`FileStreamSinkV1Suite`, `FileStreamSinkV2Suite` and `PlannerSuite` in
`sql/core`, and `V1WriteHiveCommandSuite`, `InsertSuite`,
`HiveSQLInsertTestSuite`, `PartitionProviderCompatibilitySuite`,
`BucketedWriteWithHiveSupportSuite` and `HiveParquetSourceSuite` in `sql/hive`.
A separate matrix wrote 0.0 and -0.0, NaNs with different bits, and NULL
interleaved in one task: Parquet, ORC, CSV and JSON, `FLOAT` and `DOUBLE`,
planned writes on and off, `maxConcurrentOutputFileWriters` 0, 1 and 8, with
and without another partition column before it. 160 of 180 writes failed before
this change, and the other 20 created both `p=0.0` and `p=-0.0`. After it, all
180 succeed with one directory per value, and static partition specs write the
same directories as before.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
--
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]
