This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a9253e979d7 [fix](iceberg) Decode Parquet timestamptz columns as
instants by the Iceberg table type (#69845)
a9253e979d7 is described below
commit a9253e979d7048e81a1bb34f9b77c5176f340996
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Nov 2 01:30:52 2026 +0000
[fix](iceberg) Decode Parquet timestamptz columns as instants by the
Iceberg table type (#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, whatever the catalog's `enable.mapping.timestamp_tz` says
(master always maps `timestamptz` to TIMESTAMPTZ), 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 |
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. Like the mapper, it accepts an ID-less wrapper
at every depth when `allow_idless_complex_wrapper_projection` is on. 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: three kinds of column are read without a
`ColumnMapping`: a hidden key column, the parent struct that is read for
the NULLs of a nested key the data file lacks, and the key columns of an
equality-delete file. Their projections now carry the markers of the
annotated file column (`equality_delete_root_projection`, built on
`attach_timestamp_semantics`, which the mapper now exports). Without
them Parquet would decode an INT96 leaf as `DATETIMEV2` under a
`TIMESTAMPTZ` file schema, and the read would fail its type check. That
holds even when the query projects the column, because reading it as a
predicate column replaces the projection its mapping marked.
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.
- 2 more BE unit tests from review:
- INT96 under a nested ID-less wrapper struct, read and used as an
equality key. It fails when the walk accepts the wrapper only at the
root.
- A nested equality key missing from an older file whose parent struct
holds an INT96 `timestamptz`. It checks the survivors with the parent
hidden and projected, that the projected timestamp reads as an instant
in Asia/Shanghai, and a plan without the marker. It fails with the
reader's type check when the parent is read without the marker.
- 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`
(hidden keys), `_build_missing_equality_delete_key_expr` (the parent of
a missing nested key) and `_load_equality_delete_file` (tombstones)
project their columns with the markers through
`equality_delete_root_projection`.
- `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
+-- equality_delete_root_projection -> attach_timestamp_semantics
(hidden key)
+-- _build_missing_equality_delete_key_expr
+-- equality_delete_root_projection
(parent of a missing key)
IcebergTableReader::_load_equality_delete_file ->
_apply_table_timestamp_semantics
+
equality_delete_root_projection (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. After
the first review round, `IcebergV2ReaderTest` passes 122 of 122,
including the new nested ID-less wrapper case. It fails when the walk
accepts the wrapper only at the root. After the second review round, the
whole filter passes 534 of 534 and `IcebergV2ReaderTest` 123 of 123,
including the new missing nested key case. It fails when the missing
key's parent is read without the marker. 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. 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 -->
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
be/src/format_v2/column_mapper.cpp | 6 +-
be/src/format_v2/column_mapper.h | 8 +
be/src/format_v2/table/iceberg_reader.cpp | 166 +++-
be/src/format_v2/table/iceberg_reader.h | 35 +-
be/src/format_v2/table/paimon_reader.cpp | 57 +-
be/src/format_v2/table/schema_history_util.cpp | 117 +++
be/src/format_v2/table/schema_history_util.h | 18 +
be/src/format_v2/table_reader.cpp | 95 +--
be/test/format_v2/table/iceberg_reader_test.cpp | 865 ++++++++++++++++++++-
.../connector/iceberg/IcebergSchemaUtils.java | 12 +-
.../connector/iceberg/IcebergSchemaUtilsTest.java | 42 +
11 files changed, 1248 insertions(+), 173 deletions(-)
diff --git a/be/src/format_v2/column_mapper.cpp
b/be/src/format_v2/column_mapper.cpp
index 2126f78478d..e8e275612d4 100644
--- a/be/src/format_v2/column_mapper.cpp
+++ b/be/src/format_v2/column_mapper.cpp
@@ -1957,9 +1957,9 @@ static bool has_timestamp_semantics(const
ColumnDefinition& column) {
std::ranges::any_of(column.children, has_timestamp_semantics);
}
-static void attach_timestamp_semantics(const std::optional<bool>&
timestamp_is_adjusted_to_utc,
- const std::vector<ColumnDefinition>&
file_children,
- LocalColumnIndex* projection) {
+void attach_timestamp_semantics(const std::optional<bool>&
timestamp_is_adjusted_to_utc,
+ const std::vector<ColumnDefinition>&
file_children,
+ LocalColumnIndex* projection) {
DORIS_CHECK(projection != nullptr);
projection->timestamp_is_adjusted_to_utc = timestamp_is_adjusted_to_utc;
for (const auto& file_child : file_children) {
diff --git a/be/src/format_v2/column_mapper.h b/be/src/format_v2/column_mapper.h
index 41c8cc953f7..0b4c1bfed0d 100644
--- a/be/src/format_v2/column_mapper.h
+++ b/be/src/format_v2/column_mapper.h
@@ -196,6 +196,14 @@ const ColumnDefinition* find_column_by_name(const
ColumnDefinition& table_column
const ColumnDefinition* find_column_by_field_id(const ColumnDefinition&
table_column,
const
std::vector<ColumnDefinition>& file_schema,
bool
allow_idless_complex_wrapper_projection);
+// Put the timestamp semantics recorded on an annotated file column and its
descendants onto a
+// projection of that column. A full projection gains a child only for a
subtree that carries a
+// semantic; a partial projection keeps its selection. Callers that project a
file column without
+// a ColumnMapping, such as a hidden equality-delete key, must call this or
the file reader keeps
+// the physical type while the annotated file schema expects the table-format
type.
+void attach_timestamp_semantics(const std::optional<bool>&
timestamp_is_adjusted_to_utc,
+ const std::vector<ColumnDefinition>&
file_children,
+ LocalColumnIndex* projection);
// Generic mapping layer from table schema to file schema.
// Iceberg uses BY_FIELD_ID. Plain by-name formats can reuse this component as
well, so keep this
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index 9463fe0d9b0..82d086d417e 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -165,6 +165,73 @@ const char* file_format_name(FileFormat format) {
return "UNKNOWN";
}
+const format::ColumnDefinition* find_matched_file_column(
+ const format::ColumnDefinition& table_column,
+ const std::vector<format::ColumnDefinition>& file_columns,
+ format::TableColumnMappingMode mode, bool
allow_idless_complex_wrapper_projection) {
+ return mode == format::TableColumnMappingMode::BY_FIELD_ID
+ ? format::find_column_by_field_id(table_column,
file_columns,
+
allow_idless_complex_wrapper_projection)
+ : format::find_column_by_name(table_column, file_columns);
+}
+
+// Copy the table field's timestamp marker onto the file column
TableColumnMapper selects for it,
+// descending the way the mapper does: ARRAY and MAP children by position,
STRUCT children by the
+// mapping mode, through an ID-less wrapper at any depth when the mapper
allows one. Only a
+// DATETIMEV2 or TIMESTAMPTZ leaf takes the marker. A TIMESTAMP_NS leaf (an
INT64 NANOS column
+// without the UTC annotation) keeps its nanoseconds: the Parquet reader
rewrites any marked
+// timestamp leaf, so a marker there would make it decode TIMESTAMPTZ while
the file schema still
+// says TIMESTAMP_NS. Returns whether any leaf took a marker.
+bool annotate_matched_timestamp_semantics(const format::ColumnDefinition&
table_column,
+ format::ColumnDefinition*
file_column,
+ format::TableColumnMappingMode mode,
+ bool
allow_idless_complex_wrapper_projection) {
+ DORIS_CHECK(file_column != nullptr);
+ DORIS_CHECK(file_column->type != nullptr);
+ switch (remove_nullable(file_column->type)->get_primitive_type()) {
+ case TYPE_DATETIMEV2:
+ case TYPE_TIMESTAMPTZ:
+ file_column->timestamp_is_adjusted_to_utc =
table_column.timestamp_is_adjusted_to_utc;
+ return file_column->timestamp_is_adjusted_to_utc.has_value();
+ case TYPE_ARRAY:
+ DORIS_CHECK(file_column->children.size() == 1);
+ // FE omits nested metadata for a field it describes only by its root.
+ return table_column.children.size() == 1 &&
+ annotate_matched_timestamp_semantics(table_column.children[0],
+ &file_column->children[0],
mode,
+
allow_idless_complex_wrapper_projection);
+ case TYPE_MAP: {
+ DORIS_CHECK(file_column->children.size() == 2);
+ if (table_column.children.size() != 2) {
+ return false;
+ }
+ const bool key = annotate_matched_timestamp_semantics(
+ table_column.children[0], &file_column->children[0], mode,
+ allow_idless_complex_wrapper_projection);
+ const bool value = annotate_matched_timestamp_semantics(
+ table_column.children[1], &file_column->children[1], mode,
+ allow_idless_complex_wrapper_projection);
+ return key || value;
+ }
+ case TYPE_STRUCT: {
+ bool annotated = false;
+ for (const auto& table_child : table_column.children) {
+ // TableColumnMapper falls back to an ID-less wrapper below the
root as well.
+ const auto* matched = find_matched_file_column(table_child,
file_column->children, mode,
+
allow_idless_complex_wrapper_projection);
+ if (matched != nullptr) {
+ annotated |= annotate_matched_timestamp_semantics(
+ table_child, &file_column->children[matched -
file_column->children.data()],
+ mode, allow_idless_complex_wrapper_projection);
+ }
+ }
+ return annotated;
+ }
+ default:
+ return false;
+ }
+}
+
} // namespace
bool IcebergTableReader::_scan_has_any_authoritative_name_mapping() const {
@@ -212,20 +279,28 @@ Status IcebergTableReader::annotate_file_schema(
// Iceberg's positional fallback is a Parquet-reader-only compatibility
path. Its ORC/Avro
// readers instead derive a name mapping from the expected schema for
id-less files, which is
// exactly the current-name resurrection that v2 scan semantics
deliberately rejects (#65784),
- // so ORC keeps its existing behavior and stays out of scope here.
+ // so ORC keeps its existing behavior and stays out of scope here. ORC
needs no timestamp
+ // marker either: its file type already separates TIMESTAMP from
TIMESTAMP_INSTANT.
if (_format != FileFormat::PARQUET) {
return Status::OK();
}
+ _apply_positional_id_fallback(file_schema);
+ return _apply_table_timestamp_semantics(file_schema,
_mapping_mode(*file_schema));
+}
+
+void IcebergTableReader::_apply_positional_id_fallback(
+ std::vector<format::ColumnDefinition>* file_schema) {
+ DORIS_CHECK(file_schema != nullptr);
// Iceberg's `hasIds` contract is existential: a single real field id
keeps the file on the id
// path and, exactly as in Iceberg, skips both the name mapping and the
positional fallback.
if (schema_has_any_field_id(*file_schema)) {
- return Status::OK();
+ return;
}
// An old-FE plan keeps the legacy name fallback (rolling upgrade), and an
explicit table-level
// name mapping stays authoritative even when it does not cover this file.
if (!supports_iceberg_scan_semantics_v2(_scan_params) ||
_scan_has_any_authoritative_name_mapping()) {
- return Status::OK();
+ return;
}
// ParquetSchemaUtil.addFallbackIds: ordinals start at 1, are counted over
the physical
// top-level fields in file order, and are not propagated to nested
fields. Reader-synthesized
@@ -245,6 +320,46 @@ Status IcebergTableReader::annotate_file_schema(
field.identifier = Field::create_field<TYPE_INT>(ordinal);
}
_positional_id_fallback = true;
+}
+
+// Iceberg decodes a Parquet timestamp by the table field's type, not by the
file's annotation.
+// Spark and Hive write a timestamptz as INT96, which has no annotation, and
`add_files` / `migrate`
+// attach those files unchanged. Without a marker FileScannerV2 decodes INT96
as a UTC wall clock
+// (DATETIMEV2) and the projection casts that to TIMESTAMPTZ in the session
time zone, so in
+// Asia/Shanghai 2024-01-01T00:00:00Z reads as 2024-01-01 00:00:00+08:00,
eight hours early. FE
+// marks each timestamptz field; copying the marker onto the matched file
column and rewriting its
+// type gives the mapping, the scan request and the equality-delete keys the
instant Iceberg reads.
+// A plan from an older FE carries no marker and keeps the file types.
+Status IcebergTableReader::_apply_table_timestamp_semantics(
+ std::vector<format::ColumnDefinition>* file_schema,
+ format::TableColumnMappingMode mode) const {
+ DORIS_CHECK(file_schema != nullptr);
+ const auto* schema = format::find_current_schema(_scan_params);
+ if (schema == nullptr || !schema->__isset.root_field ||
!schema->root_field.__isset.fields) {
+ return Status::OK();
+ }
+ // As in configure_mapper_options(); only Parquet files get here.
+ const bool allow_idless_complex_wrapper_projection =
+ supports_iceberg_scan_semantics_v1(_scan_params);
+ for (const auto& field_ptr : schema->root_field.fields) {
+ const auto* field = format::get_field_ptr(field_ptr);
+ if (field == nullptr) {
+ continue;
+ }
+ const auto table_column =
format::build_schema_identity_from_external_field(*field);
+ const auto* matched = find_matched_file_column(table_column,
*file_schema, mode,
+
allow_idless_complex_wrapper_projection);
+ if (matched == nullptr) {
+ continue;
+ }
+ auto* file_column = &(*file_schema)[matched - file_schema->data()];
+ // Rebuilding a container recreates its type from the children, so
leave every column
+ // without a marked leaf exactly as the file reader reported it.
+ if (annotate_matched_timestamp_semantics(table_column, file_column,
mode,
+
allow_idless_complex_wrapper_projection)) {
+
RETURN_IF_ERROR(format::apply_file_timestamp_semantics(file_column));
+ }
+ }
return Status::OK();
}
@@ -1075,6 +1190,23 @@ static Status build_equality_delete_key_expr(
return build_nested_equality_delete_key_expr(path, std::move(root_expr),
key_expr);
}
+// Project the root file column of an equality key read without a
ColumnMapping, carrying the
+// timestamp markers annotate_file_schema() and _load_equality_delete_file()
put on the file schema.
+// The file block and the key expressions take the rewritten types from that
schema, but the file
+// reader learns them only from the projection. For example, an older file
holds payload
+// {existing INT, ts INT96 timestamptz} and the table later added payload.k: a
delete on k reads
+// `payload` for its NULLs, and a bare top-level projection makes Parquet read
+// STRUCT<existing, DATETIMEV2(6)> into a STRUCT<existing, TIMESTAMPTZ(6)>
file block, which fails
+// the scan.
+static format::LocalColumnIndex equality_delete_root_projection(
+ const format::ColumnDefinition& root) {
+ auto projection =
+
format::LocalColumnIndex::top_level(format::LocalColumnId(root.file_local_id()));
+ format::attach_timestamp_semantics(root.timestamp_is_adjusted_to_utc,
root.children,
+ &projection);
+ return projection;
+}
+
// Whether a missing key's default is NULL. `missing_path` runs from the
missing root to the key,
// and the root's initial default is walked down it with the key expression's
child matching. The
// root having a default is not enough: a missing `payload` defaulting to the
struct `{}` takes
@@ -1203,6 +1335,9 @@ static Status equality_delete_comparison_type(int
field_id, const DataTypePtr& d
// The projection converts that DATETIMEV2 to the table's TIMESTAMPTZ in
the session time zone
// and the comparison cast does the same, so a tombstone matches exactly
the value a query
// returns: in Asia/Shanghai the key 2024-01-01 08:00:00 compares as
2024-01-01 00:00:00 UTC.
+ // When FE marks timestamptz fields, annotate_file_schema() and
_load_equality_delete_file()
+ // read both Parquet sides as TIMESTAMPTZ already, so for Parquet this
rule now serves plans
+ // from older FEs.
if (remove_nullable(table_type)->get_primitive_type() == TYPE_TIMESTAMPTZ
&&
((delete_primitive == TYPE_DATETIMEV2 && data_primitive ==
TYPE_TIMESTAMPTZ) ||
(delete_primitive == TYPE_TIMESTAMPTZ && data_primitive ==
TYPE_DATETIMEV2))) {
@@ -1919,7 +2054,10 @@ Status
IcebergTableReader::_build_missing_equality_delete_key_expr(
const auto* root = data_path.front();
const auto field_column_id = format::LocalColumnId(root->file_local_id());
- _append_file_scan_column(request, field_column_id,
&request->predicate_columns);
+ // The whole ancestor is read for its NULLs, so a marked timestamp leaf
inside it, such as an
+ // INT96 timestamptz sibling of the missing key, needs its marker as well.
+
RETURN_IF_ERROR(format::FileScanRequestBuilder(request).add_predicate_column(
+ equality_delete_root_projection(*root)));
const auto block_position =
request->local_positions.at(field_column_id).value();
VExprSPtr ancestor_expr;
RETURN_IF_ERROR(build_equality_delete_key_expr(data_path, block_position,
&ancestor_expr));
@@ -1961,7 +2099,14 @@ Status
IcebergTableReader::_append_equality_delete_predicates(format::FileScanRe
} else {
const auto* root = data_path.front();
const auto field_column_id =
format::LocalColumnId(root->file_local_id());
- _append_file_scan_column(request, field_column_id,
&request->predicate_columns);
+ // A hidden key, like the missing key's ancestor above and the
delete file's keys,
+ // has no ColumnMapping to carry the timestamp marker
annotate_file_schema() put on
+ // its file column. Without it Parquet decodes an INT96
timestamptz key as
+ // DATETIMEV2 while the key expression below, built from the
annotated file schema,
+ // expects TIMESTAMPTZ. A projected key, or a projected
ancestor of a missing key,
+ // moves into the predicate columns and must keep the marker
its mapping carried.
+
RETURN_IF_ERROR(format::FileScanRequestBuilder(request).add_predicate_column(
+ equality_delete_root_projection(*root)));
const auto block_position =
request->local_positions.at(field_column_id).value();
RETURN_IF_ERROR(
build_equality_delete_key_expr(data_path,
block_position, &key_expr));
@@ -2284,6 +2429,15 @@ Status
IcebergTableReader::_load_equality_delete_file(const TIcebergDeleteFileDe
std::vector<format::ColumnDefinition> schema;
RETURN_IF_ERROR(reader->get_schema(&schema));
+ if (delete_file.file_format == TFileFormatType::FORMAT_PARQUET) {
+ // Decode tombstones of a timestamptz key as instants, like the data
keys they are
+ // compared with. Otherwise a catalog with
enable_mapping_timestamp_tz=false reads them as
+ // session-local wall clocks, which cannot tell the two instants of a
DST fall-back apart:
+ // in America/New_York a tombstone for 2024-11-03T05:30:00Z (01:30
EDT) would then match
+ // the row at 06:30:00Z (01:30 EST) as well, or instead. Delete files
carry field ids.
+ RETURN_IF_ERROR(_apply_table_timestamp_semantics(
+ &schema, format::TableColumnMappingMode::BY_FIELD_ID));
+ }
std::vector<EqualityDeleteColumnPath> delete_paths;
RETURN_IF_ERROR(_resolve_equality_delete_fields(delete_file, schema,
&delete_paths, result));
@@ -2292,7 +2446,7 @@ Status
IcebergTableReader::_load_equality_delete_file(const TIcebergDeleteFileDe
for (const auto& path : delete_paths) {
DORIS_CHECK(!path.empty());
RETURN_IF_ERROR(request_builder.add_non_predicate_column(
- format::LocalColumnId(path.front()->file_local_id())));
+ equality_delete_root_projection(*path.front())));
}
Block file_block_template;
std::vector<const format::ColumnDefinition*>
roots(request->local_positions.size());
diff --git a/be/src/format_v2/table/iceberg_reader.h
b/be/src/format_v2/table/iceberg_reader.h
index 47f71353137..90feaac4f8f 100644
--- a/be/src/format_v2/table/iceberg_reader.h
+++ b/be/src/format_v2/table/iceberg_reader.h
@@ -68,19 +68,7 @@ public:
format::ColumnDefinition* column) const
override;
std::string debug_string() const override;
format::TableColumnMappingMode mapping_mode() const override {
- const bool has_field_ids =
supports_iceberg_scan_semantics_v1(_scan_params)
- ?
schema_has_any_field_id(_data_reader.file_schema)
- :
schema_has_all_field_ids(_data_reader.file_schema);
- if (!_data_reader.file_schema.empty() && has_field_ids) {
- return format::TableColumnMappingMode::BY_FIELD_ID;
- }
- if (!_data_reader.file_schema.empty() &&
supports_iceberg_scan_semantics_v2(_scan_params) &&
- !_scan_has_any_authoritative_name_mapping()) {
- // ID-less migrated files are name-readable only while Iceberg's
explicit default name
- // mapping exists; current names must not resurrect file fields
after it is removed.
- return format::TableColumnMappingMode::BY_FIELD_ID;
- }
- return format::TableColumnMappingMode::BY_NAME;
+ return _mapping_mode(_data_reader.file_schema);
}
protected:
@@ -149,6 +137,27 @@ private:
static constexpr size_t ICEBERG_ROW_POS_BLOCK_POSITION = 1;
bool _scan_has_any_authoritative_name_mapping() const;
+ // annotate_file_schema() runs before TableReader stores the file schema,
so the mode is also
+ // computable from the schema being annotated.
+ format::TableColumnMappingMode _mapping_mode(
+ const std::vector<format::ColumnDefinition>& file_schema) const {
+ const bool has_field_ids =
supports_iceberg_scan_semantics_v1(_scan_params)
+ ?
schema_has_any_field_id(file_schema)
+ :
schema_has_all_field_ids(file_schema);
+ if (!file_schema.empty() && has_field_ids) {
+ return format::TableColumnMappingMode::BY_FIELD_ID;
+ }
+ if (!file_schema.empty() &&
supports_iceberg_scan_semantics_v2(_scan_params) &&
+ !_scan_has_any_authoritative_name_mapping()) {
+ // ID-less migrated files are name-readable only while Iceberg's
explicit default name
+ // mapping exists; current names must not resurrect file fields
after it is removed.
+ return format::TableColumnMappingMode::BY_FIELD_ID;
+ }
+ return format::TableColumnMappingMode::BY_NAME;
+ }
+ void _apply_positional_id_fallback(std::vector<format::ColumnDefinition>*
file_schema);
+ Status
_apply_table_timestamp_semantics(std::vector<format::ColumnDefinition>*
file_schema,
+ format::TableColumnMappingMode
mode) const;
class PositionDeleteRowsCollector final {
public:
diff --git a/be/src/format_v2/table/paimon_reader.cpp
b/be/src/format_v2/table/paimon_reader.cpp
index 4473240011d..2230b67ec21 100644
--- a/be/src/format_v2/table/paimon_reader.cpp
+++ b/be/src/format_v2/table/paimon_reader.cpp
@@ -31,7 +31,6 @@
#include "core/column/column_string.h"
#include "core/column/column_vector.h"
#include "core/data_type/data_type_array.h"
-#include "core/data_type/data_type_factory.hpp"
#include "core/data_type/data_type_map.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
@@ -50,58 +49,6 @@
namespace doris::format::paimon {
namespace {
-DataTypePtr nullable_like_original(const DataTypePtr& original, DataTypePtr
nested) {
- return original != nullptr && original->is_nullable() ?
make_nullable(nested) : nested;
-}
-
-// Unannotated INT96 cannot distinguish Paimon TIMESTAMP from TIMESTAMP_LTZ.
Apply the
-// table schema's instant marker to each timestamp leaf, then rebuild its
containers so the
-// parent type agrees with the children. Preserve precision and the original
nullability.
-Status apply_paimon_timestamp_semantics(format::ColumnDefinition* column) {
- DORIS_CHECK(column != nullptr);
- DORIS_CHECK(column->type != nullptr);
- const auto primitive = remove_nullable(column->type)->get_primitive_type();
- // Validate before changing children: silently skipping a malformed parent
leaves stale types.
- if ((primitive == TYPE_ARRAY && column->children.size() != 1) ||
- (primitive == TYPE_MAP && column->children.size() != 2)) {
- return Status::InvalidArgument("Invalid Paimon {} column '{}': {}
children",
- column->type->get_name(), column->name,
- column->children.size());
- }
- if (column->timestamp_is_adjusted_to_utc.has_value() &&
- (primitive == TYPE_DATETIMEV2 || primitive == TYPE_TIMESTAMPTZ)) {
- const auto target =
- *column->timestamp_is_adjusted_to_utc ? TYPE_TIMESTAMPTZ :
TYPE_DATETIMEV2;
- column->type = DataTypeFactory::instance().create_data_type(
- target, column->type->is_nullable(), 0,
column->type->get_scale());
- return Status::OK();
- }
-
- std::vector<DataTypePtr> child_types;
- child_types.reserve(column->children.size());
- for (auto& child : column->children) {
- RETURN_IF_ERROR(apply_paimon_timestamp_semantics(&child));
- child_types.push_back(child.type);
- }
- if (primitive == TYPE_ARRAY) {
- column->type = nullable_like_original(column->type,
-
std::make_shared<DataTypeArray>(child_types.front()));
- } else if (primitive == TYPE_MAP) {
- column->type = nullable_like_original(
- column->type,
std::make_shared<DataTypeMap>(make_nullable(child_types[0]),
-
make_nullable(child_types[1])));
- } else if (primitive == TYPE_STRUCT) {
- Strings child_names;
- child_names.reserve(column->children.size());
- for (const auto& child : column->children) {
- child_names.push_back(child.name);
- }
- column->type = nullable_like_original(
- column->type, std::make_shared<DataTypeStruct>(child_types,
child_names));
- }
- return Status::OK();
-}
-
ColumnDefinition* find_file_column(const ColumnDefinition& table_column,
std::vector<ColumnDefinition>* file_schema,
TableColumnMappingMode mode) {
@@ -288,8 +235,10 @@ Status
PaimonReader::annotate_file_schema(std::vector<format::ColumnDefinition>*
file_schema));
}
if (_format == format::FileFormat::PARQUET) {
+ // Unannotated INT96 cannot distinguish Paimon TIMESTAMP from
TIMESTAMP_LTZ; the markers
+ // copied from the split's history schema decide.
for (auto& column : *file_schema) {
- RETURN_IF_ERROR(apply_paimon_timestamp_semantics(&column));
+ RETURN_IF_ERROR(format::apply_file_timestamp_semantics(&column));
}
const bool projects_variant =
std::ranges::any_of(_projected_columns, contains_variant_type);
diff --git a/be/src/format_v2/table/schema_history_util.cpp
b/be/src/format_v2/table/schema_history_util.cpp
index a3dbad7c1e3..47015ff43cd 100644
--- a/be/src/format_v2/table/schema_history_util.cpp
+++ b/be/src/format_v2/table/schema_history_util.cpp
@@ -21,12 +21,21 @@
#include <ranges>
#include <string>
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_struct.h"
#include "core/field.h"
#include "util/string_util.h"
namespace doris::format {
namespace {
+DataTypePtr nullable_like_original(const DataTypePtr& original, DataTypePtr
nested) {
+ return original != nullptr && original->is_nullable() ?
make_nullable(nested) : nested;
+}
+
const schema::external::TField* find_child_field_by_name(
const std::vector<schema::external::TFieldPtr>& fields, const
std::string& name) {
for (const auto& field_ptr : fields) {
@@ -120,6 +129,69 @@ const schema::external::TSchema* find_history_schema(const
TFileScanRangeParams*
return nullptr;
}
+const schema::external::TSchema* find_current_schema(const
TFileScanRangeParams* params) {
+ if (params == nullptr || !params->__isset.history_schema_info ||
+ params->history_schema_info.empty()) {
+ return nullptr;
+ }
+ if (params->__isset.current_schema_id) {
+ if (const auto* schema = find_history_schema(params,
params->current_schema_id);
+ schema != nullptr) {
+ return schema;
+ }
+ }
+ return ¶ms->history_schema_info.front();
+}
+
+ColumnDefinition build_schema_identity_from_external_field(const
schema::external::TField& field) {
+ ColumnDefinition identity;
+ if (field.__isset.id) {
+ identity.identifier = Field::create_field<TYPE_INT>(field.id);
+ }
+ identity.name = field.__isset.name ? field.name : "";
+ identity.name_mapping =
+ field.__isset.name_mapping ? field.name_mapping :
std::vector<std::string> {};
+ identity.has_name_mapping =
+ field.__isset.name_mapping_is_authoritative &&
field.name_mapping_is_authoritative;
+ identity.timestamp_is_adjusted_to_utc =
+ field.__isset.timestamp_is_adjusted_to_utc
+ ? std::make_optional(field.timestamp_is_adjusted_to_utc)
+ : std::nullopt;
+ if (!field.__isset.nestedField) {
+ return identity;
+ }
+ if (field.nestedField.__isset.struct_field &&
field.nestedField.struct_field.__isset.fields) {
+ for (const auto& child_ptr : field.nestedField.struct_field.fields) {
+ if (const auto* child = get_field_ptr(child_ptr); child !=
nullptr) {
+
identity.children.push_back(build_schema_identity_from_external_field(*child));
+ }
+ }
+ } else if (field.nestedField.__isset.array_field &&
+ field.nestedField.array_field.__isset.item_field) {
+ if (const auto* child =
get_field_ptr(field.nestedField.array_field.item_field);
+ child != nullptr) {
+
identity.children.push_back(build_schema_identity_from_external_field(*child));
+ identity.children.back().name = "element";
+ }
+ } else if (field.nestedField.__isset.map_field) {
+ if (field.nestedField.map_field.__isset.key_field) {
+ if (const auto* child =
get_field_ptr(field.nestedField.map_field.key_field);
+ child != nullptr) {
+
identity.children.push_back(build_schema_identity_from_external_field(*child));
+ identity.children.back().name = "key";
+ }
+ }
+ if (field.nestedField.map_field.__isset.value_field) {
+ if (const auto* child =
get_field_ptr(field.nestedField.map_field.value_field);
+ child != nullptr) {
+
identity.children.push_back(build_schema_identity_from_external_field(*child));
+ identity.children.back().name = "value";
+ }
+ }
+ }
+ return identity;
+}
+
bool can_map_by_history_schema(const TFileScanRangeParams* params, int64_t
split_schema_id) {
if (split_schema_id < 0 || params == nullptr ||
!params->__isset.current_schema_id ||
!params->__isset.history_schema_info) {
@@ -146,4 +218,49 @@ Status annotate_file_schema_from_history(const
TFileScanRangeParams* params,
return Status::OK();
}
+Status apply_file_timestamp_semantics(ColumnDefinition* column) {
+ DORIS_CHECK(column != nullptr);
+ DORIS_CHECK(column->type != nullptr);
+ const auto primitive = remove_nullable(column->type)->get_primitive_type();
+ // Validate before changing children: silently skipping a malformed parent
leaves stale types.
+ if ((primitive == TYPE_ARRAY && column->children.size() != 1) ||
+ (primitive == TYPE_MAP && column->children.size() != 2)) {
+ return Status::InvalidArgument("Invalid {} column '{}': {} children",
+ column->type->get_name(), column->name,
+ column->children.size());
+ }
+ if (column->timestamp_is_adjusted_to_utc.has_value() &&
+ (primitive == TYPE_DATETIMEV2 || primitive == TYPE_TIMESTAMPTZ)) {
+ const auto target =
+ *column->timestamp_is_adjusted_to_utc ? TYPE_TIMESTAMPTZ :
TYPE_DATETIMEV2;
+ column->type = DataTypeFactory::instance().create_data_type(
+ target, column->type->is_nullable(), 0,
column->type->get_scale());
+ return Status::OK();
+ }
+
+ std::vector<DataTypePtr> child_types;
+ child_types.reserve(column->children.size());
+ for (auto& child : column->children) {
+ RETURN_IF_ERROR(apply_file_timestamp_semantics(&child));
+ child_types.push_back(child.type);
+ }
+ if (primitive == TYPE_ARRAY) {
+ column->type = nullable_like_original(column->type,
+
std::make_shared<DataTypeArray>(child_types.front()));
+ } else if (primitive == TYPE_MAP) {
+ column->type = nullable_like_original(
+ column->type,
std::make_shared<DataTypeMap>(make_nullable(child_types[0]),
+
make_nullable(child_types[1])));
+ } else if (primitive == TYPE_STRUCT) {
+ Strings child_names;
+ child_names.reserve(column->children.size());
+ for (const auto& child : column->children) {
+ child_names.push_back(child.name);
+ }
+ column->type = nullable_like_original(
+ column->type, std::make_shared<DataTypeStruct>(child_types,
child_names));
+ }
+ return Status::OK();
+}
+
} // namespace doris::format
diff --git a/be/src/format_v2/table/schema_history_util.h
b/be/src/format_v2/table/schema_history_util.h
index 49704c5fb98..26abe338f4d 100644
--- a/be/src/format_v2/table/schema_history_util.h
+++ b/be/src/format_v2/table/schema_history_util.h
@@ -38,6 +38,14 @@ inline const schema::external::TField* get_field_ptr(const
schema::external::TFi
const schema::external::TSchema* find_history_schema(const
TFileScanRangeParams* params,
int64_t schema_id);
+// The schema named by current_schema_id, or the first one when FE sends no
id; nullptr when the
+// plan carries no schema dictionary.
+const schema::external::TSchema* find_current_schema(const
TFileScanRangeParams* params);
+
+// Build the identity of a table field and its nested fields: field id, name,
name mapping and
+// timestamp marker, but no type. Array items are named `element` and map
children `key`/`value`.
+ColumnDefinition build_schema_identity_from_external_field(const
schema::external::TField& field);
+
bool can_map_by_history_schema(const TFileScanRangeParams* params, int64_t
split_schema_id);
// Annotate a file-local schema with the field ids and name mappings from the
historical table
@@ -48,4 +56,14 @@ Status annotate_file_schema_from_history(const
TFileScanRangeParams* params,
int64_t split_schema_id,
std::vector<ColumnDefinition>*
file_schema);
+// Parquet INT96 carries no instant annotation, so the same physical column
can hold a wall clock
+// (Paimon TIMESTAMP) or an instant (Paimon TIMESTAMP_LTZ, an Iceberg
timestamptz written by Spark
+// or Hive). Rewrite each DATETIMEV2/TIMESTAMPTZ leaf that carries the table
schema's marker
+// (`timestamp_is_adjusted_to_utc`) to TIMESTAMPTZ or DATETIMEV2, then rebuild
its containers so a
+// parent type agrees with its children. Precision and nullability are
preserved, and a leaf
+// without a marker keeps its file type. For example, an INT96 leaf read as
DATETIMEV2(6) and marked
+// as an instant becomes TIMESTAMPTZ(6), and an ARRAY<DATETIMEV2(6)> around it
becomes
+// ARRAY<TIMESTAMPTZ(6)>.
+Status apply_file_timestamp_semantics(ColumnDefinition* column);
+
} // namespace doris::format
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index 146c7f2d9f2..fdbc5441f7f 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -265,51 +265,6 @@ bool find_column_identity_path_by_id(const
std::vector<ColumnDefinition>& fields
return false;
}
-ColumnDefinition build_schema_identity_from_external_field(const
schema::external::TField& field) {
- ColumnDefinition identity;
- if (field.__isset.id) {
- identity.identifier = Field::create_field<TYPE_INT>(field.id);
- }
- identity.name = field.__isset.name ? field.name : "";
- identity.name_mapping =
- field.__isset.name_mapping ? field.name_mapping :
std::vector<std::string> {};
- identity.has_name_mapping =
- field.__isset.name_mapping_is_authoritative &&
field.name_mapping_is_authoritative;
- if (!field.__isset.nestedField) {
- return identity;
- }
- if (field.nestedField.__isset.struct_field &&
field.nestedField.struct_field.__isset.fields) {
- for (const auto& child_ptr : field.nestedField.struct_field.fields) {
- if (const auto* child = get_field_ptr(child_ptr); child !=
nullptr) {
-
identity.children.push_back(build_schema_identity_from_external_field(*child));
- }
- }
- } else if (field.nestedField.__isset.array_field &&
- field.nestedField.array_field.__isset.item_field) {
- if (const auto* child =
get_field_ptr(field.nestedField.array_field.item_field);
- child != nullptr) {
-
identity.children.push_back(build_schema_identity_from_external_field(*child));
- identity.children.back().name = "element";
- }
- } else if (field.nestedField.__isset.map_field) {
- if (field.nestedField.map_field.__isset.key_field) {
- if (const auto* child =
get_field_ptr(field.nestedField.map_field.key_field);
- child != nullptr) {
-
identity.children.push_back(build_schema_identity_from_external_field(*child));
- identity.children.back().name = "key";
- }
- }
- if (field.nestedField.map_field.__isset.value_field) {
- if (const auto* child =
get_field_ptr(field.nestedField.map_field.value_field);
- child != nullptr) {
-
identity.children.push_back(build_schema_identity_from_external_field(*child));
- identity.children.back().name = "value";
- }
- }
- }
- return identity;
-}
-
const ColumnDefinition* find_identity_child(const ColumnDefinition&
projected_child,
const ColumnDefinition&
identity_parent) {
const auto child_it = std::ranges::find_if(
@@ -517,21 +472,8 @@ ColumnDefinition
build_schema_column_from_external_field(const schema::external:
const schema::external::TField* find_external_root_field(const
TFileScanRangeParams* params,
const
ColumnDefinition& column) {
- if (params == nullptr || !params->__isset.history_schema_info ||
- params->history_schema_info.empty()) {
- return nullptr;
- }
- const auto* schema = ¶ms->history_schema_info.front();
- if (params->__isset.current_schema_id) {
- for (const auto& candidate_schema : params->history_schema_info) {
- if (candidate_schema.__isset.schema_id &&
- candidate_schema.schema_id == params->current_schema_id) {
- schema = &candidate_schema;
- break;
- }
- }
- }
- if (!schema->__isset.root_field || !schema->root_field.__isset.fields) {
+ const auto* schema = find_current_schema(params);
+ if (schema == nullptr || !schema->__isset.root_field ||
!schema->root_field.__isset.fields) {
return nullptr;
}
if (!supports_iceberg_scan_semantics_v1(params)) {
@@ -771,16 +713,7 @@ std::optional<ColumnDefinition>
TableReader::_find_table_column_by_field_id(
: nullptr;
};
- const auto* current_schema = &_scan_params->history_schema_info.front();
- if (_scan_params->__isset.current_schema_id) {
- for (const auto& candidate_schema : _scan_params->history_schema_info)
{
- if (candidate_schema.__isset.schema_id &&
- candidate_schema.schema_id == _scan_params->current_schema_id)
{
- current_schema = &candidate_schema;
- break;
- }
- }
- }
+ const auto* current_schema = find_current_schema(_scan_params);
if (const auto* field = find_field(*current_schema); field != nullptr) {
return build_schema_column_from_external_field(
*field, std::move(type),
supports_iceberg_scan_semantics_v1(_scan_params));
@@ -858,16 +791,7 @@ std::optional<std::vector<ColumnDefinition>>
TableReader::_find_table_column_pat
return result;
};
- const auto* current_schema = &_scan_params->history_schema_info.front();
- if (_scan_params->__isset.current_schema_id) {
- for (const auto& candidate_schema : _scan_params->history_schema_info)
{
- if (candidate_schema.__isset.schema_id &&
- candidate_schema.schema_id == _scan_params->current_schema_id)
{
- current_schema = &candidate_schema;
- break;
- }
- }
- }
+ const auto* current_schema = find_current_schema(_scan_params);
if (auto path = build_path(*current_schema); path.has_value()) {
return path;
}
@@ -922,16 +846,7 @@
TableReader::_find_table_column_identity_path_by_field_id(int32_t field_id,
return std::nullopt;
};
- const auto* current_schema = &_scan_params->history_schema_info.front();
- if (_scan_params->__isset.current_schema_id) {
- for (const auto& candidate_schema : _scan_params->history_schema_info)
{
- if (candidate_schema.__isset.schema_id &&
- candidate_schema.schema_id == _scan_params->current_schema_id)
{
- current_schema = &candidate_schema;
- break;
- }
- }
- }
+ const auto* current_schema = find_current_schema(_scan_params);
if (auto path = find_path(*current_schema); path.has_value()) {
return path;
}
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index 8b41271b196..d5dcea52cde 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -259,6 +259,17 @@ public:
}
};
+class IcebergTableReaderAnnotateTestHelper final
+ : public doris::format::iceberg::IcebergTableReader {
+public:
+ Status annotate(std::vector<ColumnDefinition>* file_schema,
TFileScanRangeParams* scan_params,
+ FileFormat format = FileFormat::PARQUET) {
+ _scan_params = scan_params;
+ _format = format;
+ return annotate_file_schema(file_schema);
+ }
+};
+
std::shared_ptr<arrow::Array> finish_array(arrow::ArrayBuilder* builder) {
std::shared_ptr<arrow::Array> array;
EXPECT_TRUE(builder->Finish(&array).ok());
@@ -1025,20 +1036,25 @@ void write_nested_key_value_orc_file(const std::string&
file_path, const std::ve
// Writes the adjusted-UTC `event_time` (field id 0) and `id` (field id 1).
With `int96`,
// `event_time` is stored as INT96, as Spark and Hive write timestamps, which
drops the annotation.
+// Without `field_ids` the file looks like one Spark wrote and `add_files`
attached to a table.
void write_timestamp_int_parquet_file(const std::string& file_path,
const std::vector<int64_t>& timestamps,
- const std::vector<int32_t>& ids, bool
int96 = false) {
+ const std::vector<int32_t>& ids, bool
int96 = false,
+ bool field_ids = true) {
ASSERT_EQ(timestamps.size(), ids.size());
- const auto timestamp_metadata =
arrow::key_value_metadata({"PARQUET:field_id"}, {"0"});
- const auto id_metadata = arrow::key_value_metadata({"PARQUET:field_id"},
{"1"});
const auto timestamp_type = arrow::timestamp(arrow::TimeUnit::MICRO,
"UTC");
arrow::TimestampBuilder timestamp_builder(timestamp_type,
arrow::default_memory_pool());
ASSERT_TRUE(timestamp_builder.AppendValues(timestamps).ok());
auto timestamp_result = timestamp_builder.Finish();
ASSERT_TRUE(timestamp_result.ok()) << timestamp_result.status();
- auto schema = arrow::schema(
- {arrow::field("event_time", timestamp_type,
false)->WithMetadata(timestamp_metadata),
- arrow::field("id", arrow::int32(),
false)->WithMetadata(id_metadata)});
+ auto timestamp_field = arrow::field("event_time", timestamp_type, false);
+ auto id_field = arrow::field("id", arrow::int32(), false);
+ if (field_ids) {
+ timestamp_field = timestamp_field->WithMetadata(
+ arrow::key_value_metadata({"PARQUET:field_id"}, {"0"}));
+ id_field =
id_field->WithMetadata(arrow::key_value_metadata({"PARQUET:field_id"}, {"1"}));
+ }
+ auto schema = arrow::schema({timestamp_field, id_field});
auto table = arrow::Table::Make(schema, {*timestamp_result,
build_int32_array(ids)});
auto file_result = arrow::io::FileOutputStream::Open(file_path);
@@ -1057,6 +1073,128 @@ void write_timestamp_int_parquet_file(const
std::string& file_path,
builder.build(),
arrow_builder.build()));
}
+// Writes `id` (field id 1), `info` STRUCT<tz (3), wall (4)> (2) and `tzs`
LIST<element (6)> (5)
+// with every timestamp stored as INT96, as Spark writes nested timestamps.
+void write_nested_int96_parquet_file(const std::string& file_path, int64_t
timestamp) {
+ const auto field_id = [](int32_t id) {
+ return arrow::key_value_metadata({"PARQUET:field_id"},
{std::to_string(id)});
+ };
+ const auto timestamp_type = arrow::timestamp(arrow::TimeUnit::MICRO,
"UTC");
+ arrow::TimestampBuilder timestamp_builder(timestamp_type,
arrow::default_memory_pool());
+ ASSERT_TRUE(timestamp_builder.Append(timestamp).ok());
+ auto timestamps = timestamp_builder.Finish();
+ ASSERT_TRUE(timestamps.ok()) << timestamps.status();
+ auto tz_field = arrow::field("tz",
timestamp_type)->WithMetadata(field_id(3));
+ auto wall_field = arrow::field("wall",
timestamp_type)->WithMetadata(field_id(4));
+ auto info = arrow::StructArray::Make({*timestamps, *timestamps},
{tz_field, wall_field});
+ ASSERT_TRUE(info.ok()) << info.status();
+ auto tzs_type = arrow::list(arrow::field("element",
timestamp_type)->WithMetadata(field_id(6)));
+ auto tzs = arrow::ListArray::FromArrays(tzs_type, *build_int32_array({0,
1}), **timestamps);
+ ASSERT_TRUE(tzs.ok()) << tzs.status();
+ auto schema = arrow::schema({
+ arrow::field("id", arrow::int32(),
false)->WithMetadata(field_id(1)),
+ arrow::field("info", (*info)->type())->WithMetadata(field_id(2)),
+ arrow::field("tzs", tzs_type)->WithMetadata(field_id(5)),
+ });
+ auto table = arrow::Table::Make(schema, {build_int32_array({1}), *info,
*tzs});
+
+ auto file_result = arrow::io::FileOutputStream::Open(file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+ ::parquet::ArrowWriterProperties::Builder arrow_builder;
+ arrow_builder.enable_force_write_int96_timestamps();
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(
+ *table, arrow::default_memory_pool(), out, 1,
+ ::parquet::WriterProperties::Builder().build(),
arrow_builder.build()));
+}
+
+// Writes `id` (field id 1) and `outer` STRUCT (10) holding an ID-less `inner`
STRUCT whose `ts` (30)
+// is an INT96 timestamp: the ID-less wrapper Iceberg's Parquet projection
keeps for a selected
+// nested ID, holding a timestamp the way Spark writes it.
+void write_idless_wrapper_int96_parquet_file(const std::string& file_path,
+ const std::vector<int64_t>&
timestamps,
+ const std::vector<int32_t>& ids) {
+ ASSERT_EQ(timestamps.size(), ids.size());
+ const auto field_id = [](int32_t id) {
+ return arrow::key_value_metadata({"PARQUET:field_id"},
{std::to_string(id)});
+ };
+ const auto timestamp_type = arrow::timestamp(arrow::TimeUnit::MICRO,
"UTC");
+ arrow::TimestampBuilder timestamp_builder(timestamp_type,
arrow::default_memory_pool());
+ ASSERT_TRUE(timestamp_builder.AppendValues(timestamps).ok());
+ auto timestamp_result = timestamp_builder.Finish();
+ ASSERT_TRUE(timestamp_result.ok()) << timestamp_result.status();
+ auto ts_field = arrow::field("ts", timestamp_type,
false)->WithMetadata(field_id(30));
+ auto inner_result = arrow::StructArray::Make({*timestamp_result},
{ts_field});
+ ASSERT_TRUE(inner_result.ok()) << inner_result.status();
+ auto inner_field = arrow::field("inner", arrow::struct_({ts_field}),
false);
+ auto outer_result = arrow::StructArray::Make({*inner_result},
{inner_field});
+ ASSERT_TRUE(outer_result.ok()) << outer_result.status();
+ auto schema = arrow::schema({
+ arrow::field("id", arrow::int32(),
false)->WithMetadata(field_id(1)),
+ arrow::field("outer", arrow::struct_({inner_field}),
false)->WithMetadata(field_id(10)),
+ });
+ auto table = arrow::Table::Make(schema, {build_int32_array(ids),
*outer_result});
+
+ auto file_result = arrow::io::FileOutputStream::Open(file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+ ::parquet::ArrowWriterProperties::Builder arrow_builder;
+ arrow_builder.enable_force_write_int96_timestamps();
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(
+ *table, arrow::default_memory_pool(), out,
static_cast<int64_t>(ids.size()),
+ ::parquet::WriterProperties::Builder().build(),
arrow_builder.build()));
+}
+
+// Writes `id` (field id 0) and a nullable `payload` STRUCT (1) of `existing`
(2) and `ts` (4), with
+// `ts` stored as INT96 the way Spark writes a timestamptz. A true
`parent_nulls` entry makes that
+// row's `payload` NULL.
+void write_nested_equality_int96_sibling_parquet_file(const std::string&
file_path,
+ const
std::vector<int32_t>& ids,
+ const
std::vector<int32_t>& values,
+ const
std::vector<int64_t>& timestamps,
+ const std::vector<bool>&
parent_nulls) {
+ ASSERT_EQ(ids.size(), values.size());
+ ASSERT_EQ(ids.size(), timestamps.size());
+ ASSERT_EQ(ids.size(), parent_nulls.size());
+ const auto field_id = [](int32_t id) {
+ return arrow::key_value_metadata({"PARQUET:field_id"},
{std::to_string(id)});
+ };
+ const auto timestamp_type = arrow::timestamp(arrow::TimeUnit::MICRO,
"UTC");
+ const auto payload_type = arrow::struct_(
+ {arrow::field("existing", arrow::int32(),
false)->WithMetadata(field_id(2)),
+ arrow::field("ts", timestamp_type,
false)->WithMetadata(field_id(4))});
+ arrow::StructBuilder payload_builder(payload_type,
arrow::default_memory_pool(),
+
{std::make_shared<arrow::Int32Builder>(),
+
std::make_shared<arrow::TimestampBuilder>(
+ timestamp_type,
arrow::default_memory_pool())});
+ auto* existing_builder =
assert_cast<arrow::Int32Builder*>(payload_builder.field_builder(0));
+ auto* ts_builder =
assert_cast<arrow::TimestampBuilder*>(payload_builder.field_builder(1));
+ for (size_t row = 0; row < ids.size(); ++row) {
+ if (parent_nulls[row]) {
+ ASSERT_TRUE(payload_builder.AppendNull().ok());
+ } else {
+ ASSERT_TRUE(payload_builder.Append().ok());
+ ASSERT_TRUE(existing_builder->Append(values[row]).ok());
+ ASSERT_TRUE(ts_builder->Append(timestamps[row]).ok());
+ }
+ }
+ auto schema = arrow::schema({
+ arrow::field("id", arrow::int32(),
false)->WithMetadata(field_id(0)),
+ arrow::field("payload", payload_type)->WithMetadata(field_id(1)),
+ });
+ auto table =
+ arrow::Table::Make(schema, {build_int32_array(ids),
finish_array(&payload_builder)});
+
+ auto file_result = arrow::io::FileOutputStream::Open(file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+ ::parquet::ArrowWriterProperties::Builder arrow_builder;
+ arrow_builder.enable_force_write_int96_timestamps();
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(
+ *table, arrow::default_memory_pool(), out,
static_cast<int64_t>(ids.size()),
+ ::parquet::WriterProperties::Builder().build(),
arrow_builder.build()));
+}
+
void write_iceberg_equality_delete_bigint_parquet_file(const std::string&
file_path,
int32_t field_id,
int64_t value) {
const auto metadata =
@@ -4645,6 +4783,156 @@ TEST(IcebergV2ReaderTest,
IcebergMissingNestedEqualityKeyPreservesNullableParent
}
}
+// An older Parquet file holds a nullable payload {existing, ts}, `ts` an
INT96 timestamptz as
+// Spark writes it, and the table later added payload.k with the default 7. An
equality delete on k
+// reads the whole `payload` for its NULLs: rows under a non-NULL payload take
the key 7, the row
+// under the NULL one keeps a NULL key. FE's marker turns `ts` into
TIMESTAMPTZ in the annotated
+// file schema, so that ancestor read must carry the marker too, or Parquet
still decodes `ts` as
+// DATETIMEV2 and the scan fails the reader's type check. A query projecting
`payload` failed the
+// same way: reading the root as a predicate column replaced the projection
its mapping had marked.
+// A plan without the marker keeps the file types and reads as before.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity): marker,
projection, tombstone cases.
+TEST(IcebergV2ReaderTest,
IcebergMissingNestedEqualityKeyAncestorKeepsInt96TimestampTzMarker) {
+ TimezoneUtils::load_timezones_to_cache();
+ const auto test_dir = std::filesystem::temp_directory_path() /
+ "doris_v2_missing_nested_equality_key_int96_sibling";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ // 2024-01-01 00:00:00.123456 UTC, then one and two seconds later.
+ constexpr int64_t T1_MICROS = 1'704'067'200'123'456L;
+ write_nested_equality_int96_sibling_parquet_file(
+ file_path, {1, 2, 3}, {10, 20, 30},
+ {T1_MICROS, T1_MICROS + 1'000'000L, T1_MICROS + 2'000'000L},
{false, true, false});
+ auto physical_reader = ::parquet::ParquetFileReader::OpenFile(file_path,
false);
+ ASSERT_EQ(::parquet::Type::INT96,
+
physical_reader->metadata()->schema()->Column(2)->physical_type());
+ physical_reader->Close();
+ // The two delete files have the same size and may share an mtime second,
so each tombstone
+ // gets its own path: one rewritten path could serve the first one from
the Parquet caches.
+ const auto delete_file_path = [&test_dir](int32_t tombstone) {
+ return (test_dir / ("equality-delete-" + std::to_string(tombstone) +
".parquet")).string();
+ };
+ for (const int32_t tombstone : {7, 8}) {
+ write_nested_equality_parquet_file(delete_file_path(tombstone), {},
{tombstone}, {false},
+ true, "k", 3);
+ }
+
+ const auto int_type = std::make_shared<DataTypeInt32>();
+ const auto tz_type = std::make_shared<DataTypeTimeStampTz>(6);
+ const auto nullable_tz_type = make_nullable(tz_type);
+ auto payload = make_table_column(1, "payload",
+
std::make_shared<DataTypeStruct>(DataTypes {int_type, tz_type},
+ Strings
{"existing", "ts"}));
+ payload.children = {make_table_column(2, "existing", int_type),
+ make_table_column(4, "ts", tz_type)};
+
+ struct Case {
+ bool marked;
+ bool project_payload;
+ int32_t tombstone;
+ std::vector<int32_t> ids;
+ // `ts` of each surviving row when `payload` is projected; nullopt
under a NULL payload.
+ std::vector<std::optional<std::string>> timestamps;
+ };
+ const std::vector<Case> cases = {
+ // A plan from an FE that sends no marker.
+ {false, false, 7, {2}, {}},
+ // The key is the default 7 under a non-NULL payload and NULL
under the NULL one.
+ {true, false, 7, {2}, {}},
+ {true, true, 7, {2}, {std::nullopt}},
+ // Instants, not session wall clocks read eight hours early in
Asia/Shanghai.
+ {true,
+ true,
+ 8,
+ {1, 2, 3},
+ {"2024-01-01 00:00:00.123456+00:00", std::nullopt,
+ "2024-01-01 00:00:02.123456+00:00"}},
+ };
+ for (const auto& c : cases) {
+ SCOPED_TRACE("marked=" + std::to_string(c.marked) +
+ " project_payload=" + std::to_string(c.project_payload) +
+ " tombstone=" + std::to_string(c.tombstone));
+ std::vector<ColumnDefinition> projected_columns {make_table_column(0,
"id", int_type)};
+ if (c.project_payload) {
+ projected_columns.push_back(payload);
+ }
+ auto ts_field =
+ external_schema_field("ts", 4, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::TIMESTAMPTZ, -1, 6));
+ if (c.marked) {
+ ts_field.field_ptr->__set_timestamp_is_adjusted_to_utc(true);
+ }
+ auto payload_field = external_struct_schema_field(
+ "payload", 1,
+ {external_schema_field("existing", 2, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::INT)),
+ std::move(ts_field),
+ external_schema_field("k", 3, {}, "7",
+
external_primitive_type(TPrimitiveType::INT), false, true)},
+ true);
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_parquet_timestamp_semantics_version(1);
+ scan_params.__set_enable_mapping_timestamp_tz(true);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info(
+ {external_schema(100, {external_schema_field("id", 0),
std::move(payload_field)})});
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ state.set_timezone("Asia/Shanghai");
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile);
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+
split_options.current_range.__set_table_format_params(make_iceberg_table_format_desc(
+ file_path,
{make_iceberg_equality_delete_file(delete_file_path(c.tombstone), {3})},
+ 3));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::optional<std::string>> timestamps;
+ bool eos = false;
+ while (!eos) {
+ Block block = build_table_block(projected_columns);
+ const auto status = reader.get_block(&block, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ if (block.rows() == 0) {
+ continue;
+ }
+ const auto& id_column =
+ assert_cast<const
ColumnInt32&>(expect_not_null_table_column(block, 0));
+ for (size_t row = 0; row < block.rows(); ++row) {
+ ids.push_back(id_column.get_element(row));
+ }
+ if (!c.project_payload) {
+ continue;
+ }
+ const auto& payload_column =
+ assert_cast<const
ColumnNullable&>(*block.get_by_position(1).column);
+ const auto& payload_struct =
+ assert_cast<const
ColumnStruct&>(payload_column.get_nested_column());
+ for (size_t row = 0; row < block.rows(); ++row) {
+ if (payload_column.is_null_at(row)) {
+ timestamps.emplace_back(std::nullopt);
+ } else {
+ timestamps.emplace_back(
+
nullable_tz_type->to_string(payload_struct.get_column(1), row));
+ }
+ }
+ }
+ EXPECT_EQ(ids, c.ids);
+ EXPECT_EQ(timestamps, c.timestamps);
+ ASSERT_TRUE(reader.close().ok());
+ }
+ std::filesystem::remove_all(test_dir);
+}
+
// Keep the shared Parquet/ORC reader setup together so both V2 paths
materialize missing struct
// literals before traversing their children.
//
NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
@@ -6391,6 +6679,571 @@ TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteComparesInt96DataKeyAsTimestampTz
std::filesystem::remove_all(test_dir);
}
+// FE marks every Iceberg timestamptz field with timestamp_is_adjusted_to_utc.
The marker lands on
+// the file leaf the mapper selects for the field, by field id at every
nesting level: a DATETIMEV2
+// leaf becomes TIMESTAMPTZ with its precision and nullability, its containers
are rebuilt around
+// it, and an unmarked timestamp field stays a wall clock. A TIMESTAMP_NS leaf
(INT64 NANOS without
+// the UTC annotation) keeps its type and takes no marker; the Parquet reader
rewrites every marked
+// timestamp leaf, so a marker there would make it decode TIMESTAMPTZ under a
TIMESTAMP_NS file
+// schema. ORC files keep their types, since ORC's file type already separates
the two semantics.
+TEST(IcebergV2ReaderTest,
IcebergAnnotatesTimestampTzMarkerOnMatchedParquetLeaves) {
+ const auto marked = [](schema::external::TFieldPtr field) {
+ field.field_ptr->__set_timestamp_is_adjusted_to_utc(true);
+ return field;
+ };
+ auto info = external_struct_schema_field(
+ "info", 4, {marked(external_schema_field("tz", 5)),
external_schema_field("wall", 6)});
+ auto tzs = external_schema_field("tzs", 7);
+ schema::external::TArrayField array_field;
+ array_field.__set_item_field(marked(external_schema_field("element", 8)));
+ tzs.field_ptr->nestedField.__set_array_field(std::move(array_field));
+ tzs.field_ptr->__isset.nestedField = true;
+ auto by_tz = external_schema_field("by_tz", 9);
+ schema::external::TMapField map_field;
+ map_field.__set_key_field(marked(external_schema_field("key", 10)));
+ map_field.__set_value_field(external_schema_field("value", 11));
+ by_tz.field_ptr->nestedField.__set_map_field(std::move(map_field));
+ by_tz.field_ptr->__isset.nestedField = true;
+
+ TFileScanRangeParams scan_params;
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info({external_schema(
+ 100, {marked(external_schema_field("tz", 1)),
external_schema_field("wall", 2),
+ marked(external_schema_field("ns", 3)), info, tzs, by_tz})});
+
+ const auto datetime = std::make_shared<DataTypeDateTimeV2>(6);
+ const auto build_file_schema = [&]() {
+ auto info_column =
+ make_file_column(4, "info",
+ std::make_shared<DataTypeStruct>(DataTypes
{datetime, datetime},
+ Strings
{"tz", "wall"}));
+ info_column.children = {make_file_column(5, "tz", datetime),
+ make_file_column(6, "wall", datetime)};
+ auto tzs_column = make_file_column(7, "tzs",
std::make_shared<DataTypeArray>(datetime));
+ tzs_column.children = {make_file_column(8, "element", datetime)};
+ auto by_tz_column =
+ make_file_column(9, "by_tz",
std::make_shared<DataTypeMap>(datetime, datetime));
+ by_tz_column.children = {make_file_column(10, "key", datetime),
+ make_file_column(11, "value", datetime)};
+ return std::vector<ColumnDefinition> {
+ make_file_column(1, "tz", datetime),
+ make_file_column(2, "wall", datetime),
+ make_file_column(3, "ns",
std::make_shared<DataTypeTimeStampNs>()),
+ std::move(info_column),
+ std::move(tzs_column),
+ std::move(by_tz_column)};
+ };
+ const auto primitive = [](const DataTypePtr& type) {
+ return remove_nullable(type)->get_primitive_type();
+ };
+
+ auto file_schema = build_file_schema();
+ IcebergTableReaderAnnotateTestHelper reader;
+ ASSERT_TRUE(reader.annotate(&file_schema, &scan_params).ok());
+ ASSERT_EQ(file_schema[0].timestamp_is_adjusted_to_utc,
std::optional<bool>(true));
+ EXPECT_EQ(primitive(file_schema[0].type), TYPE_TIMESTAMPTZ);
+ EXPECT_EQ(file_schema[0].type->get_scale(), 6);
+ EXPECT_TRUE(file_schema[0].type->is_nullable());
+ EXPECT_FALSE(file_schema[1].timestamp_is_adjusted_to_utc.has_value());
+ EXPECT_EQ(primitive(file_schema[1].type), TYPE_DATETIMEV2);
+ EXPECT_FALSE(file_schema[2].timestamp_is_adjusted_to_utc.has_value());
+ EXPECT_EQ(primitive(file_schema[2].type), TYPE_TIMESTAMP_NS);
+
+ const auto& struct_type =
+ assert_cast<const
DataTypeStruct&>(*remove_nullable(file_schema[3].type));
+ EXPECT_EQ(primitive(struct_type.get_element(0)), TYPE_TIMESTAMPTZ);
+ EXPECT_EQ(primitive(struct_type.get_element(1)), TYPE_DATETIMEV2);
+ EXPECT_EQ(file_schema[3].children[0].timestamp_is_adjusted_to_utc,
std::optional<bool>(true));
+
EXPECT_FALSE(file_schema[3].children[1].timestamp_is_adjusted_to_utc.has_value());
+
+ const auto& array_type =
+ assert_cast<const
DataTypeArray&>(*remove_nullable(file_schema[4].type));
+ EXPECT_EQ(primitive(array_type.get_nested_type()), TYPE_TIMESTAMPTZ);
+ EXPECT_EQ(file_schema[4].children[0].timestamp_is_adjusted_to_utc,
std::optional<bool>(true));
+
+ const auto& map_type = assert_cast<const
DataTypeMap&>(*remove_nullable(file_schema[5].type));
+ EXPECT_EQ(primitive(map_type.get_key_type()), TYPE_TIMESTAMPTZ);
+ EXPECT_EQ(primitive(map_type.get_value_type()), TYPE_DATETIMEV2);
+ EXPECT_EQ(file_schema[5].children[0].timestamp_is_adjusted_to_utc,
std::optional<bool>(true));
+
EXPECT_FALSE(file_schema[5].children[1].timestamp_is_adjusted_to_utc.has_value());
+
+ auto orc_schema = build_file_schema();
+ ASSERT_TRUE(reader.annotate(&orc_schema, &scan_params,
FileFormat::ORC).ok());
+ EXPECT_FALSE(orc_schema[0].timestamp_is_adjusted_to_utc.has_value());
+ EXPECT_EQ(primitive(orc_schema[0].type), TYPE_DATETIMEV2);
+ EXPECT_EQ(primitive(assert_cast<const
DataTypeArray&>(*remove_nullable(orc_schema[4].type))
+ .get_nested_type()),
+ TYPE_DATETIMEV2);
+}
+
+// Spark and Hive store an Iceberg timestamptz as INT96, which carries no
instant annotation, and FE
+// always sends Parquet timestamp-semantics version 1, under which INT96
decodes as a UTC wall
+// clock. Without FE's timestamptz marker the projection then casts that wall
clock to TIMESTAMPTZ
+// in the session time zone, so in Asia/Shanghai 2024-01-01T00:00:00.123456Z
reads as
+// 2023-12-31 16:00:00.123456 UTC, eight hours early. With the marker the
column reads as the
+// instant whether the file has field ids, relies on the table's name mapping
(an `add_files`
+// import), or falls back to positional ids, and whatever the catalog's
timestamp mapping flag is.
+// A plan from an older FE carries no marker and keeps the previous result.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity): one file per
match rule, two plans.
+TEST(IcebergV2ReaderTest, IcebergInt96TimestampTzReadsInstantWithTableMarker) {
+ TimezoneUtils::load_timezones_to_cache();
+ const auto test_dir =
+ std::filesystem::temp_directory_path() /
"doris_iceberg_int96_timestamp_tz_marker_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ // 2024-01-01 00:00:00.123456 UTC and 2024-07-01 12:34:56.654321 UTC.
+ constexpr int64_t T1_MICROS = 1'704'067'200'123'456L;
+ constexpr int64_t T2_MICROS = 1'719'837'296'654'321L;
+
+ enum class Match { FIELD_ID, NAME_MAPPING, POSITION };
+ for (const auto match : {Match::FIELD_ID, Match::NAME_MAPPING,
Match::POSITION}) {
+ const auto file_path =
+ (test_dir / ("split_" +
std::to_string(static_cast<int>(match)) + ".parquet"))
+ .string();
+ write_timestamp_int_parquet_file(file_path, {T1_MICROS, T2_MICROS},
{1, 2}, true,
+ match == Match::FIELD_ID);
+ // The positional fallback numbers the physical top-level columns from
1.
+ const int32_t event_time_id = match == Match::POSITION ? 1 : 0;
+ const int32_t id_id = match == Match::POSITION ? 2 : 1;
+ for (const bool marked : {true, false}) {
+ for (const bool mapping_timestamp_tz : {true, false}) {
+ SCOPED_TRACE("match=" +
std::to_string(static_cast<int>(match)) +
+ " marked=" + std::to_string(marked) +
+ " mapping_timestamp_tz=" +
std::to_string(mapping_timestamp_tz));
+ auto event_time = external_schema_field(
+ "event_time", event_time_id, {}, std::nullopt,
+ external_primitive_type(TPrimitiveType::TIMESTAMPTZ,
-1, 6));
+ auto id = external_schema_field("id", id_id);
+ if (marked) {
+
event_time.field_ptr->__set_timestamp_is_adjusted_to_utc(true);
+ }
+ std::vector<ColumnDefinition> projected_columns {
+ make_table_column(id_id, "id",
std::make_shared<DataTypeInt32>()),
+ make_table_column(event_time_id, "event_time",
+
std::make_shared<DataTypeTimeStampTz>(6))};
+ if (match == Match::NAME_MAPPING) {
+ for (auto* field : {&event_time, &id}) {
+
field->field_ptr->__set_name_mapping({field->field_ptr->name});
+
field->field_ptr->__set_name_mapping_is_authoritative(true);
+ }
+ for (auto& column : projected_columns) {
+ column.name_mapping = {column.name};
+ column.has_name_mapping = true;
+ }
+ }
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_parquet_timestamp_semantics_version(1);
+
scan_params.__set_enable_mapping_timestamp_tz(mapping_timestamp_tz);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info({external_schema(100,
{event_time, id})});
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ state.set_timezone("Asia/Shanghai");
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats,
&file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params,
io_ctx, &state,
+ &profile);
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(file_path, {}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ std::vector<std::string> event_times;
+ bool eos = false;
+ while (!eos) {
+ Block block = build_table_block(projected_columns);
+ const auto status = reader.get_block(&block, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ for (size_t row = 0; row < block.rows(); ++row) {
+
event_times.push_back(projected_columns[1].type->to_string(
+ *block.get_by_position(1).column, row));
+ }
+ }
+ EXPECT_EQ(event_times,
+ marked ? std::vector<std::string>({"2024-01-01
00:00:00.123456+00:00",
+ "2024-07-01
12:34:56.654321+00:00"})
+ : std::vector<std::string>({"2023-12-31
16:00:00.123456+00:00",
+ "2024-07-01
04:34:56.654321+00:00"}));
+ ASSERT_TRUE(reader.close().ok());
+ }
+ }
+ }
+ std::filesystem::remove_all(test_dir);
+}
+
+// The marker also reaches INT96 leaves inside a STRUCT and a LIST, while a
timestamp sibling in the
+// same struct keeps reading as the UTC wall clock Iceberg gives a timestamp
field.
+TEST(IcebergV2ReaderTest,
IcebergNestedInt96TimestampTzReadsInstantWithTableMarker) {
+ TimezoneUtils::load_timezones_to_cache();
+ const auto test_dir = std::filesystem::temp_directory_path() /
+
"doris_iceberg_nested_int96_timestamp_tz_marker_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ // 2024-01-01 00:00:00.123456 UTC.
+ write_nested_int96_parquet_file(file_path, 1'704'067'200'123'456L);
+ auto physical_reader = ::parquet::ParquetFileReader::OpenFile(file_path,
false);
+ for (int leaf = 1; leaf < 4; ++leaf) {
+ ASSERT_EQ(::parquet::Type::INT96,
+
physical_reader->metadata()->schema()->Column(leaf)->physical_type());
+ }
+ physical_reader->Close();
+
+ const auto tz_type = std::make_shared<DataTypeTimeStampTz>(6);
+ const auto wall_type = std::make_shared<DataTypeDateTimeV2>(6);
+ auto info = make_table_column(2, "info",
+ std::make_shared<DataTypeStruct>(DataTypes
{tz_type, wall_type},
+ Strings
{"tz", "wall"}));
+ info.children = {make_table_column(3, "tz", tz_type), make_table_column(4,
"wall", wall_type)};
+ auto tzs = make_table_column(5, "tzs",
std::make_shared<DataTypeArray>(tz_type));
+ tzs.children = {make_table_column(6, "element", tz_type)};
+ std::vector<ColumnDefinition> projected_columns {
+ make_table_column(1, "id", std::make_shared<DataTypeInt32>()),
info, tzs};
+
+ auto tz_field = external_schema_field("tz", 3);
+ tz_field.field_ptr->__set_timestamp_is_adjusted_to_utc(true);
+ auto element_field = external_schema_field("element", 6);
+ element_field.field_ptr->__set_timestamp_is_adjusted_to_utc(true);
+ auto tzs_field = external_schema_field("tzs", 5);
+ schema::external::TArrayField array_field;
+ array_field.__set_item_field(std::move(element_field));
+ tzs_field.field_ptr->nestedField.__set_array_field(std::move(array_field));
+ tzs_field.field_ptr->__isset.nestedField = true;
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_parquet_timestamp_semantics_version(1);
+ scan_params.__set_enable_mapping_timestamp_tz(true);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info({external_schema(
+ 100, {external_schema_field("id", 1),
+ external_struct_schema_field(
+ "info", 2, {std::move(tz_field),
external_schema_field("wall", 4)}),
+ std::move(tzs_field)})});
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ state.set_timezone("Asia/Shanghai");
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile);
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(file_path, {}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ // Table children are nullable, as in an external scan descriptor.
+ const auto nullable_tz_type = make_nullable(tz_type);
+ const auto nullable_wall_type = make_nullable(wall_type);
+ std::vector<std::string> values;
+ bool eos = false;
+ while (!eos) {
+ Block block = build_table_block(projected_columns);
+ const auto status = reader.get_block(&block, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ if (block.rows() == 0) {
+ continue;
+ }
+ ASSERT_EQ(block.rows(), 1);
+ const auto& info_column =
+ assert_cast<const
ColumnStruct&>(expect_not_null_table_column(block, 1));
+
values.push_back(nullable_tz_type->to_string(info_column.get_column(0), 0));
+
values.push_back(nullable_wall_type->to_string(info_column.get_column(1), 0));
+ const auto& tzs_column =
+ assert_cast<const
ColumnArray&>(expect_not_null_table_column(block, 2));
+ ASSERT_EQ(tzs_column.get_offsets()[0], 1);
+ values.push_back(nullable_tz_type->to_string(tzs_column.get_data(),
0));
+ }
+ EXPECT_EQ(values, std::vector<std::string>({"2024-01-01
00:00:00.123456+00:00",
+ "2024-01-01 00:00:00.123456",
+ "2024-01-01
00:00:00.123456+00:00"}));
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
+// TableColumnMapper resolves a nested field ID through an ID-less STRUCT
wrapper below the root as
+// well (ParquetRecursivelyRetainsIdlessWrapperForSelectedLeafId), so the
table's timestamptz marker
+// has to reach the leaf the same way. Otherwise the INT96 leaf decodes as a
UTC wall clock that the
+// projection reads in the session time zone, eight hours early in
Asia/Shanghai, and an equality
+// key through the wrapper compares that wall clock with the adjusted-UTC
tombstone an Iceberg
+// writer stores, so the tombstone misses its row.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity): read and
equality-delete matrix.
+TEST(IcebergV2ReaderTest,
IcebergInt96TimestampTzUnderIdlessWrapperReadsInstantWithTableMarker) {
+ TimezoneUtils::load_timezones_to_cache();
+ const auto test_dir = std::filesystem::temp_directory_path() /
+
"doris_iceberg_idless_wrapper_int96_timestamp_tz_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ const auto delete_file_path = (test_dir /
"equality-delete.parquet").string();
+ // 2024-01-01 00:00:00.123456 UTC and one second later.
+ constexpr int64_t T1_MICROS = 1'704'067'200'123'456L;
+ constexpr int64_t T2_MICROS = T1_MICROS + 1'000'000L;
+ write_idless_wrapper_int96_parquet_file(file_path, {T1_MICROS, T2_MICROS},
{1, 2});
+ write_iceberg_timestamp_equality_delete_parquet_file(delete_file_path, 30,
T1_MICROS, "ts",
+ true);
+
+ const auto tz_type = std::make_shared<DataTypeTimeStampTz>(6);
+ const auto nullable_tz_type = make_nullable(tz_type);
+ auto inner_type = std::make_shared<DataTypeStruct>(DataTypes {tz_type},
Strings {"ts"});
+ auto inner = make_table_column(20, "inner", inner_type);
+ inner.children = {make_table_column(30, "ts", tz_type)};
+ auto outer = make_table_column(
+ 10, "outer",
+ std::make_shared<DataTypeStruct>(DataTypes {inner_type}, Strings
{"inner"}));
+ outer.children = {inner};
+
+ struct Case {
+ bool with_delete;
+ bool project_outer;
+ std::vector<int32_t> ids;
+ std::vector<std::string> timestamps;
+ };
+ const std::vector<Case> cases = {
+ {false,
+ true,
+ {1, 2},
+ {"2024-01-01 00:00:00.123456+00:00", "2024-01-01
00:00:01.123456+00:00"}},
+ {true, true, {2}, {"2024-01-01 00:00:01.123456+00:00"}},
+ {true, false, {2}, {}},
+ };
+ for (const auto& c : cases) {
+ SCOPED_TRACE("with_delete=" + std::to_string(c.with_delete) +
+ " project_outer=" + std::to_string(c.project_outer));
+ std::vector<ColumnDefinition> projected_columns {
+ make_table_column(1, "id", std::make_shared<DataTypeInt32>())};
+ if (c.project_outer) {
+ projected_columns.push_back(outer);
+ }
+ auto ts_field =
+ external_schema_field("ts", 30, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::TIMESTAMPTZ, -1, 6));
+ ts_field.field_ptr->__set_timestamp_is_adjusted_to_utc(true);
+ auto inner_field = external_struct_schema_field("inner", 20,
{std::move(ts_field)});
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_parquet_timestamp_semantics_version(1);
+ scan_params.__set_enable_mapping_timestamp_tz(true);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info({external_schema(
+ 100, {external_schema_field("id", 1),
+ external_struct_schema_field("outer", 10,
{std::move(inner_field)})})});
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ state.set_timezone("Asia/Shanghai");
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile);
+ std::vector<TIcebergDeleteFileDesc> delete_files;
+ if (c.with_delete) {
+
delete_files.push_back(make_iceberg_equality_delete_file(delete_file_path,
{30}));
+ }
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+ split_options.current_range.__set_table_format_params(
+ make_iceberg_table_format_desc(file_path, delete_files));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> timestamps;
+ bool eos = false;
+ while (!eos) {
+ Block block = build_table_block(projected_columns);
+ const auto status = reader.get_block(&block, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ if (block.rows() == 0) {
+ continue;
+ }
+ const auto& id_column =
+ assert_cast<const
ColumnInt32&>(expect_not_null_table_column(block, 0));
+ for (size_t row = 0; row < block.rows(); ++row) {
+ ids.push_back(id_column.get_element(row));
+ }
+ if (!c.project_outer) {
+ continue;
+ }
+ const auto& outer_column =
+ assert_cast<const
ColumnStruct&>(expect_not_null_table_column(block, 1));
+ const auto& inner_column = assert_cast<const ColumnStruct&>(
+
expect_not_null_nullable_nested_column(outer_column.get_column(0)));
+ for (size_t row = 0; row < block.rows(); ++row) {
+
timestamps.push_back(nullable_tz_type->to_string(inner_column.get_column(0),
row));
+ }
+ }
+ EXPECT_EQ(ids, c.ids);
+ EXPECT_EQ(timestamps, c.timestamps);
+ ASSERT_TRUE(reader.close().ok());
+ }
+ std::filesystem::remove_all(test_dir);
+}
+
+// A Spark INT96 key and the adjusted-UTC tombstone an Iceberg writer stores
for the same
+// timestamptz instant decode to the same TIMESTAMPTZ once FE marks the field,
in both the data file
+// and the delete file. Under timestamp-semantics version 1 the data key used
to read as a UTC wall
+// clock and was converted back in the session time zone, so in Asia/Shanghai
the tombstone T1 missed
+// the row holding T1 by eight hours; with enable_mapping_timestamp_tz=false
the tombstone also read
+// as a Shanghai wall clock, and the two wall clocks never matched.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity): projection and
mapping-flag matrix.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteMatchesInt96TimestampTzKeyWithTableMarker) {
+ TimezoneUtils::load_timezones_to_cache();
+ const auto test_dir = std::filesystem::temp_directory_path() /
+ "doris_iceberg_equality_delete_int96_marker_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ const auto delete_file_path = (test_dir /
"equality-delete.parquet").string();
+ // 2024-01-01 00:00:00.123456 UTC and one second later.
+ constexpr int64_t T1_MICROS = 1'704'067'200'123'456L;
+ constexpr int64_t T2_MICROS = T1_MICROS + 1'000'000L;
+ write_timestamp_int_parquet_file(file_path, {T1_MICROS, T2_MICROS}, {1,
2}, true);
+ write_iceberg_timestamp_equality_delete_parquet_file(delete_file_path, 0,
T1_MICROS,
+ "event_time", true);
+
+ for (const bool project_key : {false, true}) {
+ for (const bool mapping_timestamp_tz : {true, false}) {
+ SCOPED_TRACE("project_key=" + std::to_string(project_key) +
+ " mapping_timestamp_tz=" +
std::to_string(mapping_timestamp_tz));
+ std::vector<ColumnDefinition> projected_columns {
+ make_table_column(1, "id",
std::make_shared<DataTypeInt32>())};
+ if (project_key) {
+ projected_columns.push_back(make_table_column(
+ 0, "event_time",
std::make_shared<DataTypeTimeStampTz>(6)));
+ }
+ auto event_time = external_schema_field(
+ "event_time", 0, {}, std::nullopt,
+ external_primitive_type(TPrimitiveType::TIMESTAMPTZ, -1,
6));
+ event_time.field_ptr->__set_timestamp_is_adjusted_to_utc(true);
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_parquet_timestamp_semantics_version(1);
+
scan_params.__set_enable_mapping_timestamp_tz(mapping_timestamp_tz);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info(
+ {external_schema(100, {event_time,
external_schema_field("id", 1)})});
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ state.set_timezone("Asia/Shanghai");
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats,
&file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params,
io_ctx, &state, &profile);
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+
split_options.current_range.__set_table_format_params(make_iceberg_table_format_desc(
+ file_path,
{make_iceberg_equality_delete_file(delete_file_path, {0})}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ std::vector<int32_t> ids;
+ std::vector<std::string> event_times;
+ bool eos = false;
+ while (!eos) {
+ Block block = build_table_block(projected_columns);
+ const auto status = reader.get_block(&block, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ if (block.rows() == 0) {
+ continue;
+ }
+ const auto& id_column =
+ assert_cast<const
ColumnInt32&>(expect_not_null_table_column(block, 0));
+ for (size_t row = 0; row < block.rows(); ++row) {
+ ids.push_back(id_column.get_element(row));
+ if (project_key) {
+
event_times.push_back(projected_columns[1].type->to_string(
+ *block.get_by_position(1).column, row));
+ }
+ }
+ }
+ EXPECT_EQ(ids, std::vector<int32_t>({2}));
+ if (project_key) {
+ EXPECT_EQ(event_times,
+ std::vector<std::string>({"2024-01-01
00:00:01.123456+00:00"}));
+ }
+ ASSERT_TRUE(reader.close().ok());
+ }
+ }
+ std::filesystem::remove_all(test_dir);
+}
+
+// A catalog with enable_mapping_timestamp_tz=false used to read both sides of
a timestamptz equality
+// delete as session-local wall clocks. During the 2024 New York fall-back
01:30 happens twice, so
+// either tombstone deleted both rows. Once FE marks the field, data keys and
tombstones both decode
+// as instants and each tombstone deletes only its own row. Decoding only the
data keys as instants
+// is not enough: the tombstone's wall clock maps back to one of the two
instants, so one of these
+// two cases would delete the wrong row.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteKeepsDstFoldInstantsApartWithTableMarker) {
+ TimezoneUtils::load_timezones_to_cache();
+ const auto test_dir = std::filesystem::temp_directory_path() /
+ "doris_iceberg_equality_delete_dst_fold_marker_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto file_path = (test_dir / "split.parquet").string();
+ // These instants are the two occurrences of 01:30 during the 2024 New
York DST fall-back.
+ constexpr int64_t FIRST_0130_MICROS = 1'730'611'800'000'000L;
+ constexpr int64_t SECOND_0130_MICROS = 1'730'615'400'000'000L;
+ write_timestamp_int_parquet_file(file_path, {FIRST_0130_MICROS,
SECOND_0130_MICROS}, {1, 2});
+
+ for (const auto& [tombstone, remaining_id] :
+ {std::pair {FIRST_0130_MICROS, 2}, std::pair {SECOND_0130_MICROS,
1}}) {
+ SCOPED_TRACE(tombstone);
+ // Same size and second-granularity mtime: one path for both files
could serve the first
+ // tombstone from the Parquet caches.
+ const auto delete_file_path =
+ (test_dir / ("equality-delete-" + std::to_string(remaining_id)
+ ".parquet"))
+ .string();
+ write_iceberg_timestamp_equality_delete_parquet_file(delete_file_path,
0, tombstone,
+ "event_time",
true);
+ std::vector<ColumnDefinition> projected_columns {
+ make_table_column(1, "id", std::make_shared<DataTypeInt32>())};
+ auto event_time =
+ external_schema_field("event_time", 0, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::TIMESTAMPTZ, -1, 6));
+ event_time.field_ptr->__set_timestamp_is_adjusted_to_utc(true);
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_parquet_timestamp_semantics_version(1);
+ scan_params.__set_enable_mapping_timestamp_tz(false);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info(
+ {external_schema(100, {event_time, external_schema_field("id",
1)})});
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ state.set_timezone("America/New_York");
+ io::FileReaderStats file_reader_stats;
+ io::FileCacheStatistics file_cache_stats;
+ auto io_ctx = make_io_context(&file_reader_stats, &file_cache_stats);
+ ShardedKVCache cache(1);
+ doris::format::iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, projected_columns, &scan_params, io_ctx,
&state, &profile);
+ auto split_options = build_split_options(file_path);
+ split_options.cache = &cache;
+
split_options.current_range.__set_table_format_params(make_iceberg_table_format_desc(
+ file_path,
{make_iceberg_equality_delete_file(delete_file_path, {0})}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+ EXPECT_EQ(read_iceberg_ids(&reader, projected_columns),
+ std::vector<int32_t>({remaining_id}));
+ ASSERT_TRUE(reader.close().ok());
+ }
+ std::filesystem::remove_all(test_dir);
+}
+
TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteUsesNameMappingWithoutFileFieldIds) {
const auto test_dir = std::filesystem::temp_directory_path() /
"doris_iceberg_equality_delete_name_mapping_test";
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java
index 1be990d5554..02e0604c8ac 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java
@@ -81,7 +81,9 @@ import java.util.Optional;
* consumes — {@code id} / {@code name} / the mapped Doris column type /
{@code name_mapping} — and, faithful
* to legacy {@code ExternalUtil}, an {@code id}/{@code name} at EVERY nesting
level (array element, map
* key/value, struct child), unlike paimon which omits them on collection
elements. The scalar type is also
- * required when BE parses an Iceberg initial default or resolves a promoted
equality-delete field.</p>
+ * required when BE parses an Iceberg initial default or resolves a promoted
equality-delete field. A
+ * timestamptz field also carries {@code timestamp_is_adjusted_to_utc}, which
BE needs to read INT96 as an
+ * instant.</p>
*/
public final class IcebergSchemaUtils {
@@ -393,6 +395,14 @@ public final class IcebergSchemaUtils {
TColumnType columnType = new TColumnType();
if (type.isPrimitiveType()) {
tField.setType(buildPrimitiveColumnType(type, enableVarbinary,
enableTimestampTz));
+ if (type.typeId() == TypeID.TIMESTAMP && ((Types.TimestampType)
type).shouldAdjustToUTC()) {
+ // Iceberg decodes a Parquet timestamp by the table type, not
by the file annotation. Spark and Hive
+ // write timestamptz as INT96, which has no annotation, so
without this marker BE reads such a
+ // column as a UTC wall clock and shifts it by the session
offset. timestamp fields stay unmarked:
+ // BE would then read an adjusted-UTC INT64 file as a UTC wall
clock in its statistics but as a
+ // session-local wall clock in its rows.
+ tField.setTimestampIsAdjustedToUtc(true);
+ }
return tField;
}
if (type.isVariantType()) {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
index 832ed2149bc..d45eaccf68f 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
@@ -676,6 +676,48 @@ public class IcebergSchemaUtilsTest {
props.getNestedField().getMapField().getValueField().getFieldPtr().getId());
}
+ @Test
+ public void timestamptzFieldsCarryInstantMarkerAtEveryLevel() {
+ // Spark and Hive write timestamptz as INT96 without an instant
annotation, so BE reads it as an instant only
+ // when the field says so. timestamp and timestamp_ns stay unmarked
whatever the catalog mapping flag is.
+ // MUTATION: drop the marker -> red; also mark timestamp fields -> red.
+ Schema schema = new Schema(
+ Types.NestedField.optional(1, "tz",
Types.TimestampType.withZone()),
+ Types.NestedField.optional(2, "wall",
Types.TimestampType.withoutZone()),
+ Types.NestedField.optional(3, "wall_ns",
Types.TimestampNanoType.withoutZone()),
+ Types.NestedField.optional(4, "info", Types.StructType.of(
+ Types.NestedField.optional(5, "tz",
Types.TimestampType.withZone()),
+ Types.NestedField.optional(6, "wall",
Types.TimestampType.withoutZone()))),
+ Types.NestedField.optional(7, "tzs",
+ Types.ListType.ofOptional(8,
Types.TimestampType.withZone())),
+ Types.NestedField.optional(9, "by_tz",
Types.MapType.ofOptional(10, 11,
+ Types.TimestampType.withZone(),
Types.TimestampType.withoutZone())));
+
+ for (boolean enableTimestampTz : new boolean[] {false, true}) {
+ Map<String, TField> fields =
topFields(IcebergSchemaUtils.buildCurrentSchema(schema,
+ Arrays.asList("tz", "wall", "wall_ns", "info", "tzs",
"by_tz"), Collections.emptyMap(),
+ false, false, enableTimestampTz));
+
+ assertInstantMarker(fields.get("tz"));
+
Assertions.assertFalse(fields.get("wall").isSetTimestampIsAdjustedToUtc());
+
Assertions.assertFalse(fields.get("wall_ns").isSetTimestampIsAdjustedToUtc());
+ TField info = fields.get("info");
+ Assertions.assertFalse(info.isSetTimestampIsAdjustedToUtc());
+ assertInstantMarker(childByName(info, "tz"));
+ Assertions.assertFalse(childByName(info,
"wall").isSetTimestampIsAdjustedToUtc());
+
assertInstantMarker(fields.get("tzs").getNestedField().getArrayField().getItemField().getFieldPtr());
+ TField byTz = fields.get("by_tz");
+
assertInstantMarker(byTz.getNestedField().getMapField().getKeyField().getFieldPtr());
+
Assertions.assertFalse(byTz.getNestedField().getMapField().getValueField().getFieldPtr()
+ .isSetTimestampIsAdjustedToUtc());
+ }
+ }
+
+ private static void assertInstantMarker(TField field) {
+ Assertions.assertTrue(field.isSetTimestampIsAdjustedToUtc(),
field.getName());
+ Assertions.assertTrue(field.isTimestampIsAdjustedToUtc(),
field.getName());
+ }
+
@Test
public void nestedStructChildNamesAreLowercased() {
// The crash repro (test_iceberg_struct_schema_evolution /
DROP_AND_ADD): a struct child whose iceberg
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
