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 978d5c8d34b [fix](be) Compare Iceberg equality-delete keys in the
wider of the delete and data file types (#69711)
978d5c8d34b is described below
commit 978d5c8d34b90de606e466b704b4a9a72d6bc233
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Oct 29 08:11:01 2026 +0000
[fix](be) Compare Iceberg equality-delete keys in the wider of the delete
and data file types (#69711)
### What problem does this PR solve?
Issue Number: None
Related PR: #69536, #68267
Problem Summary:
**In short.** FileScannerV2 now compares an Iceberg equality-delete key
in the wider of the delete file's and the data file's own key types,
instead of in the one table type the FE sends for the whole scan. For
some table histories that single type is wrong: the reader then deletes
rows it must keep, keeps rows it must delete, or fails the scan. This
follows the review of #69536, where these cases came up.
**Background.** An equality delete removes a data row whose key values
equal a tombstone's. After a type promotion the two sides can be stored
in different types: tombstones written before an INT -> LONG promotion
hold INT values that must still match LONG data. #68267 made
FileScannerV2 compare in the current schema's type, and for a key that
the current schema no longer has, the FE sends a type recovered from the
retained schemas (#69536). The reader casts both the tombstones and the
data keys to that one type.
**The problem, and what it cost.** One type per scan cannot describe
every delete file:
| History | Table type from the FE | Delete file | Data file | Before |
After |
|---|---|---|---|---|---|
| `expireSnapshots().cleanExpiredMetadata(true)` removed the schema of a
schema-only INT -> LONG promotion of a dropped key | INT | LONG
4294967296 | INT, one NULL key | the tombstone becomes NULL in INT and
deletes the NULL-key row | row kept |
| A read pinned before a later DATE -> TIMESTAMP_NS promotion of the key
| TIMESTAMP_NS | DATE 1000-01-01 | DATE 1000-01-01 | scan fails:
`TIMESTAMP_NS overflow when casting row ...` | row deleted |
| No retained schema gives the key a concrete type, so the FE sends
Iceberg `unknown` as STRING | STRING | TIMESTAMP_NS 2024-01-02 00:00 |
DATE 2024-01-02 | `2024-01-02` != `2024-01-02 00:00:00.000000000`, the
row stays visible | row deleted |
| The delete was prepared under a TIMESTAMP_NS schema that the table
later abandoned for TIMESTAMP | TIMESTAMP | TIMESTAMP_NS
12:00:00.000000500 | TIMESTAMP 12:00:00.000000 | the tombstone is
rounded to microseconds and deletes the row | row kept |
Three of the four return wrong rows without any error. The V1 reader is
not changed by this PR.
**How this PR fixes it.**
- The delete block keeps the delete file's own key types. When the
predicate for a data file is built, each key is compared in the wider of
the delete file's type and the data file's type
(`equality_delete_comparison_type`). The accepted widenings are the
Iceberg promotions (INT -> LONG, FLOAT -> DOUBLE, decimal precision,
DATE -> TIMESTAMP and DATE -> TIMESTAMP_NS), a wider fractional scale of
the same date-time kind, wider integers, STRING -> VARBINARY (an Iceberg
BINARY key decodes as either one, depending on
`enable_mapping_varbinary`), and TIMESTAMP -> TIMESTAMP_NS. The last one
is not an Iceberg promotion; it is accepted for the abandoned-schema
case above, because scaling microseconds to nanoseconds is exact. When
the table field is `timestamptz`, a `DATETIMEV2` key (Parquet INT96,
which carries no instant annotation, as Spark and Hive write it) and a
`TIMESTAMPTZ` key compare as `TIMESTAMPTZ` at the larger scale, the
`DATETIMEV2` side converted with the session time zone as the projection
converts that column. When the table field is the STRING placeholder the
FE sends for an Iceberg TIME, which Doris cannot represent, an ORC TIME
key (a LONG annotated `iceberg.long-type=TIME`, read as `BIGINT`) and a
Parquet one (`TIME(MICROS)`, read as `TIMEV2(6)`) compare as
`TIMEV2(6)`, the `BIGINT` side converted exactly as microseconds from
midnight by a dedicated expression (`CAST(BIGINT AS TIME)` reads the
integer's digits as HHMMSS), and a value outside
00:00:00..23:59:59.999999 fails the scan. Two types that convert in
neither direction fail the scan instead of being compared in a third
type.
- The data keys are cast to that type. When it is the data file's wider
type, the tombstones are widened once per scanner, and the converted
delete block and its hash index are cached next to the original
(`_convert_equality_delete_filter`). A table without type changes is
compared as before, with no cast and no copy.
- A key that the data file lacks is compared through its initial
default, in the wider of the delete file's type and the table type. When
that default is NULL (no initial default, or a struct default `{}` whose
leaf has none) the key is NULL in that file whatever its table type, so
it is compared in the delete file's type (an `unknown` key arrives as
STRING). A non-NULL default behind the STRING placeholder of an Iceberg
TIME is Iceberg's text of the value (`00:00`), with at most six
fractional digits, so it is parsed exactly in the tombstones' type,
`TIMEV2` (`TIMEV2(6)` against an ORC TIME tombstone), by the strict
comparison cast. A non-NULL default of a `timestamptz_ns` key fails the
scan with `NotSupported`: V2 reads its tombstones as `TIMESTAMPTZ(6)`
(Parquet floors the nanoseconds, ORC rounds them), so no parse of the
nanosecond default compares exactly; the default epoch would match a
tombstone 100 ns later and delete every row of the file.
- `AncestorNullDefaultExpr`, which gives a missing child the NULL of its
NULL parent and its default otherwise, now opens its children. Its
`open()` returned before opening them, because `prepare()` had already
set the flag its guard checks, so a default cast to the comparison type
below it was never opened and aborted the BE with `Check failed:
_open_finished` in `Cast`.
- The comparison casts are strict whatever the session's cast mode,
through a new `strict` flag on `EvolutionCast`: a non-strict overflow
would turn a key into NULL and let a NULL tombstone delete it. DATE ->
TIMESTAMP_NS was already strict; other `EvolutionCast` users are
unchanged.
- A delete file without rows is skipped before its keys are resolved: it
deletes nothing, but the predicate evaluates its keys first, so its
strict casts could fail the scan (a DATE key 1000-01-01 against an empty
TIMESTAMP_NS delete file).
- The table types from the FE still resolve table fields and the
defaults of missing keys; the delete-file cache key is unchanged.
**Results.**
- 14 new BE unit tests in `iceberg_reader_test.cpp`: the four histories
above; INT tombstones against LONG, INT and LONG data files of one scan
(the third split reuses the cached widened copy); a missing key's INT
default against a LONG tombstone; a missing key without a default under
an `unknown` table type; a microsecond key outside the nanosecond range,
which must fail; and incomparable key types. With the reader before this
PR, 6 of them fail; the other three pin behavior the old code shares.
Removing the TIMESTAMP -> TIMESTAMP_NS widening fails 2, and making the
casts non-strict fails 1. Three more came from review: an INT96 data key
against an adjusted-UTC tombstone in Asia/Shanghai, a NULL leaf under a
struct default `{}`, and an empty TIMESTAMP_NS delete file against a
DATE key; removing the INT96 rule, the NULL-leaf walk or the empty-file
skip fails its test. Two more came from the third round: Iceberg TIME
keys across all four ORC / Parquet data and delete file pairs, with NULL
and non-NULL tombstones, and a nested Iceberg TIME key added with a text
default, against Parquet TIME and ORC TIME tombstones, under NULL and
non-NULL parents. Removing the TIME pair rule or the default parsing, or
converting the ORC TIME with a SQL cast, fails them; leaving
`AncestorNullDefaultExpr`'s children unopened aborts. One more came from
the fourth round: a nested `timestamptz_ns` key with the defaults
`...0000001`, the epoch and `...0000005` fails against an epoch, a 100
ns and a 500 ns tombstone, while a key without a default still matches
NULL tombstones only; parsing those defaults again fails it.
- End to end on a local cluster: a v2 table in a `file://` Hadoop
catalog, built with the Iceberg 1.12.0 Java API: `k INT` with rows (1,
NULL), (2, 0), (3, 1); `k` promoted to LONG without a snapshot; a delete
of `k = 4294967296` written under that schema; `k` dropped; the delete
committed; then
`expireSnapshots().retainLast(2).cleanExpiredMetadata(true)`, which
leaves schema 0 (`k INT`) and schema 2 (no `k`). `SELECT id` through
FileScannerV2 returned 2, 3 before this PR and returns 1, 2, 3 with it.
branch-4.1 and branch-4.2 have the same FileScannerV2 code.
### Release note
Fix FileScannerV2 comparing Iceberg equality-delete keys in a single
scan-wide type, which could delete or keep the wrong rows, or fail the
scan, after a key's type changed (for example a schema-only promotion
removed by `cleanExpiredMetadata`, or a key whose retained schemas only
have it as `unknown`).
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] Unit Test
- `run-be-ut.sh --run
--filter='IcebergV2ReaderTest.*:*EqualityDelete*:*ColumnMapper*:*EvolutionCast*:*TableReader*'`:
526 of 526 pass on macOS arm64 (ASAN), including the 14 new cases.
- Rebased onto master `43f9ceb48d7`: #69639 moved the strict-mode
override of `EvolutionCast` from `prepare()` into
`configure_cast_options()`, so the new `strict` flag now sets
`CastOptions::strict_mode` there, next to the existing DATE ->
TIMESTAMP_NS rule. The run above is on the content before the rebase
(the third-round commit applies to both unchanged); the 67 TUs that this
PR changes or that master changed since `5caaeb3e3d4` were
syntax-checked against the rebased tree with the UT build flags, and CI
runs the suites again.
- [x] Manual test (add detailed scripts or steps below)
- The local end-to-end check described under Results, with an ASAN BE
built from this change.
- [ ] No need to test or manual test. Explain why:
- Behavior changed:
- [ ] No.
- [x] Yes. FileScannerV2 compares equality-delete keys per delete file
and data file; two key types that convert in neither direction now fail
the scan; the comparison casts are strict.
- 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/expr/evolution_cast.cpp | 6 +-
be/src/format_v2/expr/evolution_cast.h | 15 +-
be/src/format_v2/table/iceberg_reader.cpp | 542 +++++++++++++--
be/src/format_v2/table/iceberg_reader.h | 20 +-
be/test/format_v2/table/iceberg_reader_test.cpp | 836 +++++++++++++++++++++++-
5 files changed, 1350 insertions(+), 69 deletions(-)
diff --git a/be/src/format_v2/expr/evolution_cast.cpp
b/be/src/format_v2/expr/evolution_cast.cpp
index d8d3133b5e0..b845fa9706b 100644
--- a/be/src/format_v2/expr/evolution_cast.cpp
+++ b/be/src/format_v2/expr/evolution_cast.cpp
@@ -75,8 +75,10 @@ Status EvolutionCast::execute_column_impl(VExprContext*
context, const Block* bl
}
void EvolutionCast::configure_cast_options(CastOptions& options) const {
- if (remove_nullable(_children[0]->data_type())->get_primitive_type() ==
TYPE_DATEV2 &&
- remove_nullable(_data_type)->get_primitive_type() ==
TYPE_TIMESTAMP_NS) {
+ if (_strict) {
+ options.strict_mode = true;
+ } else if
(remove_nullable(_children[0]->data_type())->get_primitive_type() ==
TYPE_DATEV2 &&
+ remove_nullable(_data_type)->get_primitive_type() ==
TYPE_TIMESTAMP_NS) {
// Iceberg DATE promotion must fail outside the nanosecond range, even
in non-strict SQL
// sessions. Otherwise a localized IS NOT NULL filter can silently
discard retained rows.
options.strict_mode = true;
diff --git a/be/src/format_v2/expr/evolution_cast.h
b/be/src/format_v2/expr/evolution_cast.h
index c7e0e306018..63616fc1b8c 100644
--- a/be/src/format_v2/expr/evolution_cast.h
+++ b/be/src/format_v2/expr/evolution_cast.h
@@ -26,15 +26,19 @@ class EvolutionCast final : public Cast {
ENABLE_FACTORY_CREATOR(EvolutionCast);
public:
- explicit EvolutionCast(const DataTypePtr& type, bool
truncate_datetimev2_precision = false)
- : Cast(type),
_truncate_datetimev2_precision(truncate_datetimev2_precision) {}
+ explicit EvolutionCast(const DataTypePtr& type, bool
truncate_datetimev2_precision = false,
+ bool strict = false)
+ : Cast(type),
+ _truncate_datetimev2_precision(truncate_datetimev2_precision),
+ _strict(strict) {}
Status execute_column_impl(VExprContext* context, const Block* block,
const Selector* selector,
size_t count, ColumnPtr& result_column) const
override;
std::string debug_string() const override;
Status clone_node(VExprSPtr* cloned_expr) const override {
DORIS_CHECK(cloned_expr != nullptr);
- *cloned_expr = EvolutionCast::create_shared(_data_type,
_truncate_datetimev2_precision);
+ *cloned_expr =
+ EvolutionCast::create_shared(_data_type,
_truncate_datetimev2_precision, _strict);
return Status::OK();
}
@@ -43,6 +47,11 @@ protected:
private:
bool _truncate_datetimev2_precision = false;
+ // Fail on every value the target type cannot hold, whatever the session's
strict cast mode.
+ // Iceberg equality deletes compare keys after a widening cast, where a
non-strict overflow
+ // would turn a key into NULL and match NULL tombstones: a data key
TIMESTAMP 1600-01-01 cast
+ // to TIMESTAMP_NS would become NULL and be deleted by a NULL tombstone.
+ bool _strict = false;
};
} // namespace doris::format
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index a879a0d1f89..9463fe0d9b0 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -42,8 +42,11 @@
#include "core/data_type/data_type_map.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_struct.h"
+#include "core/data_type/data_type_time.h"
+#include "core/data_type/data_type_timestamptz.h"
#include "core/data_type/define_primitive_type.h"
#include "core/field.h"
+#include "core/value/time_value.h"
#include "exprs/vliteral.h"
#include "exprs/vslot_ref.h"
#include "format/table/deletion_vector_reader.h"
@@ -846,7 +849,12 @@ public:
Status open(RuntimeState* state, VExprContext* context,
FunctionContext::FunctionStateScope scope) override {
- RETURN_IF_ERROR_OR_PREPARED(VExpr::open(state, context, scope));
+ DCHECK(_prepare_finished);
+ // Opens the children, as EqualityDeletePredicate does.
RETURN_IF_ERROR_OR_PREPARED would
+ // return at once, since prepare() has set `_prepared`, leaving a
default cast to the
+ // comparison type, e.g. the TIME text `00:00` parsed as TIMEV2,
unopened, so executing it
+ // fails "Check failed: _open_finished" in Cast.
+ RETURN_IF_ERROR(VExpr::open(state, context, scope));
_open_finished = true;
return Status::OK();
}
@@ -915,6 +923,108 @@ private:
std::string _expr_name;
};
+// Converts an Iceberg TIME read from ORC, microseconds from midnight as
BIGINT, to TIMEV2, which
+// holds microseconds as well (as a double, exactly): 3600000001 becomes
01:00:00.000001. An
+// ordinary CAST(BIGINT AS TIME) reads the integer's decimal digits as HHMMSS
instead, turning
+// 3600 into 00:36:00. A value outside 00:00:00..23:59:59.999999 is no Iceberg
TIME and fails, as
+// the strict comparison casts do.
+class IcebergTimeMicrosExpr final : public VExpr {
+public:
+ IcebergTimeMicrosExpr(DataTypePtr data_type, std::string expr_name)
+ : VExpr(std::move(data_type), false),
_expr_name(std::move(expr_name)) {
+ _node_type = TExprNodeType::FUNCTION_CALL;
+ DORIS_CHECK(remove_nullable(_data_type)->get_primitive_type() ==
TYPE_TIMEV2);
+ }
+
+ Status prepare(RuntimeState* state, const RowDescriptor& row_desc,
+ VExprContext* context) override {
+ RETURN_IF_ERROR_OR_PREPARED(VExpr::prepare(state, row_desc, context));
+ _prepare_finished = true;
+ return Status::OK();
+ }
+
+ Status open(RuntimeState* state, VExprContext* context,
+ FunctionContext::FunctionStateScope scope) override {
+ DCHECK(_prepare_finished);
+ // As in AncestorNullDefaultExpr: the child, e.g. a nested key's
NestedStructFieldExpr,
+ // must be opened, which RETURN_IF_ERROR_OR_PREPARED would skip after
prepare().
+ RETURN_IF_ERROR(VExpr::open(state, context, scope));
+ _open_finished = true;
+ return Status::OK();
+ }
+
+ void close(VExprContext* context, FunctionContext::FunctionStateScope
scope) override {
+ VExpr::close(context, scope);
+ }
+
+ Status execute_column_impl(VExprContext* context, const Block* block,
const Selector* selector,
+ size_t count, ColumnPtr& result_column) const
override {
+ DORIS_CHECK(_children.size() == 1);
+ ColumnPtr input;
+ RETURN_IF_ERROR(_children.front()->execute_column(context, block,
selector, count, input));
+ input = input->convert_to_full_column_if_const();
+ DORIS_CHECK(input->size() == count);
+ const NullMap* null_map = nullptr;
+ const IColumn* micros_column = input.get();
+ if (const auto* nullable =
check_and_get_column<ColumnNullable>(*input);
+ nullable != nullptr) {
+ null_map = &nullable->get_null_map_data();
+ micros_column = &nullable->get_nested_column();
+ }
+ DORIS_CHECK(_data_type->is_nullable() == (null_map != nullptr));
+ const auto& micros = assert_cast<const
ColumnInt64&>(*micros_column).get_data();
+ auto times = ColumnTimeV2::create();
+ auto& values = times->get_data();
+ values.resize_fill(count, 0);
+ for (size_t row = 0; row < count; ++row) {
+ if (null_map != nullptr && (*null_map)[row] != 0) {
+ continue;
+ }
+ if (micros[row] < 0 || micros[row] >= MICROS_PER_DAY) {
+ return Status::InvalidArgument(
+ "Iceberg TIME value {} of {} is outside
00:00:00..23:59:59.999999",
+ micros[row], _expr_name);
+ }
+ values[row] = static_cast<double>(micros[row]);
+ }
+ if (null_map == nullptr) {
+ result_column = std::move(times);
+ } else {
+ auto nulls = ColumnUInt8::create();
+ nulls->get_data().assign(null_map->begin(), null_map->end());
+ result_column = ColumnNullable::create(std::move(times),
std::move(nulls));
+ }
+ return Status::OK();
+ }
+
+ const std::string& expr_name() const override { return _expr_name; }
+
+ Status clone_node(VExprSPtr* cloned_expr) const override {
+ DORIS_CHECK(cloned_expr != nullptr);
+ *cloned_expr = std::make_shared<IcebergTimeMicrosExpr>(*this);
+ return Status::OK();
+ }
+
+private:
+ static constexpr int64_t MICROS_PER_DAY = 24 *
TimeValue::ONE_HOUR_MICROSECONDS;
+
+ std::string _expr_name;
+};
+
+// The struct element of `parent` that holds `child` on an equality key's
path: matched by field id
+// when both carry one, otherwise by name.
+static size_t equality_delete_child_index(const format::ColumnDefinition&
parent,
+ const format::ColumnDefinition&
child) {
+ const auto child_it = std::ranges::find_if(parent.children, [&child](const
auto& candidate) {
+ if (candidate.has_identifier_field_id() &&
child.has_identifier_field_id()) {
+ return candidate.get_identifier_field_id() ==
child.get_identifier_field_id();
+ }
+ return candidate.name == child.name;
+ });
+ DORIS_CHECK(child_it != parent.children.end());
+ return cast_set<size_t>(child_it - parent.children.begin());
+}
+
static Status build_nested_equality_delete_key_expr(
const std::vector<const format::ColumnDefinition*>& path, VExprSPtr
root_expr,
VExprSPtr* key_expr) {
@@ -939,16 +1049,7 @@ static Status build_nested_equality_delete_key_expr(
"Iceberg equality delete field {} has non-struct ancestor
{}", child->name,
parent->name);
}
- const auto child_it =
- std::ranges::find_if(parent->children, [child](const auto&
candidate) {
- if (candidate.has_identifier_field_id() &&
child->has_identifier_field_id()) {
- return candidate.get_identifier_field_id() ==
- child->get_identifier_field_id();
- }
- return candidate.name == child->name;
- });
- DORIS_CHECK(child_it != parent->children.end());
- child_indexes.push_back(cast_set<size_t>(child_it -
parent->children.begin()));
+ child_indexes.push_back(equality_delete_child_index(*parent, *child));
expr_name += "." + child->name;
}
if (!child_indexes.empty()) {
@@ -974,6 +1075,208 @@ static Status build_equality_delete_key_expr(
return build_nested_equality_delete_key_expr(path, std::move(root_expr),
key_expr);
}
+// 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
+// each child's own default, so payload.k is NULL when `k` has none.
+static Status equality_delete_default_leaf_is_null(
+ const std::vector<const format::ColumnDefinition*>& missing_path,
bool* is_null) {
+ DORIS_CHECK(!missing_path.empty());
+ DORIS_CHECK(is_null != nullptr);
+ const auto* root = missing_path.front();
+ DORIS_CHECK(root != nullptr);
+ if (!root->initial_default_value.has_value()) {
+ *is_null = true;
+ return Status::OK();
+ }
+ std::deque<std::string> binary_storage;
+ Field root_default;
+ RETURN_IF_ERROR(
+ build_v2_initial_default_field(*root, root->type, &binary_storage,
&root_default));
+ const Field* value = &root_default;
+ for (size_t index = 1; index < missing_path.size() && !value->is_null();
++index) {
+ DORIS_CHECK(value->get_type() == TYPE_STRUCT);
+ const auto& children = value->get<TYPE_STRUCT>();
+ const size_t child_index =
+ equality_delete_child_index(*missing_path[index - 1],
*missing_path[index]);
+ DORIS_CHECK(child_index < children.size());
+ value = &children[child_index];
+ }
+ *is_null = value->is_null();
+ return Status::OK();
+}
+
+static int equality_delete_integer_width(PrimitiveType type) {
+ switch (type) {
+ case TYPE_TINYINT:
+ return 1;
+ case TYPE_SMALLINT:
+ return 2;
+ case TYPE_INT:
+ return 4;
+ case TYPE_BIGINT:
+ return 8;
+ case TYPE_LARGEINT:
+ return 16;
+ default:
+ return 0;
+ }
+}
+
+// Whether every `from` value converts exactly to a `to` value that compares
equal to the same
+// Iceberg value. These are the Iceberg promotions (INT -> LONG, FLOAT ->
DOUBLE, decimal
+// precision, DATE -> TIMESTAMP and DATE -> TIMESTAMP_NS), a wider fractional
scale of the same
+// date-time kind, and STRING -> VARBINARY: an Iceberg BINARY key decodes as
either one, depending
+// on enable_mapping_varbinary, and both hold the same bytes. TIMESTAMP ->
TIMESTAMP_NS is not an
+// Iceberg promotion, but a delete can be prepared under a nanosecond schema
that the table later
+// abandons. Scaling microseconds by 1000 is exact, whereas rounding the
tombstone
+// 12:00:00.000000500 to microseconds would delete the data key
12:00:00.000000.
+static bool is_equality_delete_widening(const DataTypePtr& from, const
DataTypePtr& to) {
+ const auto from_type = remove_nullable(from);
+ const auto to_type = remove_nullable(to);
+ if (from_type->equals(*to_type)) {
+ return true;
+ }
+ const auto from_primitive = from_type->get_primitive_type();
+ const auto to_primitive = to_type->get_primitive_type();
+ if (const int from_width = equality_delete_integer_width(from_primitive);
from_width > 0) {
+ return equality_delete_integer_width(to_primitive) >= from_width;
+ }
+ switch (from_primitive) {
+ case TYPE_FLOAT:
+ return to_primitive == TYPE_DOUBLE;
+ case TYPE_DECIMAL32:
+ case TYPE_DECIMAL64:
+ case TYPE_DECIMAL128I:
+ case TYPE_DECIMAL256:
+ return is_decimalv3(to_primitive) && to_type->get_scale() ==
from_type->get_scale() &&
+ to_type->get_precision() >= from_type->get_precision();
+ case TYPE_DATEV2:
+ return to_primitive == TYPE_DATETIMEV2 || to_primitive ==
TYPE_TIMESTAMP_NS;
+ case TYPE_DATETIMEV2:
+ return (to_primitive == TYPE_DATETIMEV2 &&
+ to_type->get_scale() >= from_type->get_scale()) ||
+ to_primitive == TYPE_TIMESTAMP_NS;
+ case TYPE_TIMESTAMPTZ:
+ case TYPE_TIMEV2:
+ return to_primitive == from_primitive && to_type->get_scale() >=
from_type->get_scale();
+ case TYPE_VARCHAR:
+ return to_primitive == TYPE_STRING || to_primitive == TYPE_VARBINARY;
+ case TYPE_STRING:
+ return to_primitive == TYPE_VARBINARY;
+ default:
+ return false;
+ }
+}
+
+// Iceberg equality is by value, so a key is compared in the wider of the
delete file's and the
+// data file's key types, and neither side is narrowed. One scan-wide table
type cannot serve: after
+// expireSnapshots().cleanExpiredMetadata(true) drops the schema of a
schema-only INT -> LONG
+// promotion, the table type of a dropped key is INT while its tombstones are
LONG, and casting
+// 4294967296 to INT yields NULL, which deletes the rows whose key is NULL.
`table_type` is the
+// key's type in the scan's schema metadata; it decides only how a timestamptz
field read as
+// DATETIMEV2 and a TIME field behind a STRING placeholder compare.
+static Status equality_delete_comparison_type(int field_id, const DataTypePtr&
delete_type,
+ const DataTypePtr& data_type,
+ const DataTypePtr& table_type,
+ const std::string&
data_file_path,
+ DataTypePtr* comparison_type) {
+ DORIS_CHECK(delete_type != nullptr);
+ DORIS_CHECK(data_type != nullptr);
+ DORIS_CHECK(table_type != nullptr);
+ DORIS_CHECK(comparison_type != nullptr);
+ if (is_equality_delete_widening(delete_type, data_type)) {
+ *comparison_type = remove_nullable(data_type);
+ return Status::OK();
+ }
+ if (is_equality_delete_widening(data_type, delete_type)) {
+ *comparison_type = remove_nullable(delete_type);
+ return Status::OK();
+ }
+ const auto delete_value_type = remove_nullable(delete_type);
+ const auto data_value_type = remove_nullable(data_type);
+ const auto delete_primitive = delete_value_type->get_primitive_type();
+ const auto data_primitive = data_value_type->get_primitive_type();
+ // Parquet INT96 carries no instant annotation, so a timestamptz key in a
Spark or Hive file
+ // that Iceberg imported reads as DATETIMEV2(6), while an adjusted-UTC
tombstone of the same
+ // field, as a Flink upsert writes it, reads as TIMESTAMPTZ(6). Neither
widens to the other.
+ // 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.
+ 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))) {
+ *comparison_type = std::make_shared<DataTypeTimeStampTz>(
+ std::max(delete_value_type->get_scale(),
data_value_type->get_scale()));
+ return Status::OK();
+ }
+ // The FE sends an Iceberg TIME, which Doris cannot represent, as a STRING
placeholder, so only
+ // the files type its values: microseconds from midnight in both formats,
but ORC stores them as
+ // a LONG annotated iceberg.long-type=TIME, which reads as BIGINT, while
Parquet's TIME(MICROS)
+ // reads as TIMEV2(6). Such a pair compares as TIMEV2(6), which holds
microseconds too, so the
+ // ORC key 3600000001 matches the Parquet tombstone 01:00:00.000001;
cast_equality_delete_key
+ // converts the BIGINT side without a SQL cast. Under any other table type
the pair fails below.
+ if (remove_nullable(table_type)->get_primitive_type() == TYPE_STRING &&
+ ((delete_primitive == TYPE_BIGINT && data_primitive == TYPE_TIMEV2) ||
+ (delete_primitive == TYPE_TIMEV2 && data_primitive == TYPE_BIGINT))) {
+ *comparison_type =
std::make_shared<DataTypeTimeV2>(TimeValue::MICROS_SCALE);
+ return Status::OK();
+ }
+ // No promotion leads from one type to the other, so no third type
compares them faithfully:
+ // STRING tombstones cast to an INT key would delete the row with key 1
for the string "1".
+ return Status::NotSupported(
+ "Iceberg equality delete field id {} is {} in the delete file and
{} for data file "
+ "{}, and neither type converts exactly to the other",
+ field_id, remove_nullable(delete_type)->get_name(),
+ remove_nullable(data_type)->get_name(), data_file_path);
+}
+
+// The comparison type of a missing key whose non-NULL initial default sits
behind the STRING
+// placeholder of an Iceberg TIME (see equality_delete_comparison_type), or
nullptr for any other
+// table or tombstone type. The default is Iceberg's text of the value,
`00:00` for midnight, with at
+// most six fractional digits, so the strict comparison cast parses it into
the tombstones' type
+// exactly, and text that does not parse fails the scan; an ORC TIME
tombstone, BIGINT, compares in
+// TIMEV2(6) as a present key does. Compared as text, the default `00:00`
never matched a tombstone
+// printing 00:00:00.000000. A timestamptz_ns default is rejected by the
caller instead.
+static DataTypePtr equality_delete_placeholder_default_type(const DataTypePtr&
table_type,
+ const DataTypePtr&
delete_type) {
+ if (remove_nullable(table_type)->get_primitive_type() != TYPE_STRING) {
+ return nullptr;
+ }
+ switch (remove_nullable(delete_type)->get_primitive_type()) {
+ case TYPE_TIMEV2:
+ return remove_nullable(delete_type);
+ case TYPE_BIGINT:
+ return std::make_shared<DataTypeTimeV2>(TimeValue::MICROS_SCALE);
+ default:
+ return nullptr;
+ }
+}
+
+// Makes `key_expr` yield `comparison_type`, keeping its nullability.
+static VExprSPtr cast_equality_delete_key(VExprSPtr key_expr, const
DataTypePtr& comparison_type) {
+ DORIS_CHECK(key_expr != nullptr);
+ DORIS_CHECK(comparison_type != nullptr);
+ const auto& source_type = key_expr->data_type();
+ if
(remove_nullable(source_type)->equals(*remove_nullable(comparison_type))) {
+ return key_expr;
+ }
+ const auto result_type = source_type->is_nullable() ?
make_nullable(comparison_type)
+ :
remove_nullable(comparison_type);
+ // Only an ORC Iceberg TIME key compares a BIGINT as TIMEV2
(equality_delete_comparison_type),
+ // and its value is microseconds, which CAST(BIGINT AS TIME) would read as
HHMMSS digits.
+ if (remove_nullable(source_type)->get_primitive_type() == TYPE_BIGINT &&
+ remove_nullable(comparison_type)->get_primitive_type() == TYPE_TIMEV2)
{
+ auto time_expr =
+ std::make_shared<IcebergTimeMicrosExpr>(result_type,
key_expr->expr_name());
+ time_expr->add_child(std::move(key_expr));
+ return time_expr;
+ }
+ auto cast_expr = EvolutionCast::create_shared(result_type, false, true);
+ cast_expr->add_child(std::move(key_expr));
+ return cast_expr;
+}
+
Status IcebergTableReader::annotate_projected_column(const TFileScanSlotInfo&
slot_info,
format::ProjectedColumnBuildContext* context,
format::ColumnDefinition*
column) const {
@@ -1173,16 +1476,20 @@ std::string IcebergTableReader::debug_string() const {
equality_filters << ", ";
}
const auto& filter = _equality_delete_filters[idx];
+ const auto append_types = [&equality_filters](const
std::vector<DataTypePtr>& types) {
+ for (size_t type_idx = 0; type_idx < types.size(); ++type_idx) {
+ if (type_idx > 0) {
+ equality_filters << ", ";
+ }
+ equality_filters << (types[type_idx] == nullptr ? "null"
+ :
types[type_idx]->get_name());
+ }
+ };
equality_filters << "EqualityDeleteFilter{field_ids="
<< join_values_for_debug(filter.field_ids) << ",
key_types=[";
- for (size_t type_idx = 0; type_idx < filter.key_types.size();
++type_idx) {
- if (type_idx > 0) {
- equality_filters << ", ";
- }
- equality_filters << (filter.key_types[type_idx] == nullptr
- ? "null"
- :
filter.key_types[type_idx]->get_name());
- }
+ append_types(filter.key_types);
+ equality_filters << "], delete_key_types=[";
+ append_types(filter.delete_key_types);
equality_filters << "], delete_block_rows=" <<
filter.delete_block.rows()
<< ", delete_block_columns=" <<
filter.delete_block.columns() << "}";
}
@@ -1542,9 +1849,10 @@ void
IcebergTableReader::_append_equality_delete_row_count_carrier(
Status IcebergTableReader::_build_missing_equality_delete_key_expr(
const EqualityDeleteFilter& filter, size_t key_idx,
const EqualityDeleteColumnPath& data_path, format::FileScanRequest*
const request,
- VExprSPtr* const key_expr) {
+ VExprSPtr* const key_expr, DataTypePtr* const comparison_type) {
DORIS_CHECK(request != nullptr);
DORIS_CHECK(key_expr != nullptr);
+ DORIS_CHECK(comparison_type != nullptr);
auto table_path =
_find_table_column_path_by_field_id(filter.field_ids[key_idx],
filter.key_types[key_idx], true);
if (!table_path.has_value() || data_path.size() >= table_path->size()) {
@@ -1569,11 +1877,41 @@ Status
IcebergTableReader::_build_missing_equality_delete_key_expr(
const auto* table_leaf = missing_path.back();
DORIS_CHECK(table_leaf != nullptr);
DORIS_CHECK(table_leaf->type != nullptr);
- if (!table_leaf->type->equals(*filter.key_types[key_idx])) {
- auto cast_expr =
EvolutionCast::create_shared(filter.key_types[key_idx]);
- cast_expr->add_child(default_expr);
- default_expr = std::move(cast_expr);
+ bool default_is_null = false;
+ RETURN_IF_ERROR(equality_delete_default_leaf_is_null(missing_path,
&default_is_null));
+ if (default_is_null) {
+ // A NULL default makes the key NULL in this file whatever its table
type, so it is
+ // compared in the delete file's type, to which the table type need
not convert: a dropped
+ // key whose retained schemas only typed it Iceberg `unknown` arrives
as STRING, while its
+ // tombstones are, for example, DATE.
+ *comparison_type = remove_nullable(filter.delete_key_types[key_idx]);
+ } else if (remove_nullable(table_leaf->type)->get_primitive_type() ==
TYPE_STRING &&
+
remove_nullable(filter.delete_key_types[key_idx])->get_primitive_type() ==
+ TYPE_TIMESTAMPTZ) {
+ // An Iceberg timestamptz_ns key behind the STRING placeholder. Its
tombstones are read as
+ // TIMESTAMPTZ(6), Parquet flooring and ORC rounding the nanoseconds,
so no parse of the
+ // nanosecond text default compares exactly: the default `1970-01-01
00:00:00.0000001+00:00`
+ // would match an epoch tombstone and delete every row of this file,
and the default epoch
+ // a tombstone 100 ns after it.
+ return Status::NotSupported(
+ "Iceberg equality delete field id {} is a timestamptz_ns key
that data file {} "
+ "does not store, with a non-NULL initial default; the delete
file's values are "
+ "read as {} without their nanoseconds, so the default cannot
be compared exactly",
+ filter.field_ids[key_idx], _data_file_path(),
+ remove_nullable(filter.delete_key_types[key_idx])->get_name());
+ } else if (auto placeholder_type =
equality_delete_placeholder_default_type(
+ table_leaf->type, filter.delete_key_types[key_idx]);
+ placeholder_type != nullptr) {
+ // A TIME `payload.k` added with the default `00:00` after the file
was written: rows
+ // under a NULL `payload` still get a NULL key from
AncestorNullDefaultExpr below, the
+ // others the parsed midnight, so a 00:00:00 tombstone deletes exactly
those.
+ *comparison_type = std::move(placeholder_type);
+ } else {
+ RETURN_IF_ERROR(equality_delete_comparison_type(
+ filter.field_ids[key_idx], filter.delete_key_types[key_idx],
table_leaf->type,
+ table_leaf->type, _data_file_path(), comparison_type));
}
+ default_expr = cast_equality_delete_key(std::move(default_expr),
*comparison_type);
if (data_path.empty()) {
*key_expr = std::move(default_expr);
return Status::OK();
@@ -1586,8 +1924,7 @@ Status
IcebergTableReader::_build_missing_equality_delete_key_expr(
VExprSPtr ancestor_expr;
RETURN_IF_ERROR(build_equality_delete_key_expr(data_path, block_position,
&ancestor_expr));
auto combined_expr = std::make_shared<AncestorNullDefaultExpr>(
- make_nullable(filter.key_types[key_idx]),
- ancestor_expr->expr_name() + "." + table_leaf->name);
+ make_nullable(*comparison_type), ancestor_expr->expr_name() + "."
+ table_leaf->name);
combined_expr->add_child(ancestor_expr);
combined_expr->add_child(default_expr);
*key_expr = std::move(combined_expr);
@@ -1597,37 +1934,62 @@ Status
IcebergTableReader::_build_missing_equality_delete_key_expr(
Status
IcebergTableReader::_append_equality_delete_predicates(format::FileScanRequest*
request) {
DORIS_CHECK(request != nullptr);
for (const auto& filter : _equality_delete_filters) {
- auto delete_predicate = std::make_shared<EqualityDeletePredicate>(
- filter.delete_block, filter.field_ids, filter.hash_index);
DCHECK_EQ(filter.field_ids.size(), filter.key_types.size());
+ DCHECK_EQ(filter.field_ids.size(), filter.delete_key_types.size());
+ // A delete file without rows deletes nothing, but
EqualityDeletePredicate evaluates its
+ // keys before it looks at the delete set, so the strict comparison
casts would still run:
+ // a DATE key 1000-01-01 compared with an empty TIMESTAMP_NS delete
file fails the scan
+ // with an overflow. It is skipped here rather than at load time, so
the loaded filter
+ // still disables aggregate pushdown as before.
+ if (filter.delete_block.rows() == 0) {
+ continue;
+ }
+ VExprSPtrs key_exprs;
+ DataTypes comparison_types;
bool has_missing_key = false;
for (size_t idx = 0; idx < filter.field_ids.size(); ++idx) {
EqualityDeleteColumnPath data_path;
bool complete_path = false;
RETURN_IF_ERROR(
_find_equality_delete_data_field(filter, idx, &data_path,
&complete_path));
+ VExprSPtr key_expr;
+ DataTypePtr comparison_type;
if (!complete_path) {
- VExprSPtr key_expr;
-
RETURN_IF_ERROR(_build_missing_equality_delete_key_expr(filter, idx, data_path,
-
request, &key_expr));
- delete_predicate->add_child(key_expr);
+ RETURN_IF_ERROR(_build_missing_equality_delete_key_expr(
+ filter, idx, data_path, request, &key_expr,
&comparison_type));
has_missing_key = true;
- continue;
- }
- const auto* root = data_path.front();
- const auto* field = data_path.back();
- const auto field_column_id =
format::LocalColumnId(root->file_local_id());
- _append_file_scan_column(request, field_column_id,
&request->predicate_columns);
- const auto block_position =
request->local_positions.at(field_column_id).value();
- VExprSPtr key_expr;
- RETURN_IF_ERROR(build_equality_delete_key_expr(data_path,
block_position, &key_expr));
- if (field->type->equals(*filter.key_types[idx])) {
- delete_predicate->add_child(key_expr);
} else {
- auto cast_expr =
EvolutionCast::create_shared(filter.key_types[idx]);
- cast_expr->add_child(key_expr);
- delete_predicate->add_child(cast_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);
+ 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));
+ RETURN_IF_ERROR(equality_delete_comparison_type(
+ filter.field_ids[idx], filter.delete_key_types[idx],
key_expr->data_type(),
+ filter.key_types[idx], _data_file_path(),
&comparison_type));
+ key_expr = cast_equality_delete_key(std::move(key_expr),
comparison_type);
}
+ key_exprs.push_back(std::move(key_expr));
+ comparison_types.push_back(std::move(comparison_type));
+ }
+ // A data file can hold a key in a wider type than the delete file,
e.g. a LONG written
+ // after an INT -> LONG promotion against INT tombstones written
before it. The tombstones
+ // are then widened once per scanner and cached, instead of narrowing
every data value.
+ EqualityDeleteFilter converted;
+ const bool needs_conversion = !std::ranges::equal(
+ filter.delete_key_types, comparison_types,
+ [](const DataTypePtr& delete_type, const DataTypePtr&
comparison_type) {
+ return
remove_nullable(delete_type)->equals(*comparison_type);
+ });
+ if (needs_conversion) {
+ RETURN_IF_ERROR(_convert_equality_delete_filter(filter,
comparison_types, &converted));
+ }
+ const auto& compared = needs_conversion ? converted : filter;
+ auto delete_predicate = std::make_shared<EqualityDeletePredicate>(
+ compared.delete_block, filter.field_ids, compared.hash_index);
+ for (auto& key_expr : key_exprs) {
+ delete_predicate->add_child(std::move(key_expr));
}
if (has_missing_key && request->predicate_columns.empty()) {
_append_equality_delete_row_count_carrier(request);
@@ -1638,6 +2000,69 @@ Status
IcebergTableReader::_append_equality_delete_predicates(format::FileScanRe
return Status::OK();
}
+Status IcebergTableReader::_convert_equality_delete_filter(const
EqualityDeleteFilter& filter,
+ const DataTypes&
comparison_types,
+
EqualityDeleteFilter* result) {
+ DORIS_CHECK(_split_cache != nullptr);
+ DORIS_CHECK(result != nullptr);
+ DORIS_CHECK(comparison_types.size() == filter.delete_key_types.size());
+ std::ostringstream cache_key;
+ cache_key << filter.cache_key << ":compare";
+ for (const auto& type : comparison_types) {
+ cache_key << ':' << type->get_name();
+ }
+ Status convert_status = Status::OK();
+ bool cache_hit = false;
+ auto* cached_filter = _split_cache->get<EqualityDeleteFilter>(
+ cache_key.str(),
+ [&]() -> EqualityDeleteFilter* {
+ auto converted = std::make_unique<EqualityDeleteFilter>();
+ convert_status =
_build_converted_equality_delete_filter(filter, comparison_types,
+
converted.get());
+ if (!convert_status.ok()) {
+ return nullptr;
+ }
+ converted->cache_key = cache_key.str();
+ return converted.release();
+ },
+ &cache_hit);
+ RETURN_IF_ERROR(convert_status);
+ DORIS_CHECK(cached_filter != nullptr);
+ if (!cache_hit) {
+ COUNTER_UPDATE(_profile.equality_delete_hash_index_memory,
+
static_cast<int64_t>(cached_filter->hash_index->memory_usage()));
+ }
+ *result = *cached_filter;
+ return Status::OK();
+}
+
+Status IcebergTableReader::_build_converted_equality_delete_filter(
+ const EqualityDeleteFilter& filter, const DataTypes& comparison_types,
+ EqualityDeleteFilter* result) const {
+ DORIS_CHECK(result != nullptr);
+ result->field_ids = filter.field_ids;
+ result->field_names = filter.field_names;
+ result->key_types = filter.key_types;
+ Block delete_block = filter.delete_block;
+ RowDescriptor row_desc;
+ for (size_t index = 0; index < comparison_types.size(); ++index) {
+ const auto& entry = delete_block.get_by_position(index);
+ VExprSPtr key_expr = VSlotRef::create_shared(cast_set<int>(index),
cast_set<int>(index), -1,
+ entry.type, entry.name);
+ auto context = VExprContext::create_shared(
+ cast_equality_delete_key(std::move(key_expr),
comparison_types[index]));
+ RETURN_IF_ERROR(context->prepare(_runtime_state, row_desc));
+ RETURN_IF_ERROR(context->open(_runtime_state));
+ ColumnWithTypeAndName key;
+ RETURN_IF_ERROR(context->execute(&delete_block, key));
+ key.name = entry.name;
+ result->delete_key_types.push_back(key.type);
+ result->delete_block.insert(std::move(key));
+ }
+ result->hash_index =
EqualityDeletePredicate::build_hash_index(result->delete_block);
+ return Status::OK();
+}
+
Status IcebergTableReader::_create_delete_file_reader(const
TIcebergDeleteFileDesc& delete_file,
const
TFileScanRangeParams& scan_params,
IcebergDeleteFileIOContext* delete_io_ctx,
@@ -1828,10 +2253,10 @@ Status
IcebergTableReader::_resolve_equality_delete_fields(
return Status::NotSupported(
"Iceberg equality delete does not support complex column
{}", field->name);
}
- // Equality comparison must run in the promoted (current snapshot
schema) type domain.
- // Narrowing a wider data key into a historical delete-file type is
lossy: an INT overflow
- // becomes NULL, and NULL-safe equality then matches a NULL delete
key. Delete values are
- // promoted into this same domain when the delete file is read.
+ // The delete block keeps the delete file's own key type. Each data
file's predicate
+ // compares a key in the wider of this type and its own
(equality_delete_comparison_type):
+ // narrowing to either side is lossy, as an INT overflow becomes NULL
and NULL-safe
+ // equality then matches a NULL key.
const DataTypePtr delete_file_type =
path.size() > 1 ? make_nullable(field->type) : field->type;
DataTypePtr key_type = delete_file_type;
@@ -1843,6 +2268,7 @@ Status
IcebergTableReader::_resolve_equality_delete_fields(
result->field_ids.push_back(field_id);
result->field_names.push_back(field->name);
result->key_types.push_back(key_type);
+ result->delete_key_types.push_back(delete_file_type);
}
return Status::OK();
}
@@ -1890,14 +2316,7 @@ Status
IcebergTableReader::_load_equality_delete_file(const TIcebergDeleteFileDe
VExprSPtr key_expr;
RETURN_IF_ERROR(build_equality_delete_key_expr(
path, request->local_positions.at(root_column_id).value(),
&key_expr));
- const auto& key_type = result->key_types[index];
- if (!key_expr->data_type()->equals(*key_type)) {
- // Historical delete values share the promoted comparison domain.
DATE-to-nanos
- // must report overflow rather than turn a non-NULL equality key
into NULL.
- auto cast_expr = EvolutionCast::create_shared(key_type);
- cast_expr->add_child(key_expr);
- key_expr = std::move(cast_expr);
- }
+
DORIS_CHECK(key_expr->data_type()->equals(*result->delete_key_types[index]));
auto context = VExprContext::create_shared(std::move(key_expr));
RETURN_IF_ERROR(context->prepare(_runtime_state, row_desc));
RETURN_IF_ERROR(context->open(_runtime_state));
@@ -1908,7 +2327,7 @@ Status
IcebergTableReader::_load_equality_delete_file(const TIcebergDeleteFileDe
Block delete_block_template;
for (size_t index = 0; index < delete_paths.size(); ++index) {
const auto* field = delete_paths[index].back();
- const auto& key_type = result->key_types[index];
+ const auto& key_type = result->delete_key_types[index];
delete_block_template.insert({key_type->create_column(), key_type,
field->name});
}
MutableBlock mutable_delete_block(delete_block_template.clone_empty());
@@ -1942,8 +2361,8 @@ Status
IcebergTableReader::_read_equality_delete_file(const TIcebergDeleteFileDe
std::ostringstream cache_key;
cache_key << _delete_file_cache_key("iceberg_v2_equality_delete_",
delete_file.path);
if (scan_params.__isset.current_schema_id) {
- // The promoted comparison type depends on the current snapshot
schema, so a cached filter
- // must not be reused across schemas.
+ // The table-side key types depend on the current snapshot schema, so
a cached filter must
+ // not be reused across schemas.
cache_key << ":schema=" << scan_params.current_schema_id;
}
cache_key << ':' << delete_file.field_ids.size();
@@ -1964,6 +2383,7 @@ Status
IcebergTableReader::_read_equality_delete_file(const TIcebergDeleteFileDe
if (!read_status.ok()) {
return nullptr;
}
+ result->cache_key = cache_key.str();
return result.release();
},
&cache_hit);
diff --git a/be/src/format_v2/table/iceberg_reader.h
b/be/src/format_v2/table/iceberg_reader.h
index 1a4292d0994..47f71353137 100644
--- a/be/src/format_v2/table/iceberg_reader.h
+++ b/be/src/format_v2/table/iceberg_reader.h
@@ -180,7 +180,15 @@ private:
size_t key_idx,
const
EqualityDeleteColumnPath& data_path,
format::FileScanRequest*
request,
- VExprSPtr* key_expr);
+ VExprSPtr* key_expr,
+ DataTypePtr*
comparison_type);
+ // Returns `filter` with its delete keys converted to `comparison_types`,
cached per scanner.
+ Status _convert_equality_delete_filter(const EqualityDeleteFilter& filter,
+ const DataTypes& comparison_types,
+ EqualityDeleteFilter* result);
+ Status _build_converted_equality_delete_filter(const EqualityDeleteFilter&
filter,
+ const DataTypes&
comparison_types,
+ EqualityDeleteFilter*
result) const;
Status _find_equality_delete_data_field(const EqualityDeleteFilter&
filter, size_t key_idx,
EqualityDeleteColumnPath*
data_path,
bool* complete_path) const;
@@ -238,9 +246,19 @@ private:
// Delete-file names are retained for Iceberg tables imported from
formats that did not
// persist field ids. In BY_NAME mode they are the fallback binding
key.
std::vector<std::string> field_names;
+ // Table-side types of the keys, from the scan's schema metadata. They
resolve table
+ // fields and build defaults for keys a data file lacks. The
comparison type comes from
+ // the file types; a table type only lets a timestamptz key read as
DATETIMEV2 compare
+ // as TIMESTAMPTZ, and a STRING placeholder (an Iceberg TIME or
timestamptz_ns, which
+ // Doris cannot represent) lets an ORC TIME key and its text default
compare as TIMEV2.
std::vector<DataTypePtr> key_types;
+ // Types of the delete block columns: the delete file's own key types,
or the comparison
+ // types of a converted filter.
+ std::vector<DataTypePtr> delete_key_types;
Block delete_block;
std::shared_ptr<const EqualityDeleteHashIndex> hash_index;
+ // Split-cache key of this filter, extended by the comparison types
for converted copies.
+ std::string cache_key;
};
std::vector<EqualityDeleteFilter> _equality_delete_filters;
// Scanner-shared cache supplied in SplitReadOptions. Parsed delete files
outlive one data-file
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index 2651501c62e..8b41271b196 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -90,6 +90,7 @@
#include "util/debug_points.h"
#include "util/defer_op.h"
#include "util/hash_util.hpp"
+#include "util/timezone_utils.h"
#include "util/url_coding.h"
namespace doris::format {
@@ -1022,9 +1023,11 @@ void write_nested_key_value_orc_file(const std::string&
file_path, const std::ve
output.write(memory_stream.getData(),
static_cast<std::streamsize>(memory_stream.getLength()));
}
+// 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.
void write_timestamp_int_parquet_file(const std::string& file_path,
const std::vector<int64_t>& timestamps,
- const std::vector<int32_t>& ids) {
+ const std::vector<int32_t>& ids, bool
int96 = false) {
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"});
@@ -1045,9 +1048,13 @@ void write_timestamp_int_parquet_file(const std::string&
file_path,
builder.version(::parquet::ParquetVersion::PARQUET_2_6);
builder.data_page_version(::parquet::ParquetDataPageVersion::V2);
builder.compression(::parquet::Compression::UNCOMPRESSED);
+ ::parquet::ArrowWriterProperties::Builder arrow_builder;
+ if (int96) {
+ arrow_builder.enable_force_write_int96_timestamps();
+ }
PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), out,
static_cast<int64_t>(timestamps.size()),
- builder.build()));
+ builder.build(),
arrow_builder.build()));
}
void write_iceberg_equality_delete_bigint_parquet_file(const std::string&
file_path,
@@ -4964,6 +4971,741 @@ TEST(IcebergV2ReaderTest,
IcebergEqualityDeletePromotesHistoricalDeleteKeyToCurr
std::filesystem::remove_all(test_dir);
}
+template <typename Builder, typename Value>
+std::shared_ptr<arrow::Array> build_nullable_array(
+ Builder* builder, const std::vector<std::optional<Value>>& values) {
+ for (const auto& value : values) {
+ if (value.has_value()) {
+ EXPECT_TRUE(builder->Append(*value).ok());
+ } else {
+ EXPECT_TRUE(builder->AppendNull().ok());
+ }
+ }
+ return finish_array(builder);
+}
+
+std::shared_ptr<arrow::Array> build_nullable_int32_keys(
+ const std::vector<std::optional<int32_t>>& values) {
+ arrow::Int32Builder builder;
+ return build_nullable_array(&builder, values);
+}
+
+std::shared_ptr<arrow::Array> build_date32_keys(const
std::vector<std::optional<int32_t>>& days) {
+ arrow::Date32Builder builder;
+ return build_nullable_array(&builder, days);
+}
+
+std::shared_ptr<arrow::Array> build_timestamp_keys(
+ arrow::TimeUnit::type unit, const std::vector<std::optional<int64_t>>&
ticks) {
+ arrow::TimestampBuilder builder(arrow::timestamp(unit),
arrow::default_memory_pool());
+ return build_nullable_array(&builder, ticks);
+}
+
+void write_parquet_table(const std::string& file_path, const
std::shared_ptr<arrow::Table>& table) {
+ 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::WriterProperties::Builder builder;
+ builder.version(::parquet::ParquetVersion::PARQUET_2_6);
+ builder.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ builder.compression(::parquet::Compression::UNCOMPRESSED);
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), out,
+
std::max<int64_t>(table->num_rows(), 1),
+ builder.build()));
+}
+
+// Writes `id` (field id 0) and, unless `keys` is null, the nullable key `k`
(field id 1).
+void write_id_and_key_parquet_file(const std::string& file_path, const
std::vector<int32_t>& ids,
+ const std::shared_ptr<arrow::Array>& keys) {
+ std::vector<std::shared_ptr<arrow::Field>> fields {
+ arrow::field("id", arrow::int32(), false)
+
->WithMetadata(arrow::key_value_metadata({"PARQUET:field_id"}, {"0"}))};
+ std::vector<std::shared_ptr<arrow::Array>> columns
{build_int32_array(ids)};
+ if (keys != nullptr) {
+ fields.push_back(
+ arrow::field("k", keys->type(), true)
+
->WithMetadata(arrow::key_value_metadata({"PARQUET:field_id"}, {"1"})));
+ columns.push_back(keys);
+ }
+ write_parquet_table(file_path, arrow::Table::Make(arrow::schema(fields),
columns));
+}
+
+struct KeyDataFile {
+ std::vector<int32_t> ids;
+ // nullptr: the data file has no `k`, as one written before the key was
added.
+ std::shared_ptr<arrow::Array> keys;
+};
+
+// Writes an equality delete file at the given path and returns its format.
+using EqualityDeleteFileWriter = std::function<TFileFormatType::type(const
std::string&)>;
+
+// Reads each data file as one split of a single reader. Every split carries
the same equality
+// delete file on `equality_field_id`, whose one column has the name and field
id of `key_field`
+// and holds `delete_keys`; for a nested key that is a struct array whose
child carries the key's
+// field id. `write_delete_file`, when given, writes that file instead, for
physical shapes Arrow
+// does not produce. `key_field` describes that column in the scan's schema
metadata, as the FE
+// sends it. Returns the surviving ids of each split.
+Status read_ids_after_key_equality_delete(const std::string& test_name,
+ const std::vector<KeyDataFile>&
data_files,
+ const std::shared_ptr<arrow::Array>&
delete_keys,
+ schema::external::TFieldPtr
key_field,
+ std::vector<std::vector<int32_t>>*
ids_per_split,
+ int32_t equality_field_id = 1,
+ bool enable_mapping_timestamp_tz =
false,
+ const EqualityDeleteFileWriter&
write_delete_file = {}) {
+ const auto test_dir = std::filesystem::temp_directory_path() / test_name;
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto delete_file_path = (test_dir /
"equality-delete.parquet").string();
+ auto delete_file_format = TFileFormatType::FORMAT_PARQUET;
+ if (write_delete_file) {
+ delete_file_format = write_delete_file(delete_file_path);
+ } else {
+ const auto& delete_column = *key_field.field_ptr;
+ write_parquet_table(
+ delete_file_path,
+ arrow::Table::Make(
+ arrow::schema({arrow::field(delete_column.name,
delete_keys->type(), true)
+
->WithMetadata(arrow::key_value_metadata(
+ {"PARQUET:field_id"},
+
{std::to_string(delete_column.id)}))}),
+ {delete_keys}));
+ }
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ auto scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ if (enable_mapping_timestamp_tz) {
+ 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(key_field)})});
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ 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);
+ for (size_t index = 0; index < data_files.size(); ++index) {
+ const auto file_path = (test_dir / ("data-" + std::to_string(index) +
".parquet")).string();
+ write_id_and_key_parquet_file(file_path, data_files[index].ids,
data_files[index].keys);
+ 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, {equality_field_id},
+
delete_file_format)}));
+ RETURN_IF_ERROR(reader.prepare_split(split_options));
+ std::vector<int32_t> ids;
+ bool eos = false;
+ while (!eos) {
+ Block block = build_table_block(projected_columns);
+ RETURN_IF_ERROR(reader.get_block(&block, &eos));
+ 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));
+ }
+ }
+ ids_per_split->push_back(std::move(ids));
+ }
+ RETURN_IF_ERROR(reader.close());
+ std::filesystem::remove_all(test_dir);
+ return Status::OK();
+}
+
+constexpr int64_t MICROS_PER_DAY = 86400LL * 1000 * 1000;
+constexpr int32_t DAYS_1000_01_01 = -354285;
+constexpr int32_t DAYS_1600_01_01 = -135140;
+constexpr int32_t DAYS_2000_01_01 = 10957;
+constexpr int32_t DAYS_2024_01_02 = 19724;
+constexpr int32_t DAYS_2024_01_03 = 19725;
+
+// After expireSnapshots().cleanExpiredMetadata(true) removes the schema of a
schema-only
+// INT -> LONG promotion, the FE types a dropped key INT although its
tombstones are LONG.
+// Comparing in INT turned 4294967296 into NULL, which deleted the NULL key of
the old file and
+// the 4294967296 key of the new one together.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteKeepsWideTombstonesUnderNarrowTableType) {
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_narrow_table_type",
+ {{.ids = {1, 2, 3}, .keys =
build_nullable_int32_keys({std::nullopt, 0, 1})},
+ {.ids = {4, 5}, .keys = build_nullable_int64_array({std::nullopt,
4294967296})}},
+ build_nullable_int64_array({4294967296}),
+ external_schema_field("k", 1, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::INT), false, true),
+ &ids);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(ids, (std::vector<std::vector<int32_t>> {{1, 2, 3}, {4}}));
+}
+
+// INT tombstones written before an INT -> LONG promotion meet LONG data files
and INT data files
+// of the same scan. LONG files compare in LONG through a widened copy of the
delete block, which
+// the third split takes from the scanner cache; the INT file uses the delete
block as read.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteWidensTombstonesPerDataFileType) {
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_widened_per_data_file",
+ {{.ids = {1, 2}, .keys = build_nullable_int64_array({7,
4294967303})},
+ {.ids = {3, 4}, .keys = build_nullable_int32_keys({7, 8})},
+ {.ids = {5, 6}, .keys = build_nullable_int64_array({4294967303,
7})}},
+ build_nullable_int32_keys({7}),
+ external_schema_field("k", 1, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::BIGINT), false, true),
+ &ids);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(ids, (std::vector<std::vector<int32_t>> {{2}, {4}, {5}}));
+}
+
+// A read pinned before a later DATE -> TIMESTAMP_NS promotion gets the
promoted table type. The
+// DATE tombstone and the DATE data compare as DATE, so 1000-01-01 does not
have to fit
+// TIMESTAMP_NS (1677-09-21..2262-04-11) and the scan no longer fails with an
overflow.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteComparesDatesUnderNanosecondTableType) {
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_wide_table_type",
+ {{.ids = {1, 2}, .keys = build_date32_keys({DAYS_1000_01_01,
DAYS_2000_01_01})}},
+ build_date32_keys({DAYS_1000_01_01}),
+ external_schema_field("k", 1, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::TIMESTAMP_NS), false,
+ true),
+ &ids);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(ids, (std::vector<std::vector<int32_t>> {{2}}));
+}
+
+// The FE encodes an Iceberg `unknown` key as STRING when no retained schema
gives it a concrete
+// type. Comparing as STRING made the DATE key "2024-01-02" differ from the
TIMESTAMP_NS tombstone
+// "2024-01-02 00:00:00.000000000", so the deleted row stayed visible.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteIgnoresUnknownTableTypeForDateAndNanosKeys) {
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_unknown_table_type",
+ {{.ids = {1, 2}, .keys = build_date32_keys({DAYS_2024_01_02,
DAYS_2024_01_03})}},
+ build_timestamp_keys(arrow::TimeUnit::NANO, {DAYS_2024_01_02 *
MICROS_PER_DAY * 1000}),
+ external_schema_field("k", 1, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::STRING), false, true),
+ &ids);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(ids, (std::vector<std::vector<int32_t>> {{2}}));
+}
+
+// A delete prepared under a TIMESTAMP_NS schema that the table later
abandoned holds nanosecond
+// tombstones while the table type is TIMESTAMP. Rounding the tombstone
12:00:00.000000500 to
+// microseconds matched one of the microsecond keys; in nanoseconds it matches
neither.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteComparesNanosecondTombstonesInNanoseconds) {
+ const int64_t noon_micros = DAYS_2024_01_02 * MICROS_PER_DAY + 12LL * 3600
* 1000 * 1000;
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_abandoned_nanos",
+ {{.ids = {1, 2},
+ .keys = build_timestamp_keys(arrow::TimeUnit::MICRO,
+ {noon_micros, noon_micros + 1})}},
+ build_timestamp_keys(arrow::TimeUnit::NANO, {noon_micros * 1000 +
500}),
+ external_schema_field("k", 1, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::DATETIMEV2, -1, 6), false,
+ true),
+ &ids);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(ids, (std::vector<std::vector<int32_t>> {{1, 2}}));
+}
+
+// Comparing microsecond keys in nanoseconds must fail for a key outside
1677-09-21..2262-04-11.
+// A non-strict cast would turn 1600-01-01 into NULL, and the NULL tombstone
would delete it.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteRejectsMicrosOutsideNanosecondRange) {
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_micros_outside_nanos",
+ {{.ids = {1, 2},
+ .keys = build_timestamp_keys(
+ arrow::TimeUnit::MICRO,
+ {DAYS_1600_01_01 * MICROS_PER_DAY, DAYS_2024_01_02 *
MICROS_PER_DAY})}},
+ build_timestamp_keys(arrow::TimeUnit::NANO, {std::nullopt}),
+ external_schema_field("k", 1, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::DATETIMEV2, -1, 6), false,
+ true),
+ &ids);
+ ASSERT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("TIMESTAMP_NS overflow"),
std::string::npos) << status;
+}
+
+// An equality delete file without rows deletes nothing, so the strict
comparison casts that a
+// nonempty one needs must not fail the scan: DATE keys meeting TIMESTAMP_NS
tombstones are cast to
+// TIMESTAMP_NS, which 1000-01-01 overflows, both as a key of the first file
and as the initial
+// default for the second. Building a predicate for the empty file fails with
+// "TIMESTAMP_NS overflow".
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteSkipsEmptyDeleteFileBeforeKeyCasts) {
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_empty_delete_file",
+ {{.ids = {1, 2}, .keys = build_date32_keys({DAYS_1000_01_01,
DAYS_2000_01_01})},
+ {.ids = {3}, .keys = nullptr}},
+ build_timestamp_keys(arrow::TimeUnit::NANO, {}),
+ external_schema_field("k", 1, {}, "1000-01-01",
+
external_primitive_type(TPrimitiveType::DATEV2), false, true),
+ &ids);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(ids, (std::vector<std::vector<int32_t>> {{1, 2}, {3}}));
+}
+
+// A data file without the key compares its initial default, typed by the
table, against the
+// tombstones in the wider of the two types: INT default 5 matches the LONG
tombstone 5.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteWidensMissingKeyDefaultToTombstoneType) {
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_missing_key_default_widened",
+ {{.ids = {1, 2, 3}, .keys = nullptr}},
build_nullable_int64_array({5}),
+ external_schema_field("k", 1, {}, "5",
external_primitive_type(TPrimitiveType::INT),
+ false, true),
+ &ids);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(ids, (std::vector<std::vector<int32_t>> {std::vector<int32_t>
{}}));
+}
+
+// Without an initial default a missing key is NULL whatever the table type,
so it is compared in
+// the tombstones' type even when the table type cannot convert to it
(`unknown` arrives as
+// STRING): the NULL tombstone deletes every row, the DATE one deletes none.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteMatchesNullMissingKeyUnderUnknownTableType) {
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_missing_key_unknown_type",
+ {{.ids = {1, 2}, .keys = nullptr}},
build_date32_keys({std::nullopt, DAYS_2024_01_02}),
+ external_schema_field("k", 1, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::STRING), false, true),
+ &ids);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(ids, (std::vector<std::vector<int32_t>> {std::vector<int32_t>
{}}));
+}
+
+// The nested delete key `payload.k` (field id 3 under `payload`) holding
`days` as DATE under
+// non-NULL parents.
+std::shared_ptr<arrow::Array> build_struct_date32_keys(
+ const std::vector<std::optional<int32_t>>& days) {
+ auto key_field = arrow::field("k", arrow::date32(), true)
+
->WithMetadata(arrow::key_value_metadata({"PARQUET:field_id"}, {"3"}));
+ auto keys = arrow::StructArray::Make({build_date32_keys(days)},
{key_field});
+ DORIS_CHECK(keys.ok());
+ return *keys;
+}
+
+// A missing struct whose initial default is `{}` takes each child's own
default, so `payload.k`
+// without one is NULL, and it compares in the tombstones' type like a missing
key without a
+// default. Here `k` is STRING, as the FE encodes Iceberg `unknown`, and DATE
does not convert to
+// it. The NULL tombstone deletes every row and the DATE one none. Deciding
NULL by the struct's
+// default alone compares STRING with DATE and fails the scan with
NotSupported.
+TEST(IcebergV2ReaderTest,
IcebergMissingWholeStructEqualityKeyComparesNullLeafInTombstoneType) {
+ const std::vector<std::pair<std::optional<int32_t>, std::vector<int32_t>>>
cases {
+ {std::nullopt, {}}, {DAYS_2024_01_02, {1, 2, 3}}};
+ for (const auto& [tombstone, expected_ids] : cases) {
+ SCOPED_TRACE(tombstone.has_value() ? "DATE tombstone" : "NULL
tombstone");
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_null_struct_default_leaf",
+ {{.ids = {1, 2, 3}, .keys = nullptr}},
build_struct_date32_keys({tombstone}),
+ external_struct_schema_field(
+ "payload", 1,
+ {external_schema_field("k", 3, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::STRING),
+ false, true)},
+ true, "{}"),
+ &ids, 3);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(ids, (std::vector<std::vector<int32_t>> {expected_ids}));
+ }
+}
+
+constexpr int64_t TIME_01_00_00_000001_MICROS = 3600LL * 1000 * 1000 + 1;
+constexpr int64_t TIME_02_00_00_MICROS = 7200LL * 1000 * 1000;
+
+// Iceberg's Parquet writer stores TIME as TIME(isAdjustedToUTC=false,
MICROS), read as TIMEV2(6),
+// and timestamptz_ns as TIMESTAMP(isAdjustedToUTC=true, NANOS), both INT64.
+std::shared_ptr<const ::parquet::LogicalType> iceberg_parquet_time_type() {
+ return ::parquet::LogicalType::Time(false,
::parquet::LogicalType::TimeUnit::MICROS);
+}
+
+// Writes the Iceberg TIME `k` (field id 1) as Iceberg stores it, microseconds
from midnight in
+// both formats: Parquet's TIME(MICROS) and ORC's LONG annotated
iceberg.long-type=TIME, which reads
+// as BIGINT. Nonempty `ids` add the INT `id` (field id 0) in front.
+void write_iceberg_time_file(const std::string& file_path, FileFormat format,
+ const std::vector<int32_t>& ids,
+ const std::vector<std::optional<int64_t>>&
micros) {
+ ASSERT_TRUE(ids.empty() || ids.size() == micros.size());
+ if (format == FileFormat::PARQUET) {
+ ::parquet::schema::NodeVector fields;
+ if (!ids.empty()) {
+ fields.push_back(::parquet::schema::PrimitiveNode::Make(
+ "id", ::parquet::Repetition::REQUIRED,
::parquet::LogicalType::None(),
+ ::parquet::Type::INT32, -1, 0));
+ }
+ fields.push_back(::parquet::schema::PrimitiveNode::Make(
+ "k", ::parquet::Repetition::OPTIONAL,
iceberg_parquet_time_type(),
+ ::parquet::Type::INT64, -1, 1));
+ auto schema = std::static_pointer_cast<::parquet::schema::GroupNode>(
+ ::parquet::schema::GroupNode::Make("schema",
::parquet::Repetition::REQUIRED,
+ fields));
+ auto out = arrow::io::FileOutputStream::Open(file_path).ValueOrDie();
+ auto writer = ::parquet::ParquetFileWriter::Open(out, schema);
+ auto* group = writer->AppendRowGroup();
+ if (!ids.empty()) {
+ auto* id_writer =
static_cast<::parquet::Int32Writer*>(group->NextColumn());
+ for (const int32_t id : ids) {
+ id_writer->WriteBatch(1, nullptr, nullptr, &id);
+ }
+ id_writer->Close();
+ }
+ auto* key_writer =
static_cast<::parquet::Int64Writer*>(group->NextColumn());
+ for (const auto& value : micros) {
+ const int16_t level = value.has_value() ? 1 : 0;
+ const int64_t micros_value = value.value_or(0);
+ key_writer->WriteBatch(1, &level, nullptr, µs_value);
+ }
+ key_writer->Close();
+ group->Close();
+ writer->Close();
+ ASSERT_TRUE(out->Close().ok());
+ return;
+ }
+ auto type = std::unique_ptr<::orc::Type>(::orc::Type::buildTypeFromString(
+ ids.empty() ? "struct<k:bigint>" : "struct<id:int,k:bigint>"));
+ const size_t key_index = ids.empty() ? 0 : 1;
+ if (!ids.empty()) {
+ type->getSubtype(0)->setAttribute("iceberg.id", "0");
+ }
+ type->getSubtype(key_index)->setAttribute("iceberg.id", "1");
+ type->getSubtype(key_index)->setAttribute("iceberg.long-type", "TIME");
+ MemoryOutputStream stream(1024 * 1024);
+ ::orc::WriterOptions options;
+ auto writer = ::orc::createWriter(*type, &stream, options);
+ auto batch = writer->createRowBatch(micros.size());
+ auto& root = dynamic_cast<::orc::StructVectorBatch&>(*batch);
+ root.numElements = micros.size();
+ if (!ids.empty()) {
+ auto& id_batch =
dynamic_cast<::orc::LongVectorBatch&>(*root.fields[0]);
+ id_batch.numElements = ids.size();
+ for (size_t row = 0; row < ids.size(); ++row) {
+ id_batch.data[row] = ids[row];
+ }
+ }
+ auto& key_batch =
dynamic_cast<::orc::LongVectorBatch&>(*root.fields[key_index]);
+ key_batch.numElements = micros.size();
+ key_batch.hasNulls = true;
+ for (size_t row = 0; row < micros.size(); ++row) {
+ key_batch.notNull[row] = micros[row].has_value();
+ key_batch.data[row] = micros[row].value_or(0);
+ }
+ writer->add(*batch);
+ writer->close();
+ std::ofstream out(file_path, std::ios::binary);
+ out.write(stream.getData(),
static_cast<std::streamsize>(stream.getLength()));
+}
+
+// The FE sends an Iceberg TIME, which Doris cannot represent, as a STRING
placeholder, so a hidden
+// TIME key compares in the files' own types: BIGINT microseconds in ORC,
TIMEV2(6) in Parquet. The
+// pair compares as TIMEV2(6) with the ORC value converted exactly, in either
direction, so the NULL
+// tombstone deletes row 1 and 01:00:00.000001 deletes row 2 (3600000001).
Without that rule the
+// mixed pairs fail with NotSupported; a SQL cast would read 3600000001 as
HHMMSS digits, and the
+// old STRING comparison never matched "3600000001" against "01:00:00.000001".
Same-format pairs
+// compare as they are.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteComparesTimeKeysAcrossOrcAndParquet) {
+ const auto test_dir =
+ std::filesystem::temp_directory_path() /
"doris_iceberg_equality_delete_time_formats";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const std::vector<std::pair<std::optional<int64_t>, std::vector<int32_t>>>
cases {
+ {std::nullopt, {2, 3}}, {TIME_01_00_00_000001_MICROS, {1, 3}}};
+ // Every run writes new files: the reader caches Parquet footers and pages
by path and size, so
+ // a same-size file rewritten at one path could be read as its earlier
content.
+ int run = 0;
+ for (auto data_format : {FileFormat::ORC, FileFormat::PARQUET}) {
+ for (auto delete_format : {FileFormat::ORC, FileFormat::PARQUET}) {
+ for (const auto& [tombstone, expected_ids] : cases) {
+ SCOPED_TRACE(::testing::Message()
+ << "data=" << static_cast<int>(data_format)
+ << " delete=" << static_cast<int>(delete_format)
<< " tombstone="
+ << (tombstone.has_value() ?
std::to_string(*tombstone) : "NULL"));
+ const auto file_path = (test_dir / ("data-" +
std::to_string(run))).string();
+ const auto delete_path = (test_dir / ("delete-" +
std::to_string(run++))).string();
+ write_iceberg_time_file(
+ file_path, data_format, {1, 2, 3},
+ {std::nullopt, TIME_01_00_00_000001_MICROS,
TIME_02_00_00_MICROS});
+ write_iceberg_time_file(delete_path, delete_format, {},
{tombstone});
+ std::vector<ColumnDefinition> columns = {
+ make_table_column(0, "id",
std::make_shared<DataTypeInt32>())};
+ auto scan_params = make_local_scan_params(data_format);
+
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, {external_schema_field("id", 0),
+ external_schema_field("k", 1, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::STRING),
+ false, true)})});
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RuntimeProfile profile("time_formats");
+ io::FileReaderStats stats;
+ io::FileCacheStatistics cache_stats;
+ auto io_ctx = make_io_context(&stats, &cache_stats);
+ ShardedKVCache cache(1);
+ iceberg::IcebergTableReader reader;
+ init_iceberg_reader(&reader, columns, &scan_params, io_ctx,
&state, &profile,
+ data_format);
+ auto split = build_split_options(file_path);
+ split.current_split_format = data_format;
+ split.cache = &cache;
+
split.current_range.__set_table_format_params(make_iceberg_table_format_desc(
+ file_path, {make_iceberg_equality_delete_file(
+ delete_path, {1},
+ delete_format == FileFormat::PARQUET
+ ?
TFileFormatType::FORMAT_PARQUET
+ :
TFileFormatType::FORMAT_ORC)}));
+ const auto status = reader.prepare_split(split);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(read_iceberg_ids(&reader, columns), expected_ids);
+ ASSERT_TRUE(reader.close().ok());
+ }
+ }
+ }
+ std::filesystem::remove_all(test_dir);
+}
+
+// `payload` (field id 1) of a data file written before `payload.k` existed: a
struct whose only
+// child is the INT `a` (field id 2). std::nullopt is a NULL `payload`.
+std::shared_ptr<arrow::Array> build_payload_without_key(
+ const std::vector<std::optional<int32_t>>& values) {
+ auto a_field = arrow::field("a", arrow::int32(), true)
+
->WithMetadata(arrow::key_value_metadata({"PARQUET:field_id"}, {"2"}));
+ std::unique_ptr<arrow::ArrayBuilder> builder;
+ EXPECT_TRUE(
+ arrow::MakeBuilder(arrow::default_memory_pool(),
arrow::struct_({a_field}), &builder)
+ .ok());
+ auto& struct_builder = static_cast<arrow::StructBuilder&>(*builder);
+ auto& a_builder =
static_cast<arrow::Int32Builder&>(*struct_builder.field_builder(0));
+ for (const auto& value : values) {
+ if (value.has_value()) {
+ EXPECT_TRUE(struct_builder.Append().ok());
+ EXPECT_TRUE(a_builder.Append(*value).ok());
+ } else {
+ EXPECT_TRUE(struct_builder.AppendNull().ok());
+ }
+ }
+ std::shared_ptr<arrow::Array> result;
+ EXPECT_TRUE(struct_builder.Finish(&result).ok());
+ return result;
+}
+
+// Writes the nested equality delete column `payload` (field id 1) whose child
`k` (field id 3) is
+// an INT64 leaf of `logical_type`, holding `keys` under non-NULL parents.
+void write_nested_int64_key_delete_parquet_file(
+ const std::string& file_path,
+ const std::shared_ptr<const ::parquet::LogicalType>& logical_type,
+ const std::vector<std::optional<int64_t>>& keys) {
+ ::parquet::schema::NodeVector key_fields
{::parquet::schema::PrimitiveNode::Make(
+ "k", ::parquet::Repetition::OPTIONAL, logical_type,
::parquet::Type::INT64, -1, 3)};
+ ::parquet::schema::NodeVector fields {
+ ::parquet::schema::GroupNode::Make("payload",
::parquet::Repetition::OPTIONAL,
+ key_fields,
::parquet::ConvertedType::NONE, 1)};
+ auto schema = std::static_pointer_cast<::parquet::schema::GroupNode>(
+ ::parquet::schema::GroupNode::Make("schema",
::parquet::Repetition::REQUIRED, fields));
+ auto out = arrow::io::FileOutputStream::Open(file_path).ValueOrDie();
+ auto writer = ::parquet::ParquetFileWriter::Open(out, schema);
+ auto* group = writer->AppendRowGroup();
+ auto* key_writer =
static_cast<::parquet::Int64Writer*>(group->NextColumn());
+ for (const auto& key : keys) {
+ // Definition level 2: `payload` and `k` present; 1: `payload` present
with a NULL `k`.
+ const int16_t level = key.has_value() ? 2 : 1;
+ const int64_t value = key.value_or(0);
+ key_writer->WriteBatch(1, &level, nullptr, &value);
+ }
+ key_writer->Close();
+ group->Close();
+ writer->Close();
+ ASSERT_TRUE(out->Close().ok());
+}
+
+// The same column in ORC, where `k` is an Iceberg TIME: a LONG annotated
iceberg.long-type=TIME.
+void write_nested_time_key_delete_orc_file(const std::string& file_path,
+ const
std::vector<std::optional<int64_t>>& micros) {
+ auto type = std::unique_ptr<::orc::Type>(
+
::orc::Type::buildTypeFromString("struct<payload:struct<k:bigint>>"));
+ type->getSubtype(0)->setAttribute("iceberg.id", "1");
+ type->getSubtype(0)->getSubtype(0)->setAttribute("iceberg.id", "3");
+ type->getSubtype(0)->getSubtype(0)->setAttribute("iceberg.long-type",
"TIME");
+ MemoryOutputStream stream(1024 * 1024);
+ ::orc::WriterOptions options;
+ auto writer = ::orc::createWriter(*type, &stream, options);
+ auto batch = writer->createRowBatch(micros.size());
+ auto& root = dynamic_cast<::orc::StructVectorBatch&>(*batch);
+ root.numElements = micros.size();
+ auto& payload = dynamic_cast<::orc::StructVectorBatch&>(*root.fields[0]);
+ payload.numElements = micros.size();
+ auto& keys = dynamic_cast<::orc::LongVectorBatch&>(*payload.fields[0]);
+ keys.numElements = micros.size();
+ keys.hasNulls = true;
+ for (size_t row = 0; row < micros.size(); ++row) {
+ keys.notNull[row] = micros[row].has_value();
+ keys.data[row] = micros[row].value_or(0);
+ }
+ writer->add(*batch);
+ writer->close();
+ std::ofstream out(file_path, std::ios::binary);
+ out.write(stream.getData(),
static_cast<std::streamsize>(stream.getLength()));
+}
+
+// Iceberg TIME has no Doris type, so the FE sends `payload.k` as a STRING
placeholder whose initial
+// default is Iceberg's text of the value. That text is parsed in the
tombstones' type (an ORC TIME
+// tombstone, BIGINT, compares as TIMEV2(6)). Rows under the data file's NULL
`payload` keep a NULL
+// key and the others take the default, so a NULL tombstone deletes only row
1, the default's
+// tombstone rows 2 and 3, and another value none. Comparing the STRING
default with the TIMEV2
+// tombstones failed the split with NotSupported.
+TEST(IcebergV2ReaderTest,
IcebergMissingNestedPlaceholderKeyParsesTextDefaultInTombstoneType) {
+ struct PlaceholderDefault {
+ std::string name;
+ std::string text_default;
+ // Writes a delete file holding the tombstone, in the file's unit.
+ std::function<TFileFormatType::type(const std::string&,
std::optional<int64_t>)> writer;
+ int64_t default_value;
+ int64_t other_value;
+ bool enable_mapping_timestamp_tz;
+ };
+ const std::vector<PlaceholderDefault> kinds {
+ {"Parquet TIME", "00:00",
+ [](const std::string& path, std::optional<int64_t> key) {
+ write_nested_int64_key_delete_parquet_file(path,
iceberg_parquet_time_type(),
+ {key});
+ return TFileFormatType::FORMAT_PARQUET;
+ },
+ 0, TIME_01_00_00_000001_MICROS, false},
+ {"ORC TIME", "00:00",
+ [](const std::string& path, std::optional<int64_t> key) {
+ write_nested_time_key_delete_orc_file(path, {key});
+ return TFileFormatType::FORMAT_ORC;
+ },
+ 0, TIME_01_00_00_000001_MICROS, false}};
+ // A directory per run, as in
IcebergEqualityDeleteComparesTimeKeysAcrossOrcAndParquet: the
+ // 1-row INT64 delete files of one kind have the same size whatever the
tombstone.
+ int run = 0;
+ for (const auto& kind : kinds) {
+ const std::vector<std::pair<std::optional<int64_t>,
std::vector<int32_t>>> cases {
+ {std::nullopt, {2, 3}}, {kind.default_value, {1}},
{kind.other_value, {1, 2, 3}}};
+ for (const auto& [tombstone, expected_ids] : cases) {
+ SCOPED_TRACE(::testing::Message()
+ << kind.name << " tombstone="
+ << (tombstone.has_value() ?
std::to_string(*tombstone) : "NULL"));
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_placeholder_text_default_" +
+ std::to_string(run++),
+ {{.ids = {1, 2, 3}, .keys =
build_payload_without_key({std::nullopt, 1, 2})}},
+ nullptr,
+ external_struct_schema_field(
+ "payload", 1,
+ {external_schema_field("a", 2, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::INT),
+ false, true),
+ external_schema_field("k", 3, {},
kind.text_default,
+
external_primitive_type(TPrimitiveType::STRING),
+ false, true)},
+ true),
+ &ids, 3, kind.enable_mapping_timestamp_tz,
+ [&kind, tombstone = tombstone](const std::string& path) {
+ return kind.writer(path, tombstone);
+ });
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(ids, (std::vector<std::vector<int32_t>> {expected_ids}));
+ }
+ }
+}
+
+// A timestamptz_ns `payload.k` behind the STRING placeholder: its Parquet
tombstones are read as
+// TIMESTAMPTZ(6), the nanoseconds floored, so a non-NULL default cannot be
compared exactly and the
+// scan fails. Parsing the default instead deleted the rows for the default
`...0000001` against an
+// epoch tombstone and for the default epoch against a tombstone 100 ns later,
and kept them for the
+// default `...0000005` against its own tombstone. Without a default the key
is NULL: a NULL
+// tombstone deletes every row, an epoch tombstone none.
+TEST(IcebergV2ReaderTest,
IcebergMissingPlaceholderTimestampTzNsDefaultFailsTheScan) {
+ const auto write_tombstone = [](std::optional<int64_t> nanos) {
+ return [nanos](const std::string& path) {
+ write_nested_int64_key_delete_parquet_file(
+ path,
+ ::parquet::LogicalType::Timestamp(true,
+
::parquet::LogicalType::TimeUnit::NANOS),
+ {nanos});
+ return TFileFormatType::FORMAT_PARQUET;
+ };
+ };
+ const auto read_ids = [](const std::string& test_name,
+ const std::optional<std::string>& text_default,
+ const EqualityDeleteFileWriter& write_delete_file,
+ std::vector<std::vector<int32_t>>* ids) {
+ return read_ids_after_key_equality_delete(
+ test_name,
+ {{.ids = {1, 2, 3}, .keys =
build_payload_without_key({std::nullopt, 1, 2})}},
+ nullptr,
+ external_struct_schema_field(
+ "payload", 1,
+ {external_schema_field("a", 2, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::INT), false,
+ true),
+ external_schema_field("k", 3, {}, text_default,
+
external_primitive_type(TPrimitiveType::STRING),
+ false, true)},
+ true),
+ ids, 3, true, write_delete_file);
+ };
+ const std::vector<std::pair<std::string, int64_t>> rejected {
+ {"1970-01-01 00:00:00.0000001+00:00", 0},
+ {"1970-01-01 00:00:00+00:00", 100},
+ {"1970-01-01 00:00:00.0000005+00:00", 500}};
+ int run = 0;
+ for (const auto& [text_default, tombstone] : rejected) {
+ SCOPED_TRACE(::testing::Message() << text_default << " tombstone=" <<
tombstone);
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids(
+ "doris_iceberg_equality_delete_timestamptz_ns_default_" +
std::to_string(run++),
+ text_default, write_tombstone(tombstone), &ids);
+ ASSERT_FALSE(status.ok()) << "ids=" << ::testing::PrintToString(ids);
+ EXPECT_TRUE(status.is<ErrorCode::NOT_IMPLEMENTED_ERROR>()) << status;
+ EXPECT_NE(status.to_string().find("timestamptz_ns key"),
std::string::npos) << status;
+ }
+
+ const std::vector<std::pair<std::optional<int64_t>, std::vector<int32_t>>>
without_default {
+ {std::nullopt, {}}, {0, {1, 2, 3}}};
+ for (const auto& [tombstone, expected_ids] : without_default) {
+ SCOPED_TRACE(::testing::Message()
+ << "no default, tombstone="
+ << (tombstone.has_value() ? std::to_string(*tombstone) :
"NULL"));
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids(
+ "doris_iceberg_equality_delete_timestamptz_ns_default_" +
std::to_string(run++),
+ std::nullopt, write_tombstone(tombstone), &ids);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(ids, (std::vector<std::vector<int32_t>> {expected_ids}));
+ }
+}
+
+// Keys whose types do not convert exactly in either direction fail the scan
instead of being
+// compared in a third type.
+TEST(IcebergV2ReaderTest, IcebergEqualityDeleteRejectsIncomparableKeyTypes) {
+ std::vector<std::vector<int32_t>> ids;
+ const auto status = read_ids_after_key_equality_delete(
+ "doris_iceberg_equality_delete_incomparable_types",
+ {{.ids = {1}, .keys = build_nullable_int32_keys({1})}},
+ build_nullable_string_array({"1"}),
+ external_schema_field("k", 1, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::INT), false, true),
+ &ids);
+ ASSERT_FALSE(status.ok());
+ EXPECT_TRUE(status.is<ErrorCode::NOT_IMPLEMENTED_ERROR>()) << status;
+ EXPECT_NE(status.to_string().find("neither type converts exactly to the
other"),
+ std::string::npos)
+ << status;
+}
+
TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteMatchesNullForMissingDataColumn) {
const auto test_dir = std::filesystem::temp_directory_path() /
"doris_iceberg_equality_delete_missing_column_test";
@@ -5559,6 +6301,96 @@ TEST(IcebergV2ReaderTest,
IcebergEqualityDeletePreservesTimestampTzAcrossDstFold
std::filesystem::remove_all(test_dir);
}
+// Iceberg imports a Spark or Hive INT96 timestamp as timestamptz, but INT96
reads as DATETIMEV2
+// while the adjusted-UTC tombstone of the same field reads as TIMESTAMPTZ,
and neither widens to
+// the other. They compare as TIMESTAMPTZ, the INT96 key converted in the
session time zone as the
+// projection converts it, so the tombstone T1 deletes the row holding T1.
This plan carries no
+// Parquet timestamp-semantics version, so INT96 also decodes in the session
time zone; converting
+// it back in a zone other than Asia/Shanghai, such as UTC, would miss T1 by
eight hours. Without
+// the TIMESTAMPTZ rule the split fails with NotSupported, with or without the
key projected.
+TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteComparesInt96DataKeyAsTimestampTz) {
+ TimezoneUtils::load_timezones_to_cache();
+ const auto test_dir = std::filesystem::temp_directory_path() /
+
"doris_iceberg_equality_delete_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_timestamp_int_parquet_file(file_path, {T1_MICROS, T2_MICROS}, {1,
2}, true);
+ auto physical_reader = ::parquet::ParquetFileReader::OpenFile(file_path,
false);
+ ASSERT_EQ(::parquet::Type::INT96,
+
physical_reader->metadata()->schema()->Column(0)->physical_type());
+ physical_reader->Close();
+ write_iceberg_timestamp_equality_delete_parquet_file(delete_file_path, 0,
T1_MICROS,
+ "event_time", true);
+
+ for (const bool project_key : {false, true}) {
+ SCOPED_TRACE(project_key ? "id and event_time projected" : "id
projected");
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(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 scan_params = make_local_parquet_scan_params();
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ 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("event_time", 0, {}, std::nullopt,
+
external_primitive_type(TPrimitiveType::TIMESTAMPTZ, -1, 6)),
+ 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) {
+ // The projection converts the INT96 value back in the same zone,
giving T2.
+ 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);
+}
+
TEST(IcebergV2ReaderTest,
IcebergEqualityDeleteUsesNameMappingWithoutFileFieldIds) {
const auto test_dir = std::filesystem::temp_directory_path() /
"doris_iceberg_equality_delete_name_mapping_test";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
