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]

Reply via email to