morningman opened a new pull request, #69845:
URL: https://github.com/apache/doris/pull/69845
### What problem does this PR solve?
Issue Number: None
Related PR: #69711
Problem Summary:
**In short.** An Iceberg `timestamptz` column stored as Parquet INT96, the
encoding Spark and Hive use for timestamps, reads shifted by the session time
zone's offset in any non-UTC session: in Asia/Shanghai, 2024-01-01T00:00:00Z
reads as `2024-01-01 00:00:00+08:00`, eight hours early, while Iceberg itself
reads `2024-01-01 08:00:00+08:00`. Equality deletes on such a key miss their
rows for the same reason. This PR has the FE mark each `timestamptz` field of
the scan's schema dictionary, and FileScannerV2 then decodes the matching
Parquet columns of data files and equality-delete files as instants, which is
how Iceberg reads them. It came up in the review of #69711.
**Background.**
- Parquet INT96 has no logical annotation, so a reader cannot tell from the
file whether an INT96 column holds an instant or a wall clock. Spark writes its
`timestamp` type (an instant) as INT96 by default, and `add_files`, `migrate`
and `snapshot` attach such files to Iceberg tables unchanged, as `timestamptz`
columns.
- Iceberg does not look at the file annotation at all: its Parquet readers
choose the timestamp reader from the table type, so a `timestamptz` field reads
INT96, and INT64 with or without the UTC annotation, as an instant.
- The FE sends `parquet_timestamp_semantics_version = 1` for every file
scan. Parquet scans therefore always use FileScannerV2, and INT96 decodes as a
UTC wall clock (`DATETIMEV2`) unless the table format supplies the semantics.
Paimon has the same ambiguity (TIMESTAMP and TIMESTAMP_LTZ share INT96) and
solves it with `timestamp_is_adjusted_to_utc` on each field of its schema
dictionary, which `PaimonReader` applies to the file schema. The Iceberg FE
never set that field.
**The problem, and what it cost.** For a `timestamptz` column, the
projection casts the INT96 wall clock to `TIMESTAMPTZ` in the session time
zone, so the instant moves by the session offset. Reproduced on master with an
INT96 file attached like `add_files` and an Iceberg-written INT64 file holding
the same instants, both in one table:
| Session | Rows | Iceberg Java | Before | After |
|---|---|---|---|---|
| UTC | INT96 and INT64 | 2024-01-01 00:00:00 UTC | correct | correct |
| Asia/Shanghai | INT64 (written by Iceberg) | `2024-01-01 08:00:00+08:00` |
correct | correct |
| Asia/Shanghai | INT96 (written by Spark) | `2024-01-01 08:00:00+08:00` |
`2024-01-01 00:00:00+08:00`, 8 hours early | `2024-01-01 08:00:00+08:00` |
The result is the same with field ids in the file, with only the table's
name mapping, under either value of the catalog's
`enable.mapping.timestamp_tz`, and with `enable_file_scanner_v2` on or off. The
same shift breaks equality deletes, which compare the projected values:
| Equality delete on a `timestamptz` key, non-UTC session | Before | After |
|---|---|---|
| INT96 data key, Iceberg-written tombstone of the same instant | the
tombstone misses the row | the row is deleted |
| Catalog with `enable.mapping.timestamp_tz = false`, INT64 data and
tombstones, America/New_York, a tombstone at one of the two 01:30 of the DST
fall-back | both rows at 01:30 are deleted (both sides read as session wall
clocks) | only the row with the tombstone's instant is deleted |
Wrong values and missed or extra deletes come back without any error.
**How this PR fixes it.**
- FE: `IcebergSchemaUtils.buildField` sets `timestamp_is_adjusted_to_utc =
true` on every `timestamptz` field, at every nesting level; the dictionary also
carries the hidden equality-delete keys. `timestamp` fields stay unmarked:
marking them `false` would make BE read the Parquet statistics of an
adjusted-UTC INT64 file as UTC wall clocks while its rows still decode as
session wall clocks, so row-group pruning could drop matching rows.
- BE, data files: `IcebergTableReader::annotate_file_schema` copies the
marker onto the Parquet file column that `TableColumnMapper` will select for
each field, using the mapper's own matchers (`find_column_by_field_id` /
`find_column_by_name`) after the positional fallback ids are stamped, so field
ids, the table's name mapping and positional ids all work. It then rewrites the
column type with the helper `PaimonReader` already used, moved to
`schema_history_util` as `apply_file_timestamp_semantics`. The mapping, the
scan request and `ParquetReader` all see `TIMESTAMPTZ`, and INT96 decodes as an
instant. Only a `DATETIMEV2` or `TIMESTAMPTZ` leaf takes the marker; a
`TIMESTAMP_NS` leaf keeps its nanoseconds. Columns without a marked leaf keep
the types the file reader reported, and a plan from an older FE carries no
marker and reads as before.
- BE, equality deletes: a hidden key column has no `ColumnMapping`, so its
projection now carries the marker of the annotated file column
(`attach_timestamp_semantics`, exported from the mapper); without it Parquet
would decode the key as `DATETIMEV2` under a `TIMESTAMPTZ` file schema.
Equality-delete files get the same annotation, so tombstones and data keys of a
`timestamptz` field are both instants. The `DATETIMEV2` / `TIMESTAMPTZ`
comparison rule from #69711 stays for plans from older FEs.
- Refactor: `find_current_schema` and
`build_schema_identity_from_external_field` (now also carrying the marker) move
to `schema_history_util`; `TableReader`'s four copies of the current-schema
lookup use the helper.
**Results.**
- FE unit test
`IcebergSchemaUtilsTest#timestamptzFieldsCarryInstantMarkerAtEveryLevel`:
top-level, struct, list and map `timestamptz` fields are marked; `timestamp`
and `timestamp_ns` fields are not, under both values of the mapping flag.
- 5 new BE unit tests in `iceberg_reader_test.cpp`, all failing without the
BE change: the file-schema annotation (field ids at every level, `TIMESTAMP_NS`
untouched, ORC untouched); INT96 `timestamptz` in Asia/Shanghai with field ids,
name mapping and positional ids, with and without the marker; INT96 inside a
struct and a list; an INT96 equality key against an Iceberg tombstone, hidden
or projected; and the DST fall-back above. Leaving the delete files unannotated
fails the DST case (the second tombstone deletes the first row); projecting a
hidden key or a tombstone without its marker fails both equality-delete cases
with a read error; marking a `TIMESTAMP_NS` leaf fails the annotation case.
With neither data nor delete files annotated (the reader before this PR), the
INT96 key deletes nothing and the DST case deletes both rows. The existing case
for an older FE's plan (no marker) still passes.
- End to end on a local cluster: an Iceberg table holding an INT96 Parquet
file attached like `add_files` (one table with field ids in the file, one
relying on `schema.name-mapping.default`) and an Iceberg-written INT64 file
with the same instants, read through three catalogs
(`enable.mapping.timestamp_tz` unset, `true`, `false`), with
`enable_file_scanner_v2` on and off, in UTC and Asia/Shanghai: 24 combinations.
Before this PR the INT96 rows were 8 hours early in every Asia/Shanghai
combination (8 of the 16 measured); with it the INT96 rows equal the INT64 rows
in all 24, and the INT64 rows and the UTC results are unchanged.
branch-4.1 and branch-4.2 have the same code.
**Classes, and how they call each other.**
- `IcebergSchemaUtils` (FE): builds the schema dictionary; now marks
`timestamptz` fields.
- `schema_history_util`: `find_current_schema`,
`build_schema_identity_from_external_field`, and
`apply_file_timestamp_semantics` (moved from `PaimonReader`).
- `IcebergTableReader`: `annotate_file_schema` →
`_apply_positional_id_fallback` (the existing logic) and
`_apply_table_timestamp_semantics`; `_append_equality_delete_predicates` and
`_load_equality_delete_file` project keys with their markers.
- `column_mapper`: `attach_timestamp_semantics` is exported.
```
FE IcebergSchemaUtils.buildField --timestamptz-->
TField.timestamp_is_adjusted_to_utc = true
(history_schema_info,
current_schema_id = -1)
BE TableReader::open_reader
+-- IcebergTableReader::annotate_file_schema
(Parquet only)
| +-- _apply_positional_id_fallback
| +-- _apply_table_timestamp_semantics
| +-- find_current_schema,
build_schema_identity_from_external_field
| +-- find_column_by_field_id | find_column_by_name (the
mapper's matchers)
| +-- apply_file_timestamp_semantics
(shared with PaimonReader)
+-- TableColumnMapper: mapping/projection take the marker from the
file column
+-- ParquetReader::open: apply_projection_timestamp_semantics -> INT96
decodes as TIMESTAMPTZ
IcebergTableReader::_append_equality_delete_predicates ->
attach_timestamp_semantics (hidden key)
IcebergTableReader::_load_equality_delete_file ->
_apply_table_timestamp_semantics
+
attach_timestamp_semantics (tombstones)
```
ORC files are not changed: ORC's file type already separates `TIMESTAMP`
from `TIMESTAMP_INSTANT`.
### Release note
Fix Iceberg `timestamptz` columns stored as Parquet INT96 (written by Spark
or Hive and attached with `add_files`, `migrate` or `snapshot`) reading shifted
by the session time zone's offset in non-UTC sessions, and equality deletes on
such keys missing their rows.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] Unit Test
- BE: `run-be-ut.sh --run
--filter='IcebergV2ReaderTest.*:Paimon*ReaderTest.*:Hudi*Test.*:TableReaderTest.*:ColumnMapper*:FlussUnionLakeReaderTest.*:EqualityDeletePredicateTest.*:*PositionDelete*'`:
532 of 532 pass on macOS arm64 (ASAN), including the 5 new cases. They ran on
a local base; the commit was then applied to master `2a4191c1dc9` without
conflicts, and the changed sources and their test files were syntax-checked
against that tree with the UT build flags.
- FE: `mvn -pl :fe-connector-iceberg -am package
-Dtest=IcebergSchemaUtilsTest`: 35 of 35 pass, checkstyle passes.
- [x] Manual test (add detailed scripts or steps below)
- The local end-to-end check under Results, with an ASAN BE and the
Iceberg connector built from this change.
- [ ] No need to test or manual test. Explain why:
- Behavior changed:
- [ ] No.
- [x] Yes. Parquet columns of Iceberg `timestamptz` fields decode as
instants by the table type, as Iceberg reads them: INT96 no longer shifts by
the session offset, an INT64 timestamp without the UTC annotation under a
`timestamptz` field reads its value as UTC, and equality deletes on
`timestamptz` keys compare instants even when `enable.mapping.timestamp_tz` is
false. Plans from an older FE are unchanged.
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should
merge into -->
--
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]
