This is an automated email from the ASF dual-hosted git repository.
morningman 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 2760a7663b9 [fix](be) Compare Iceberg V1 equality-delete keys in the
promoted type (#69691)
2760a7663b9 is described below
commit 2760a7663b95558b142e4457383a8f7ec7a2d2bb
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Oct 29 04:58:57 2026 +0000
[fix](be) Compare Iceberg V1 equality-delete keys in the promoted type
(#69691)
### What problem does this PR solve?
Issue Number: None
Related PR: #66538 (branch-4.1), #68267
Problem Summary:
**In short.** This brings the V1-reader part of branch-4.1 #66538
(Iceberg V3 default values) to master: Iceberg equality deletes keep
working after the type of their key column is promoted (INT to BIGINT,
FLOAT to DOUBLE, a wider DECIMAL precision). It does not copy
branch-4.1's approach, which compares a key in the delete file's older
type; it compares in the scan schema's type, as #68267 already does in
the V2 reader.
**Background.** Iceberg allows widening a column's type, and files keep
the type they were written with: an equality delete written while `id`
was `int` stores its key as INT32, and so does a data file written then.
The V1 reader (`enable_file_scanner_v2 = false`) loads each delete file
into a delete block and compares it with the data rows, which the
Parquet and ORC readers already return in the table's current type.
**The problem, and what it cost.** The V1 reader compared every key in
the type of the delete file it came from. After a promotion, any of
these fails:
| Case (V1 reader, `id` or `payload.existing` promoted from INT to
BIGINT) | master |
|---|---|
| `SELECT id`, delete written before the promotion | Query fails: `Bad
cast from type ColumnVector<BIGINT> to ColumnVector<INT>` |
| `SELECT id`, one delete written before the promotion and one after |
Query fails: the INT and BIGINT keys are merged into one delete block
(debug builds abort on the block's type check) |
| `SELECT v` (key not projected), same two deletes | Same |
| `SELECT id`, delete on `payload.existing` written after the promotion,
data file written before it | The INT values of the old file are copied
into a BIGINT column: debug builds abort with `Bad cast`, release builds
read the wrong bytes and keep or delete the wrong rows |
branch-4.1 / branch-4.2 handle these by casting the data key down to the
delete file's type. That narrowing is what #68267 fixed in the V2
reader: a data file written after the promotion can hold 4294967296,
which becomes NULL as INT and is then deleted by the NULL key of a
delete file written before the promotion (equality deletes use null-safe
equality); with `enable_strict_cast` the whole scan fails instead.
**How this PR fixes it.**
- The type a key is compared in is the field's type in the scan schema
(`equality_delete_key_type`, the same rule as V2's
`restore_current_primitive_type`). A plan without a scan schema keeps
the delete file's type, as before, and STRING / VARBINARY keys keep
theirs too, since they match by bytes.
- Delete keys, and nested keys extracted from the physical root of an
older data file, are widened to that type with
`converter::ColumnTypeConverter`, the converter the V1 Parquet and ORC
readers already apply to the data column of a promoted field, so both
sides hold the same value (`widen_equality_delete_key`).
`NestedEqualityDeleteColumn` records the leaf's source type for that.
- Delete blocks and hidden key carriers are built in the comparison
type, so deletes written before and after a promotion share one block.
**Results.**
| Case (V1 reader) | master | This PR |
|---|---|---|
| `SELECT id`, INT delete `id = 2`, data 1, 2, 3 | `Bad cast` | 1, 3 |
| `SELECT id`, INT delete `id = 2` + BIGINT delete `id = 3` | aborts
(debug) / fails | 1 |
| `SELECT v`, same two deletes | aborts (debug) / fails | 10 |
| `SELECT id`, BIGINT delete `payload.existing = 20` on an old INT file
| `Bad cast` (debug) / wrong rows (release) | 1, 3 |
| `SELECT id`, data `NULL, 0, 1, 4294967296`, INT delete holding NULL |
`Bad cast` (branch-4.x: 0, 1) | 0, 1, 4294967296 |
---
be/src/format/table/equality_delete.cpp | 67 +++-
be/src/format/table/equality_delete.h | 28 ++
be/src/format/table/iceberg_reader.cpp | 22 +-
be/src/format/table/iceberg_reader_mixin.h | 26 +-
.../format/table/iceberg/iceberg_reader_test.cpp | 395 ++++++++++++++++++++-
5 files changed, 517 insertions(+), 21 deletions(-)
diff --git a/be/src/format/table/equality_delete.cpp
b/be/src/format/table/equality_delete.cpp
index 21071a89114..193526e1266 100644
--- a/be/src/format/table/equality_delete.cpp
+++ b/be/src/format/table/equality_delete.cpp
@@ -17,23 +17,82 @@
#include "format/table/equality_delete.h"
+#include <gen_cpp/ExternalTableSchema_types.h>
+
+#include "core/assert_cast.h"
#include "core/column/column_const.h"
#include "core/column/column_nullable.h"
+#include "core/data_type/data_type_factory.hpp"
#include "exprs/create_predicate_function.h"
+#include "format/column_type_convert.h"
#include "util/hash_util.hpp"
namespace doris {
-namespace {
+DataTypePtr equality_delete_key_type(const schema::external::TField* field,
+ const DataTypePtr& delete_key_type) {
+ DORIS_CHECK(delete_key_type->is_nullable());
+ // A plan from an FE that sends no scan schema, or no type for the field,
leaves only the delete
+ // file's own type, which is how keys were compared before.
+ if (field == nullptr || !field->__isset.type ||
is_equality_delete_byte_type(delete_key_type)) {
+ return delete_key_type;
+ }
+ const auto primitive_type = thrift_to_type(field->type.type);
+ DORIS_CHECK(!is_complex_type(primitive_type));
+ return DataTypeFactory::instance().create_data_type(
+ primitive_type, true, field->type.__isset.precision ?
field->type.precision : 0,
+ field->type.__isset.scale ? field->type.scale : 0,
+ field->type.__isset.len ? field->type.len : -1);
+}
-// Iceberg BINARY / FIXED keys are read as VARBINARY and string keys as
STRING, and the delete file and
-// the data file may carry the same key in different ones of these classes (a
delete block holding
-// STRING bytes for a VARBINARY data column, say). Keys of these types
therefore match by their bytes.
bool is_equality_delete_byte_type(const DataTypePtr& type) {
const auto primitive_type = remove_nullable(type)->get_primitive_type();
return is_string_type(primitive_type) || is_varbinary(primitive_type);
}
+Status widen_equality_delete_key(ColumnPtr column, const DataTypePtr&
column_type,
+ const DataTypePtr& key_type, ColumnPtr*
result) {
+ DORIS_CHECK(result != nullptr);
+ DORIS_CHECK(column_type->is_nullable());
+ DORIS_CHECK(key_type->is_nullable());
+ if (column_type->equals(*key_type)) {
+ *result = std::move(column);
+ return Status::OK();
+ }
+ // Byte keys keep the type they were read in and match by their bytes, and
the converters below
+ // would turn VARBINARY into its text form.
+ if (is_equality_delete_byte_type(column_type) ||
is_equality_delete_byte_type(key_type))
+ [[unlikely]] {
+ return Status::InternalError("Unexpected type change of equality
delete key from {} to {}",
+ column_type->get_name(),
key_type->get_name());
+ }
+ // The Parquet and ORC readers widen the data column of a promoted field
with the same
+ // converters, so a widened delete key holds exactly the value of the data
key it must match.
+ auto converter =
+ converter::ColumnTypeConverter::get_converter(column_type,
key_type, converter::COMMON);
+ if (!converter->support()) {
+ return Status::InternalError(
+ "Not support type change of equality delete key from {} to {}:
{}",
+ column_type->get_name(), key_type->get_name(),
converter->get_error_msg());
+ }
+ if (converter->is_consistent()) {
+ // The same column layout, for example DATETIMEV2 of another scale.
+ *result = std::move(column);
+ return Status::OK();
+ }
+ MutableColumnPtr widened = key_type->create_column();
+ // The converters fill only the nested values; as in
OrcReader::_orc_column_to_doris_column, the
+ // null map is copied before converting.
+ const auto& source_null_map = assert_cast<const
ColumnNullable&>(*column).get_null_map_data();
+
assert_cast<ColumnNullable&>(*widened).get_null_map_data().insert(source_null_map.begin(),
+
source_null_map.end());
+ RETURN_IF_ERROR(converter->convert(column, widened));
+ *result = std::move(widened);
+ return Status::OK();
+}
+
+namespace {
+
// The rows of a byte-key column. A const column repeats its single value,
which can be NULL: when the FE does
// not send Iceberg scan semantics v2, a key column that the data file lacks
is filled from
// Const(Nullable(String) NULL), and its rows must match a NULL delete key
rather than an empty string.
diff --git a/be/src/format/table/equality_delete.h
b/be/src/format/table/equality_delete.h
index 572766091a6..058b6445bdf 100644
--- a/be/src/format/table/equality_delete.h
+++ b/be/src/format/table/equality_delete.h
@@ -19,8 +19,36 @@
#include "exprs/hybrid_set.h"
#include "runtime/runtime_profile.h"
+namespace doris::schema::external {
+class TField;
+} // namespace doris::schema::external
+
namespace doris {
+// The type an equality-delete key is compared in: the field's type in the
scan schema (`field`, which
+// is null when the plan carries no schema), as in the V2 reader. Data files
are read in that type, so
+// an equality delete written before a type promotion is widened to it rather
than the data narrowed
+// to the delete file's type. Example: `id` promoted from INT to BIGINT, a
data file written after the
+// promotion holds 4294967296, and a delete file written before it holds the
NULL key. Narrowed to
+// INT, 4294967296 became NULL and was deleted by that NULL key; widened, the
delete key is compared
+// as BIGINT and deletes only rows whose id is NULL. STRING / VARBINARY keys
keep `delete_key_type`:
+// they match by their bytes whatever the two sides were read as.
+DataTypePtr equality_delete_key_type(const schema::external::TField* field,
+ const DataTypePtr& delete_key_type);
+
+// Iceberg BINARY / FIXED keys are read as VARBINARY and string keys as
STRING, and the delete file and
+// the data file may carry the same key in different ones of these classes (a
delete block holding
+// STRING bytes for a VARBINARY data column, say). Keys of these types
therefore match by their bytes.
+bool is_equality_delete_byte_type(const DataTypePtr& type);
+
+// Widens an equality-delete key read in `column_type` to `key_type`, the type
the key is compared in.
+// A file written before an Iceberg type promotion holds the key in the old
type: the INT key of a
+// column since promoted to BIGINT becomes BIGINT here. Iceberg's promotions
(INT to BIGINT, FLOAT to
+// DOUBLE, a wider DECIMAL precision) only widen, so every value is kept.
Returns `column` itself when
+// the types are equal.
+Status widen_equality_delete_key(ColumnPtr column, const DataTypePtr&
column_type,
+ const DataTypePtr& key_type, ColumnPtr*
result);
+
/**
* Support Iceberg equality delete.
* If there's only one delete column in delete file, use
`SimpleEqualityDelete`,
diff --git a/be/src/format/table/iceberg_reader.cpp
b/be/src/format/table/iceberg_reader.cpp
index d49d690a01e..5c14badbbc5 100644
--- a/be/src/format/table/iceberg_reader.cpp
+++ b/be/src/format/table/iceberg_reader.cpp
@@ -728,8 +728,10 @@ Status
IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
RETURN_IF_ERROR(find_projected_equality_key_path(*root_path.front(),
key_path,
slot->second->type(),
child_indexes,
&leaf_type, &found));
- // The projected leaf has the table's current type. A key written
before a type promotion
- // (INT promoted to BIGINT) keeps its own type, so it still reads the
physical root.
+ // The projected leaf has the table's current type, which is the type
a key is compared in
+ // (equality_delete_key_type) even when the key was written before a
type promotion. Only a
+ // STRING / VARBINARY key keeps its delete file's type: a STRING key
under a VARBINARY
+ // projected leaf still reads the physical root, where the leaf is
read as STRING too.
if (found &&
remove_nullable(leaf_type)->equals(*remove_nullable(key_type)) &&
projected_root_reads_physical_path(*ctx->table_info_node,
slot->second->col_name(),
slot->second->type(),
*child_indexes, file_path)) {
@@ -834,6 +836,7 @@ Status
IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
.field_id = field_id,
.block_name = table_col_name,
.leaf_type = _expand_columns[i].type,
+ .source_leaf_type = _expand_columns[i].type,
.child_indexes = std::move(projected_child_indexes),
.missing_value = nullptr,
.source_block_name = std::move(projected_root_name),
@@ -846,6 +849,14 @@ Status
IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
field_id, _expand_columns[i].type,
file_path.fields.size(),
&missing_value));
}
+ // The physical root holds the leaf in the data file's own type,
which is narrower than
+ // the compared type when the file was written before a type
promotion: payload.existing
+ // promoted from INT to BIGINT is INT in older files, and copying
those INT values into a
+ // BIGINT column read the wrong bytes. A leaf the file lacks is
the missing default,
+ // which is built in the compared type.
+ const DataTypePtr source_leaf_type =
+ complete_file_path ?
make_nullable(file_path.fields.back()->data_type)
+ : _expand_columns[i].type;
const auto root_carrier =
physical_root_carriers.find(file_column->name);
if (root_carrier != physical_root_carriers.end()) {
// Keys payload.a and payload.b: the carrier of payload.a
reads payload, and
@@ -854,6 +865,7 @@ Status
IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
.field_id = field_id,
.block_name = table_col_name,
.leaf_type = _expand_columns[i].type,
+ .source_leaf_type = source_leaf_type,
.child_indexes = file_path.child_indexes,
.missing_value = std::move(missing_value),
.source_block_name = root_carrier->second,
@@ -865,6 +877,7 @@ Status
IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
.field_id = field_id,
.block_name = table_col_name,
.leaf_type = _expand_columns[i].type,
+ .source_leaf_type = source_leaf_type,
.child_indexes = file_path.child_indexes,
.missing_value = std::move(missing_value),
});
@@ -1234,16 +1247,21 @@ Status
IcebergOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
.field_id = field_id,
.block_name = table_col_name,
.leaf_type = _expand_columns[i].type,
+ .source_leaf_type = _expand_columns[i].type,
.child_indexes = file_path.child_indexes,
.missing_value = std::move(missing_value),
});
_expand_columns[i].type =
make_nullable(convert_to_doris_type(file_column));
_expand_columns[i].column =
_expand_columns[i].type->create_column();
} else if (!file_path.child_indexes.empty()) {
+ // The root is read in the data file's own types; a leaf written
before a type
+ // promotion (INT payload.existing of a column now BIGINT) is
widened after extraction.
_nested_equality_delete_columns.push_back({
.field_id = field_id,
.block_name = table_col_name,
.leaf_type = _expand_columns[i].type,
+ .source_leaf_type =
+
make_nullable(convert_to_doris_type(file_path.fields.back())),
.child_indexes = file_path.child_indexes,
.missing_value = nullptr,
});
diff --git a/be/src/format/table/iceberg_reader_mixin.h
b/be/src/format/table/iceberg_reader_mixin.h
index bf47010289e..ca5b63f5ddd 100644
--- a/be/src/format/table/iceberg_reader_mixin.h
+++ b/be/src/format/table/iceberg_reader_mixin.h
@@ -362,7 +362,12 @@ protected:
struct NestedEqualityDeleteColumn {
int32_t field_id = -1;
std::string block_name;
+ // The type the key is compared in (equality_delete_key_type).
DataTypePtr leaf_type;
+ // The type of the leaf in the column it is extracted from, widened to
leaf_type afterwards.
+ // A file written before a type promotion holds the old type:
payload.existing promoted from
+ // INT to BIGINT is still INT in the delete files and data files
written before.
+ DataTypePtr source_leaf_type;
std::vector<size_t> child_indexes;
ColumnPtr missing_value;
// Block column holding the key's root struct. Empty when the key's
own column reads the
@@ -677,12 +682,15 @@ Status
IcebergReaderMixin<BaseReader>::_build_parquet_equality_delete_read_specs
return Status::NotSupported(
"Iceberg equality delete does not support complex column
{}", leaf->name);
}
+ const auto delete_key_type = make_nullable(leaf->data_type);
read_specs->push_back({
.nested_field =
{
.field_id = field_id,
.block_name = leaf->name,
- .leaf_type = make_nullable(leaf->data_type),
+ .leaf_type =
equality_delete_key_type(_find_schema_field(field_id),
+
delete_key_type),
+ .source_leaf_type = delete_key_type,
.child_indexes = std::move(child_indexes),
.missing_value = nullptr,
},
@@ -724,12 +732,15 @@ Status
IcebergReaderMixin<BaseReader>::_build_orc_equality_delete_read_specs(
return Status::NotSupported(
"Iceberg equality delete does not support complex column
{}", names.back());
}
+ const auto delete_key_type =
make_nullable(reader->convert_to_doris_type(leaf));
read_specs->push_back({
.nested_field =
{
.field_id = field_id,
.block_name = names.back(),
- .leaf_type =
make_nullable(reader->convert_to_doris_type(leaf)),
+ .leaf_type =
equality_delete_key_type(_find_schema_field(field_id),
+
delete_key_type),
+ .source_leaf_type = delete_key_type,
.child_indexes = std::move(child_indexes),
.missing_value = nullptr,
},
@@ -1109,8 +1120,9 @@ Status
IcebergReaderMixin<BaseReader>::_extract_nested_equality_delete_column(
}
}
- auto result =
ColumnNullable::create(remove_nullable(nested_field.leaf_type)->create_column(),
- ColumnUInt8::create());
+ DORIS_CHECK(nested_field.source_leaf_type != nullptr);
+ auto result = ColumnNullable::create(
+ remove_nullable(nested_field.source_leaf_type)->create_column(),
ColumnUInt8::create());
auto& result_data = result->get_nested_column();
auto& result_null_map = result->get_null_map_data();
result_data.reserve(root_column->size());
@@ -1131,8 +1143,9 @@ Status
IcebergReaderMixin<BaseReader>::_extract_nested_equality_delete_column(
result_null_map.push_back(0);
}
}
- *leaf_column = std::move(result);
- return Status::OK();
+ return widen_equality_delete_key(std::move(result),
+
make_nullable(nested_field.source_leaf_type),
+ make_nullable(nested_field.leaf_type),
leaf_column);
}
template <typename BaseReader>
@@ -1232,6 +1245,7 @@ Status
IcebergReaderMixin<BaseReader>::_create_missing_equality_delete_value(
.field_id = field_id,
.block_name = "",
.leaf_type = delete_key_type,
+ .source_leaf_type = delete_key_type,
.child_indexes = std::vector<size_t>(table_path.size() -
missing_index - 1, 0),
.missing_value = nullptr,
};
diff --git a/be/test/format/table/iceberg/iceberg_reader_test.cpp
b/be/test/format/table/iceberg/iceberg_reader_test.cpp
index 08767f5e022..2fbabcabc7f 100644
--- a/be/test/format/table/iceberg/iceberg_reader_test.cpp
+++ b/be/test/format/table/iceberg/iceberg_reader_test.cpp
@@ -84,9 +84,10 @@
namespace doris {
void write_iceberg_int_orc_file(const std::string& file_path, const
std::string& field_name,
- int32_t field_id, const std::vector<int64_t>&
values) {
+ int32_t field_id, const std::vector<int64_t>&
values,
+ const std::string& orc_type = "int") {
auto type = std::unique_ptr<::orc::Type>(
- ::orc::Type::buildTypeFromString("struct<" + field_name +
":int>"));
+ ::orc::Type::buildTypeFromString("struct<" + field_name + ":" +
orc_type + ">"));
type->getSubtype(0)->setAttribute("iceberg.id", std::to_string(field_id));
MemoryOutputStream memory_stream(1024 * 1024);
@@ -456,10 +457,11 @@ void write_iceberg_nested_struct_orc_file(const
std::string& file_path,
const std::vector<int64_t>& values,
const std::vector<bool>&
parent_nulls,
const std::string& child_name =
"existing",
- int32_t child_id = 2) {
+ int32_t child_id = 2,
+ const std::string& child_type =
"int") {
DORIS_CHECK(values.size() == parent_nulls.size());
- auto type = std::unique_ptr<::orc::Type>(
- ::orc::Type::buildTypeFromString("struct<payload:struct<" +
child_name + ":int>>"));
+ auto type = std::unique_ptr<::orc::Type>(::orc::Type::buildTypeFromString(
+ "struct<payload:struct<" + child_name + ":" + child_type + ">>"));
type->getSubtype(0)->setAttribute("iceberg.id", "1");
type->getSubtype(0)->getSubtype(0)->setAttribute("iceberg.id",
std::to_string(child_id));
@@ -592,9 +594,82 @@ void write_iceberg_int_equality_delete_parquet_file(const
std::string& file_path
1, properties.build()));
}
-Status create_single_int_tuple_descriptor(ObjectPool* object_pool, const
std::string& column_name,
- int32_t field_id, DescriptorTbl**
descriptor_table,
- const TupleDescriptor**
tuple_descriptor) {
+// One optional INT32 or INT64 column `field_name` (field `field_id`);
std::nullopt writes NULL.
+void write_iceberg_single_integer_parquet_file(const std::string& file_path,
+ const std::string& field_name,
int32_t field_id,
+ bool int64_column,
+ const
std::vector<std::optional<int64_t>>& values) {
+ std::shared_ptr<arrow::Array> array;
+ if (int64_column) {
+ arrow::Int64Builder builder;
+ for (const auto& value : values) {
+ DORIS_CHECK((value.has_value() ? builder.Append(*value) :
builder.AppendNull()).ok());
+ }
+ DORIS_CHECK(builder.Finish(&array).ok());
+ } else {
+ arrow::Int32Builder builder;
+ for (const auto& value : values) {
+ DORIS_CHECK((value.has_value() ?
builder.Append(static_cast<int32_t>(*value))
+ : builder.AppendNull())
+ .ok());
+ }
+ DORIS_CHECK(builder.Finish(&array).ok());
+ }
+ auto field = arrow::field(field_name, int64_column ? arrow::int64() :
arrow::int32(), true)
+
->WithMetadata(arrow::key_value_metadata({"PARQUET:field_id"},
+
{std::to_string(field_id)}));
+ auto table = arrow::Table::Make(arrow::schema({field}), {array});
+ auto output = arrow::io::FileOutputStream::Open(file_path);
+ DORIS_CHECK(output.ok());
+ ::parquet::WriterProperties::Builder properties;
+ properties.version(::parquet::ParquetVersion::PARQUET_2_6);
+ properties.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ properties.compression(::parquet::Compression::UNCOMPRESSED);
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), *output,
+
static_cast<int64_t>(values.size()),
+ properties.build()));
+}
+
+// `payload STRUCT<existing BIGINT>` (field 1, child 2): an equality delete on
payload.existing
+// written after the field was promoted from INT.
+void write_iceberg_bigint_nested_struct_parquet_file(const std::string&
file_path,
+ const
std::vector<int64_t>& values) {
+ auto existing_field =
+ arrow::field("existing", arrow::int64(), false)
+
->WithMetadata(arrow::key_value_metadata({"PARQUET:field_id"}, {"2"}));
+ auto struct_type = arrow::struct_({existing_field});
+ std::vector<std::shared_ptr<arrow::ArrayBuilder>> field_builders;
+ field_builders.emplace_back(std::make_shared<arrow::Int64Builder>());
+ arrow::StructBuilder builder(struct_type, arrow::default_memory_pool(),
+ std::move(field_builders));
+ auto* existing_builder =
assert_cast<arrow::Int64Builder*>(builder.field_builder(0));
+ for (const auto value : values) {
+ DORIS_CHECK(builder.Append().ok());
+ DORIS_CHECK(existing_builder->Append(value).ok());
+ }
+ std::shared_ptr<arrow::Array> payload;
+ DORIS_CHECK(builder.Finish(&payload).ok());
+ auto payload_field =
+ arrow::field("payload", struct_type, true)
+
->WithMetadata(arrow::key_value_metadata({"PARQUET:field_id"}, {"1"}));
+ auto table = arrow::Table::Make(arrow::schema({payload_field}), {payload});
+
+ auto output = arrow::io::FileOutputStream::Open(file_path);
+ DORIS_CHECK(output.ok());
+ ::parquet::WriterProperties::Builder properties;
+ properties.version(::parquet::ParquetVersion::PARQUET_2_6);
+ properties.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ properties.compression(::parquet::Compression::UNCOMPRESSED);
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), *output,
+
static_cast<int64_t>(values.size()),
+ properties.build()));
+}
+
+Status create_single_scalar_tuple_descriptor(ObjectPool* object_pool,
+ const std::string& column_name,
int32_t field_id,
+ TPrimitiveType::type slot_type,
+ DescriptorTbl** descriptor_table,
+ const TupleDescriptor**
tuple_descriptor) {
TDescriptorTable thrift_table;
TTableDescriptor table_descriptor;
table_descriptor.__set_id(0);
@@ -606,7 +681,7 @@ Status create_single_int_tuple_descriptor(ObjectPool*
object_pool, const std::st
TTypeNode type_node;
type_node.__set_type(TTypeNodeType::SCALAR);
TScalarType scalar_type;
- scalar_type.__set_type(TPrimitiveType::INT);
+ scalar_type.__set_type(slot_type);
type_node.__set_scalar_type(scalar_type);
TTypeDesc type;
type.__set_types({type_node});
@@ -637,6 +712,14 @@ Status create_single_int_tuple_descriptor(ObjectPool*
object_pool, const std::st
return Status::OK();
}
+Status create_single_int_tuple_descriptor(ObjectPool* object_pool, const
std::string& column_name,
+ int32_t field_id, DescriptorTbl**
descriptor_table,
+ const TupleDescriptor**
tuple_descriptor) {
+ return create_single_scalar_tuple_descriptor(object_pool, column_name,
field_id,
+ TPrimitiveType::INT,
descriptor_table,
+ tuple_descriptor);
+}
+
Status create_two_int_tuple_descriptor(
ObjectPool* object_pool, const std::array<std::pair<std::string,
int32_t>, 2>& columns,
DescriptorTbl** descriptor_table, const TupleDescriptor**
tuple_descriptor) {
@@ -1045,6 +1128,15 @@ schema::external::TFieldPtr
make_external_int_field(const std::string& name, int
return field_ptr;
}
+// A field promoted from INT to BIGINT: its type in the scan schema is BIGINT.
+schema::external::TFieldPtr make_external_bigint_field(const std::string&
name, int32_t field_id) {
+ auto field_ptr = make_external_int_field(name, field_id, std::nullopt);
+ TColumnType type;
+ type.__set_type(TPrimitiveType::BIGINT);
+ field_ptr.field_ptr->__set_type(type);
+ return field_ptr;
+}
+
schema::external::TFieldPtr make_external_struct_field(
const std::string& name, int32_t field_id,
std::vector<schema::external::TFieldPtr> children,
std::vector<std::string> aliases = {},
@@ -3506,6 +3598,291 @@ TEST_F(IcebergReaderTest,
&timezone_obj,
cache.get(), true);
}
+// Reads `data_file` through the V1 Iceberg reader with the equality deletes
`delete_files` (path and
+// key field ids) applied, under the scan schema `current_root`. `block` holds
the projected slots of
+// `tuple_descriptor` in order; all rows fit in one batch.
+Status read_v1_with_equality_deletes(
+ TFileFormatType::type file_format, const std::string& data_file,
+ const std::vector<std::pair<std::string, std::vector<int32_t>>>&
delete_files,
+ schema::external::TStructField current_root, const TupleDescriptor*
tuple_descriptor,
+ Block* block, const cctz::time_zone* timezone, FileMetaCache*
meta_cache) {
+ schema::external::TSchema current_schema;
+ current_schema.__set_schema_id(100);
+ current_schema.__set_root_field(std::move(current_root));
+ TFileScanRangeParams scan_params;
+
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ scan_params.__set_file_type(TFileType::FILE_LOCAL);
+ scan_params.__set_format_type(file_format);
+ scan_params.__set_current_schema_id(100);
+ scan_params.__set_history_schema_info({current_schema});
+
+ std::vector<TIcebergDeleteFileDesc> delete_descriptors;
+ for (const auto& [delete_file, field_ids] : delete_files) {
+ TIcebergDeleteFileDesc delete_descriptor;
+ delete_descriptor.__set_content(2);
+ delete_descriptor.__set_path(delete_file);
+ delete_descriptor.__set_field_ids(field_ids);
+ delete_descriptor.__set_file_format(file_format);
+ delete_descriptors.push_back(std::move(delete_descriptor));
+ }
+ TIcebergFileDesc iceberg_descriptor;
+ iceberg_descriptor.__set_format_version(2);
+ iceberg_descriptor.__set_original_file_path(data_file);
+ iceberg_descriptor.__set_delete_files(std::move(delete_descriptors));
+ TTableFormatFileDesc table_format_descriptor;
+
table_format_descriptor.__set_iceberg_params(std::move(iceberg_descriptor));
+
+ TFileRangeDesc scan_range;
+ scan_range.__set_fs_name("");
+ scan_range.__set_path(data_file);
+ scan_range.__set_start_offset(0);
+
scan_range.__set_size(static_cast<int64_t>(std::filesystem::file_size(data_file)));
+
scan_range.__set_file_size(static_cast<int64_t>(std::filesystem::file_size(data_file)));
+ scan_range.__set_table_format_params(std::move(table_format_descriptor));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()};
+ runtime_state.set_timezone("UTC");
+ io::IOContext io_ctx;
+ ShardedKVCache kv_cache(8);
+ std::vector<ColumnDescriptor> column_descriptors(block->columns());
+ std::unordered_map<std::string, uint32_t> block_positions;
+ for (uint32_t position = 0; position < block->columns(); ++position) {
+ column_descriptors[position].name =
block->get_by_position(position).name;
+ block_positions.emplace(column_descriptors[position].name, position);
+ }
+ size_t read_rows = 0;
+ bool eof = false;
+ if (file_format == TFileFormatType::FORMAT_PARQUET) {
+ IcebergParquetReader reader(&kv_cache, &profile, scan_params,
scan_range, 1024, timezone,
+ &io_ctx, &runtime_state, meta_cache);
+ io::FileReaderSPtr file_reader;
+ RETURN_IF_ERROR(io::global_local_filesystem()->open_file(data_file,
&file_reader));
+ reader.set_file_reader(file_reader);
+ ParquetInitContext context;
+ context.column_descs = &column_descriptors;
+ context.col_name_to_block_idx = &block_positions;
+ context.tuple_descriptor = tuple_descriptor;
+ context.params = &scan_params;
+ context.range = &scan_range;
+ RETURN_IF_ERROR(reader.init_reader(&context));
+ return reader.get_next_block(block, &read_rows, &eof);
+ }
+ IcebergOrcReader reader(&kv_cache, &profile, &runtime_state, scan_params,
scan_range, 1024,
+ "UTC", &io_ctx, meta_cache);
+ OrcInitContext context;
+ context.column_descs = &column_descriptors;
+ context.col_name_to_block_idx = &block_positions;
+ context.tuple_descriptor = tuple_descriptor;
+ context.params = &scan_params;
+ context.range = &scan_range;
+ RETURN_IF_ERROR(reader.init_reader(&context));
+ return reader.get_next_block(block, &read_rows, &eof);
+}
+
+// Iceberg allows promoting int to long, and an equality delete keeps the key
type it was written
+// with. Data `id, v, w` holds (1,10,100), (2,20,200) and (3,30,300) with `id`
still int32 in the
+// file, the table type of `id` is now long, a delete written before the
promotion holds id = 2 as
+// INT and one written after it id = 3 as BIGINT. Both are compared as BIGINT,
the type the data is
+// read in. `project_key` selects `SELECT id` (the key is the projected
column) or `SELECT v` (the
+// key is read into a hidden carrier).
+//
NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
+void check_equality_deletes_match_promoted_key(TFileFormatType::type
file_format, bool project_key,
+ bool delete_after_promotion,
+ const cctz::time_zone* timezone,
+ FileMetaCache* meta_cache) {
+ const bool is_parquet = file_format == TFileFormatType::FORMAT_PARQUET;
+ const std::string format_name = is_parquet ? "parquet" : "orc";
+ const std::string case_name = format_name + (project_key ? "_id" : "_v") +
+ (delete_after_promotion ? "_both_deletes" :
"_int_delete");
+ SCOPED_TRACE(case_name);
+ const auto test_dir = std::filesystem::temp_directory_path() /
+ ("doris_v1_equality_delete_promoted_key_" +
case_name);
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto data_file = (test_dir / ("data." + format_name)).string();
+ const auto int_delete_file = (test_dir / ("int-delete." +
format_name)).string();
+ const auto bigint_delete_file = (test_dir / ("bigint-delete." +
format_name)).string();
+ if (is_parquet) {
+ write_iceberg_three_int_parquet_file(data_file, "id", 0, {1, 2, 3},
"v", 1, {10, 20, 30},
+ "w", 2, {100, 200, 300});
+ write_iceberg_single_integer_parquet_file(int_delete_file, "id", 0,
false, {2});
+ write_iceberg_single_integer_parquet_file(bigint_delete_file, "id", 0,
true, {3});
+ } else {
+ write_iceberg_three_int_orc_file(data_file, "id", 0, {1, 2, 3}, "v",
1, {10, 20, 30}, "w",
+ 2, {100, 200, 300});
+ write_iceberg_int_orc_file(int_delete_file, "id", 0, {2});
+ write_iceberg_int_orc_file(bigint_delete_file, "id", 0, {3}, "bigint");
+ }
+ std::vector<std::pair<std::string, std::vector<int32_t>>> delete_files = {
+ {int_delete_file, {0}}};
+ if (delete_after_promotion) {
+ delete_files.emplace_back(bigint_delete_file, std::vector<int32_t>
{0});
+ }
+
+ schema::external::TStructField current_root;
+ current_root.__set_fields({make_external_bigint_field("id", 0),
+ make_external_int_field("v", 1, std::nullopt),
+ make_external_int_field("w", 2, std::nullopt)});
+ ObjectPool object_pool;
+ DescriptorTbl* descriptor_table = nullptr;
+ const TupleDescriptor* tuple_descriptor = nullptr;
+ ASSERT_TRUE(create_single_scalar_tuple_descriptor(
+ &object_pool, project_key ? "id" : "v", project_key ?
0 : 1,
+ project_key ? TPrimitiveType::BIGINT :
TPrimitiveType::INT,
+ &descriptor_table, &tuple_descriptor)
+ .ok());
+ const DataTypePtr type = project_key ?
make_nullable(std::make_shared<DataTypeInt64>())
+ :
make_nullable(std::make_shared<DataTypeInt32>());
+ Block block;
+ block.insert({type->create_column(), type, project_key ? "id" : "v"});
+ const Status status = read_v1_with_equality_deletes(file_format,
data_file, delete_files,
+
std::move(current_root), tuple_descriptor,
+ &block, timezone,
meta_cache);
+
+ ASSERT_TRUE(status.ok()) << status;
+ std::vector<std::string> expected = project_key ? std::vector<std::string>
{"1", "3"}
+ : std::vector<std::string>
{"10", "30"};
+ if (delete_after_promotion) {
+ expected.pop_back();
+ }
+ ASSERT_EQ(block.rows(), expected.size());
+ for (size_t row = 0; row < expected.size(); ++row) {
+ EXPECT_EQ(type->to_string(*block.get_by_position(0).column, row),
expected[row]);
+ }
+ std::filesystem::remove_all(test_dir);
+}
+
+TEST_F(IcebergReaderTest,
v1_parquet_equality_delete_matches_key_after_int_to_long_promotion) {
+ check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_PARQUET,
true, false,
+ &timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest,
v1_orc_equality_delete_matches_key_after_int_to_long_promotion) {
+ check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_ORC,
true, false,
+ &timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest,
v1_parquet_equality_deletes_before_and_after_int_to_long_promotion) {
+ check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_PARQUET,
true, true,
+ &timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest,
v1_orc_equality_deletes_before_and_after_int_to_long_promotion) {
+ check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_ORC,
true, true,
+ &timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest,
+
v1_parquet_hidden_equality_deletes_before_and_after_int_to_long_promotion) {
+ check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_PARQUET,
false, true,
+ &timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest,
v1_orc_hidden_equality_deletes_before_and_after_int_to_long_promotion) {
+ check_equality_deletes_match_promoted_key(TFileFormatType::FORMAT_ORC,
false, true,
+ &timezone_obj, cache.get());
+}
+
+// A nested key promoted from int to long: data `id, payload STRUCT<existing>`
holds (1,{10}),
+// (2,{20}) and (3,{30}) with payload.existing still int32 in the file, and a
delete written after
+// the promotion holds payload.existing = 20 as BIGINT. The query projects
only `id`, so the key is
+// extracted from the physical payload root and must be widened before it is
compared; copying the
+// INT values into a BIGINT column read the wrong bytes.
+void check_nested_equality_delete_after_promotion_reads_older_file(
+ TFileFormatType::type file_format, const cctz::time_zone* timezone,
+ FileMetaCache* meta_cache) {
+ const bool is_parquet = file_format == TFileFormatType::FORMAT_PARQUET;
+ const std::string format_name = is_parquet ? "parquet" : "orc";
+ SCOPED_TRACE(format_name);
+ const auto test_dir = std::filesystem::temp_directory_path() /
+ ("doris_v1_nested_equality_delete_promoted_key_" +
format_name);
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto data_file = (test_dir / ("data." + format_name)).string();
+ const auto delete_file = (test_dir / ("equality-delete." +
format_name)).string();
+ if (is_parquet) {
+ write_iceberg_id_and_nested_struct_parquet_file(data_file, {1, 2, 3},
{10, 20, 30},
+ {false, false, false});
+ write_iceberg_bigint_nested_struct_parquet_file(delete_file, {20});
+ } else {
+ write_iceberg_id_and_nested_struct_orc_file(data_file, {1, 2, 3}, {10,
20, 30},
+ {false, false, false});
+ write_iceberg_nested_struct_orc_file(delete_file, {20}, {false},
"existing", 2, "bigint");
+ }
+
+ schema::external::TStructField current_root;
+ current_root.__set_fields({make_external_int_field("id", 0, std::nullopt),
+ make_external_struct_field(
+ "payload", 1,
{make_external_bigint_field("existing", 2)})});
+ ObjectPool object_pool;
+ DescriptorTbl* descriptor_table = nullptr;
+ const TupleDescriptor* tuple_descriptor = nullptr;
+ ASSERT_TRUE(create_single_int_tuple_descriptor(&object_pool, "id", 0,
&descriptor_table,
+ &tuple_descriptor)
+ .ok());
+ const auto int_type = make_nullable(std::make_shared<DataTypeInt32>());
+ Block block;
+ block.insert({int_type->create_column(), int_type, "id"});
+ const Status status = read_v1_with_equality_deletes(
+ file_format, data_file, {{delete_file, {2}}},
std::move(current_root), tuple_descriptor,
+ &block, timezone, meta_cache);
+
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(block.rows(), 2);
+ EXPECT_EQ(int_type->to_string(*block.get_by_position(0).column, 0), "1");
+ EXPECT_EQ(int_type->to_string(*block.get_by_position(0).column, 1), "3");
+ std::filesystem::remove_all(test_dir);
+}
+
+TEST_F(IcebergReaderTest,
v1_parquet_nested_equality_delete_after_promotion_reads_older_file) {
+
check_nested_equality_delete_after_promotion_reads_older_file(TFileFormatType::FORMAT_PARQUET,
+
&timezone_obj, cache.get());
+}
+
+TEST_F(IcebergReaderTest,
v1_orc_nested_equality_delete_after_promotion_reads_older_file) {
+
check_nested_equality_delete_after_promotion_reads_older_file(TFileFormatType::FORMAT_ORC,
+
&timezone_obj, cache.get());
+}
+
+// The key is compared in the promoted type, not in the delete file's older
one. A data file written
+// after `id` became long holds NULL, 0, 1 and 4294967296, and a delete
written before the promotion
+// holds the NULL key as INT. Narrowed to INT, 4294967296 would become NULL
and be deleted by that
+// key; compared as BIGINT, only the NULL row is deleted.
+TEST_F(IcebergReaderTest,
v1_parquet_equality_delete_keeps_out_of_range_promoted_value) {
+ const auto test_dir = std::filesystem::temp_directory_path() /
+
"doris_v1_equality_delete_out_of_range_promoted_value";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+ const auto data_file = (test_dir / "data.parquet").string();
+ const auto delete_file = (test_dir / "equality-delete.parquet").string();
+ write_iceberg_single_integer_parquet_file(data_file, "id", 0, true,
+ {std::nullopt, 0, 1,
4294967296});
+ write_iceberg_single_integer_parquet_file(delete_file, "id", 0, false,
{std::nullopt});
+
+ schema::external::TStructField current_root;
+ current_root.__set_fields({make_external_bigint_field("id", 0)});
+ ObjectPool object_pool;
+ DescriptorTbl* descriptor_table = nullptr;
+ const TupleDescriptor* tuple_descriptor = nullptr;
+ ASSERT_TRUE(create_single_scalar_tuple_descriptor(&object_pool, "id", 0,
TPrimitiveType::BIGINT,
+ &descriptor_table,
&tuple_descriptor)
+ .ok());
+ const auto bigint_type = make_nullable(std::make_shared<DataTypeInt64>());
+ Block block;
+ block.insert({bigint_type->create_column(), bigint_type, "id"});
+ const Status status = read_v1_with_equality_deletes(
+ TFileFormatType::FORMAT_PARQUET, data_file, {{delete_file, {0}}},
+ std::move(current_root), tuple_descriptor, &block, &timezone_obj,
cache.get());
+
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(block.rows(), 3);
+ EXPECT_EQ(bigint_type->to_string(*block.get_by_position(0).column, 0),
"0");
+ EXPECT_EQ(bigint_type->to_string(*block.get_by_position(0).column, 1),
"1");
+ EXPECT_EQ(bigint_type->to_string(*block.get_by_position(0).column, 2),
"4294967296");
+ std::filesystem::remove_all(test_dir);
+}
+
// Data `id INT, payload STRUCT<existing INT, k INT>` holds (1,{10,100}),
(2,{20,200}) and
// (3,{30,300}). An equality delete on `delete_field_ids` (2 =
payload.existing, 3 = payload.k)
// with the row {20,200} removes id 2. `projected_children` lists the payload
children the query
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
