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 d8fd23f7f38 [fix](iceberg) Allow optional position delete fields
without nulls (#65401)
d8fd23f7f38 is described below
commit d8fd23f7f3868a2031f0c38aa8231e7dddcb132d
Author: Gabriel <[email protected]>
AuthorDate: Thu Jul 9 22:35:42 2026 +0800
[fix](iceberg) Allow optional position delete fields without nulls (#65401)
Relax Iceberg position delete Parquet reads so Doris can consume delete
files where `file_path` or `pos` are declared optional in Parquet
metadata but contain no actual null values.
The reader now materializes position delete columns as nullable,
validates that each batch has no nulls, then strips the nullable wrapper
before building delete ranges. Actual null values in `file_path` or
`pos` still fail with a corruption error.
This covers the legacy Iceberg Parquet reader, the shared delete-file
helper used by Iceberg writes, and the format_v2 position delete
collector.
## Problem
Some writers, including DuckLake/DuckDB-produced data seen in the field,
can produce Iceberg position delete files whose required Iceberg fields
are encoded as optional Parquet fields. Doris previously rejected these
files before checking whether the data actually contained nulls.
---
.../table/iceberg_delete_file_reader_helper.cpp | 40 +++--
be/src/format/table/iceberg_reader.cpp | 25 +--
be/src/format/table/iceberg_reader_mixin.h | 23 ++-
be/src/format_v2/table/iceberg_reader.cpp | 18 ++-
.../iceberg_delete_file_reader_helper_test.cpp | 121 ++++++++++++++
be/test/format_v2/table/iceberg_reader_test.cpp | 179 +++++++++++++++++++++
6 files changed, 378 insertions(+), 28 deletions(-)
diff --git a/be/src/format/table/iceberg_delete_file_reader_helper.cpp
b/be/src/format/table/iceberg_delete_file_reader_helper.cpp
index 21bf9865868..50738587d8b 100644
--- a/be/src/format/table/iceberg_delete_file_reader_helper.cpp
+++ b/be/src/format/table/iceberg_delete_file_reader_helper.cpp
@@ -30,8 +30,10 @@
#include "core/block/block.h"
#include "core/block/column_with_type_and_name.h"
#include "core/column/column_dictionary.h"
+#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_vector.h"
+#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
#include "exec/common/endian.h"
@@ -90,9 +92,26 @@ Status visit_position_delete_block(const Block& block,
size_t read_rows,
return Status::InternalError("Position delete block is missing
required columns");
}
- const auto* pos_column =
- assert_cast<const
ColumnInt64*>(block.get_by_position(pos_it->second).column.get());
- const auto* path_column =
block.get_by_position(path_it->second).column.get();
+ ColumnPtr path_column_ptr = block.get_by_position(path_it->second).column;
+ ColumnPtr pos_column_ptr = block.get_by_position(pos_it->second).column;
+ if (const auto* nullable_col =
check_and_get_column<ColumnNullable>(*path_column_ptr);
+ nullable_col != nullptr) {
+ if (nullable_col->has_null(0, read_rows)) {
+ return Status::Corruption(
+ "Iceberg position delete column file_path contains null
values");
+ }
+ path_column_ptr = remove_nullable(path_column_ptr);
+ }
+ if (const auto* nullable_col =
check_and_get_column<ColumnNullable>(*pos_column_ptr);
+ nullable_col != nullptr) {
+ if (nullable_col->has_null(0, read_rows)) {
+ return Status::Corruption("Iceberg position delete column pos
contains null values");
+ }
+ pos_column_ptr = remove_nullable(pos_column_ptr);
+ }
+
+ const auto* pos_column = assert_cast<const
ColumnInt64*>(pos_column_ptr.get());
+ const auto* path_column = path_column_ptr.get();
if (const auto* string_column =
check_and_get_column<ColumnString>(path_column);
string_column != nullptr) {
@@ -252,16 +271,15 @@ Status read_iceberg_position_delete_file(const
TIcebergDeleteFileDesc& delete_fi
while (!eof) {
Block block;
if (dictionary_coded) {
- block.insert(ColumnWithTypeAndName(ColumnDictI32::create(),
-
std::make_shared<DataTypeString>(),
- ICEBERG_FILE_PATH));
+ block.insert(ColumnWithTypeAndName(
+ ColumnNullable::create(ColumnDictI32::create(),
ColumnUInt8::create()),
+ make_nullable(std::make_shared<DataTypeString>()),
ICEBERG_FILE_PATH));
} else {
- block.insert(ColumnWithTypeAndName(ColumnString::create(),
-
std::make_shared<DataTypeString>(),
- ICEBERG_FILE_PATH));
+ block.insert(ColumnWithTypeAndName(
+ make_nullable(std::make_shared<DataTypeString>()),
ICEBERG_FILE_PATH));
}
- block.insert(ColumnWithTypeAndName(ColumnInt64::create(),
-
std::make_shared<DataTypeInt64>(), ICEBERG_ROW_POS));
+
block.insert(ColumnWithTypeAndName(make_nullable(std::make_shared<DataTypeInt64>()),
+ ICEBERG_ROW_POS));
size_t read_rows = 0;
RETURN_IF_ERROR(reader.get_next_block(&block, &read_rows, &eof));
RETURN_IF_ERROR(visit_position_delete_block(block, read_rows,
visitor));
diff --git a/be/src/format/table/iceberg_reader.cpp
b/be/src/format/table/iceberg_reader.cpp
index 6f6d0b9e9d5..110a10f74c0 100644
--- a/be/src/format/table/iceberg_reader.cpp
+++ b/be/src/format/table/iceberg_reader.cpp
@@ -37,9 +37,11 @@
#include "core/block/block.h"
#include "core/block/column_with_type_and_name.h"
#include "core/column/column.h"
+#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_vector.h"
#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_nullable.h"
#include "core/data_type/define_primitive_type.h"
#include "core/data_type/primitive_type.h"
#include "core/string_ref.h"
@@ -386,23 +388,26 @@ Status
IcebergParquetReader::_read_position_delete_file(const TFileRangeDesc* de
break;
}
}
- DataTypePtr data_type_file_path {new DataTypeString};
- DataTypePtr data_type_pos {new DataTypeInt64};
+ DataTypePtr data_type_file_path =
make_nullable(std::make_shared<DataTypeString>());
+ DataTypePtr data_type_pos =
make_nullable(std::make_shared<DataTypeInt64>());
bool eof = false;
while (!eof) {
- Block block = {dictionary_coded
- ? ColumnWithTypeAndName
{ColumnDictI32::create(),
- data_type_file_path,
ICEBERG_FILE_PATH}
- : ColumnWithTypeAndName {data_type_file_path,
ICEBERG_FILE_PATH},
-
- {data_type_pos, ICEBERG_ROW_POS}};
+ Block block = {
+ dictionary_coded
+ ? ColumnWithTypeAndName
{ColumnNullable::create(ColumnDictI32::create(),
+
ColumnUInt8::create()),
+ data_type_file_path,
ICEBERG_FILE_PATH}
+ : ColumnWithTypeAndName {data_type_file_path,
ICEBERG_FILE_PATH},
+
+ {data_type_pos, ICEBERG_ROW_POS}};
size_t read_rows = 0;
RETURN_IF_ERROR(parquet_delete_reader.get_next_block(&block,
&read_rows, &eof));
if (read_rows <= 0) {
break;
}
- _gen_position_delete_file_range(block, position_delete, read_rows,
dictionary_coded);
+ RETURN_IF_ERROR(_gen_position_delete_file_range(block,
position_delete, read_rows,
+ dictionary_coded));
}
return Status::OK();
};
@@ -650,7 +655,7 @@ Status IcebergOrcReader::_read_position_delete_file(const
TFileRangeDesc* delete
size_t read_rows = 0;
RETURN_IF_ERROR(orc_delete_reader.get_next_block(&block, &read_rows,
&eof));
- _gen_position_delete_file_range(block, position_delete, read_rows,
false);
+ RETURN_IF_ERROR(_gen_position_delete_file_range(block,
position_delete, read_rows, false));
}
return Status::OK();
}
diff --git a/be/src/format/table/iceberg_reader_mixin.h
b/be/src/format/table/iceberg_reader_mixin.h
index 2e9bd06b6fd..c0b2969678a 100644
--- a/be/src/format/table/iceberg_reader_mixin.h
+++ b/be/src/format/table/iceberg_reader_mixin.h
@@ -247,8 +247,9 @@ protected:
PositionDeleteRange _get_range(const ColumnString& file_path_column);
static void _sort_delete_rows(const std::vector<std::vector<int64_t>*>&
delete_rows_array,
int64_t num_delete_rows,
std::vector<int64_t>& result);
- void _gen_position_delete_file_range(Block& block, DeleteFile*
position_delete,
- size_t read_rows, bool
file_path_column_dictionary_coded);
+ Status _gen_position_delete_file_range(Block& block, DeleteFile*
position_delete,
+ size_t read_rows,
+ bool
file_path_column_dictionary_coded);
void _generate_equality_delete_block(Block* block,
const std::vector<std::string>&
equality_delete_col_names,
const std::vector<DataTypePtr>&
equality_delete_col_types);
@@ -776,7 +777,7 @@ void IcebergReaderMixin<BaseReader>::_sort_delete_rows(
}
template <typename BaseReader>
-void IcebergReaderMixin<BaseReader>::_gen_position_delete_file_range(
+Status IcebergReaderMixin<BaseReader>::_gen_position_delete_file_range(
Block& block, DeleteFile* position_delete, size_t read_rows,
bool file_path_column_dictionary_coded) {
SCOPED_TIMER(_iceberg_profile.parse_delete_file_time);
@@ -784,6 +785,21 @@ void
IcebergReaderMixin<BaseReader>::_gen_position_delete_file_range(
ColumnPtr path_column =
block.get_by_position(name_to_pos_map[ICEBERG_FILE_PATH]).column;
DCHECK_EQ(path_column->size(), read_rows);
ColumnPtr pos_column =
block.get_by_position(name_to_pos_map[ICEBERG_ROW_POS]).column;
+ if (const auto* nullable_col =
check_and_get_column<ColumnNullable>(*path_column);
+ nullable_col != nullptr) {
+ if (nullable_col->has_null(0, read_rows)) {
+ return Status::Corruption(
+ "Iceberg position delete column file_path contains null
values");
+ }
+ path_column = remove_nullable(path_column);
+ }
+ if (const auto* nullable_col =
check_and_get_column<ColumnNullable>(*pos_column);
+ nullable_col != nullptr) {
+ if (nullable_col->has_null(0, read_rows)) {
+ return Status::Corruption("Iceberg position delete column pos
contains null values");
+ }
+ pos_column = remove_nullable(pos_column);
+ }
using ColumnType = typename PrimitiveTypeTraits<TYPE_BIGINT>::ColumnType;
const int64_t* src_data = assert_cast<const
ColumnType&>(*pos_column).get_data().data();
PositionDeleteRange range;
@@ -810,6 +826,7 @@ void
IcebergReaderMixin<BaseReader>::_gen_position_delete_file_range(
int64_t* dest_position = &(*delete_rows)[origin_size];
memcpy(dest_position, cpy_start, cpy_count * sizeof(int64_t));
}
+ return Status::OK();
}
template <typename BaseReader>
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index eb1172ed358..7e81b860536 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -150,10 +150,20 @@ Status
IcebergTableReader::PositionDeleteRowsCollector::collect(const Block& blo
if (read_rows == 0) {
return Status::OK();
}
- const auto& file_path_column = assert_cast<const ColumnString&>(
-
*remove_nullable((block.get_by_position(ICEBERG_FILE_PATH_BLOCK_POSITION).column)));
- const auto& pos_column = assert_cast<const ColumnInt64&>(
-
*remove_nullable(block.get_by_position(ICEBERG_ROW_POS_BLOCK_POSITION).column));
+ const auto& file_path_column_ptr =
+ block.get_by_position(ICEBERG_FILE_PATH_BLOCK_POSITION).column;
+ const auto& pos_column_ptr =
block.get_by_position(ICEBERG_ROW_POS_BLOCK_POSITION).column;
+ if (const auto* nullable_column =
check_and_get_column<ColumnNullable>(*file_path_column_ptr);
+ nullable_column != nullptr && nullable_column->has_null(0, read_rows))
{
+ return Status::Corruption("Iceberg position delete column file_path
contains null values");
+ }
+ if (const auto* nullable_column =
check_and_get_column<ColumnNullable>(*pos_column_ptr);
+ nullable_column != nullptr && nullable_column->has_null(0, read_rows))
{
+ return Status::Corruption("Iceberg position delete column pos contains
null values");
+ }
+ const auto& file_path_column =
+ assert_cast<const
ColumnString&>(*remove_nullable(file_path_column_ptr));
+ const auto& pos_column = assert_cast<const
ColumnInt64&>(*remove_nullable(pos_column_ptr));
for (size_t row = 0; row < read_rows; ++row) {
const auto file_path = file_path_column.get_data_at(row).to_string();
if (file_path == _data_file_path) {
diff --git
a/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
b/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
index 34d2e60f712..cb23fe9f036 100644
--- a/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
+++ b/be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp
@@ -17,13 +17,17 @@
#include "format/table/iceberg_delete_file_reader_helper.h"
+#include <arrow/api.h>
+#include <arrow/io/api.h>
#include <gen_cpp/PlanNodes_types.h>
#include <gen_cpp/Types_types.h>
#include <gtest/gtest.h>
+#include <parquet/arrow/writer.h>
#include <cstring>
#include <filesystem>
#include <fstream>
+#include <optional>
#include <string>
#include <thread>
#include <unordered_map>
@@ -99,6 +103,45 @@ int64_t write_iceberg_deletion_vector_file(const
std::string& file_path,
return static_cast<int64_t>(blob.size());
}
+std::string temp_parquet_path(const std::string& filename) {
+ return (std::filesystem::temp_directory_path() / filename).string();
+}
+
+void write_position_delete_parquet(const std::string& path,
+ const
std::vector<std::optional<std::string>>& file_paths,
+ const std::vector<std::optional<int64_t>>&
positions) {
+ ASSERT_EQ(file_paths.size(), positions.size());
+
+ arrow::StringBuilder path_builder;
+ arrow::Int64Builder pos_builder;
+ for (size_t i = 0; i < file_paths.size(); ++i) {
+ if (file_paths[i].has_value()) {
+ ASSERT_TRUE(path_builder.Append(*file_paths[i]).ok());
+ } else {
+ ASSERT_TRUE(path_builder.AppendNull().ok());
+ }
+ if (positions[i].has_value()) {
+ ASSERT_TRUE(pos_builder.Append(*positions[i]).ok());
+ } else {
+ ASSERT_TRUE(pos_builder.AppendNull().ok());
+ }
+ }
+
+ std::shared_ptr<arrow::Array> path_array;
+ std::shared_ptr<arrow::Array> pos_array;
+ ASSERT_TRUE(path_builder.Finish(&path_array).ok());
+ ASSERT_TRUE(pos_builder.Finish(&pos_array).ok());
+
+ auto schema = arrow::schema({arrow::field("file_path", arrow::utf8(),
true),
+ arrow::field("pos", arrow::int64(), true)});
+ auto table = arrow::Table::Make(schema, {path_array, pos_array});
+
+ std::shared_ptr<arrow::io::FileOutputStream> output;
+ ASSERT_TRUE(arrow::io::FileOutputStream::Open(path).Value(&output).ok());
+ PARQUET_THROW_NOT_OK(
+ parquet::arrow::WriteTable(*table, arrow::default_memory_pool(),
output, 1024));
+}
+
IcebergDeleteFileReaderOptions make_delete_file_reader_options(
RuntimeState* state, RuntimeProfile* profile, const
TFileScanRangeParams* scan_params,
io::IOContext* io_ctx) {
@@ -110,6 +153,21 @@ IcebergDeleteFileReaderOptions
make_delete_file_reader_options(
};
}
+IcebergDeleteFileReaderOptions delete_reader_options(RuntimeState*
runtime_state,
+ RuntimeProfile* profile,
+ TFileScanRangeParams*
scan_params,
+
IcebergDeleteFileIOContext* io_context,
+ FileMetaCache*
meta_cache) {
+ IcebergDeleteFileReaderOptions options;
+ options.state = runtime_state;
+ options.profile = profile;
+ options.scan_params = scan_params;
+ options.io_ctx = &io_context->io_ctx;
+ options.meta_cache = meta_cache;
+ options.batch_size = 1024;
+ return options;
+}
+
} // namespace
TEST(IcebergDeleteFileReaderHelperTest, BuildDeleteFileRange) {
@@ -397,4 +455,67 @@ TEST(IcebergDeleteFileReaderHelperTest,
ReadMixedEncodingParquetPositionDeleteFi
EXPECT_EQ(it->second, expected_positions);
}
+TEST(IcebergDeleteFileReaderHelperTest,
ReadOptionalParquetPositionDeleteColumnsWithoutNulls) {
+ const auto delete_file_path =
+
temp_parquet_path("iceberg_optional_position_delete_without_nulls.parquet");
+ write_position_delete_parquet(delete_file_path,
+ {std::string(kTargetDataFilePath),
+ std::string(kTargetDataFilePath),
"s3://other/file.parquet"},
+ {3, 9, 11});
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState runtime_state((TQueryOptions()), TQueryGlobals());
+ FileMetaCache meta_cache(1024);
+ IcebergDeleteFileIOContext io_context(&runtime_state);
+
+ TFileScanRangeParams scan_params;
+ scan_params.file_type = TFileType::FILE_LOCAL;
+ scan_params.format_type = TFileFormatType::FORMAT_PARQUET;
+
+ TIcebergDeleteFileDesc delete_file;
+ delete_file.path = delete_file_path;
+ delete_file.file_format = TFileFormatType::FORMAT_PARQUET;
+ delete_file.__isset.file_format = true;
+
+ CollectPositionDeleteVisitor visitor;
+ auto options =
+ delete_reader_options(&runtime_state, &profile, &scan_params,
&io_context, &meta_cache);
+ auto st = read_iceberg_position_delete_file(delete_file, options,
&visitor);
+
+ std::filesystem::remove(delete_file_path);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(visitor.total_rows, 3);
+ EXPECT_EQ(visitor.delete_rows[kTargetDataFilePath],
std::vector<int64_t>({3, 9}));
+}
+
+TEST(IcebergDeleteFileReaderHelperTest,
RejectParquetPositionDeleteColumnsWithActualNulls) {
+ const auto delete_file_path =
temp_parquet_path("iceberg_position_delete_with_nulls.parquet");
+ write_position_delete_parquet(delete_file_path,
+ {std::string(kTargetDataFilePath),
std::nullopt}, {3, 9});
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState runtime_state((TQueryOptions()), TQueryGlobals());
+ FileMetaCache meta_cache(1024);
+ IcebergDeleteFileIOContext io_context(&runtime_state);
+
+ TFileScanRangeParams scan_params;
+ scan_params.file_type = TFileType::FILE_LOCAL;
+ scan_params.format_type = TFileFormatType::FORMAT_PARQUET;
+
+ TIcebergDeleteFileDesc delete_file;
+ delete_file.path = delete_file_path;
+ delete_file.file_format = TFileFormatType::FORMAT_PARQUET;
+ delete_file.__isset.file_format = true;
+
+ CollectPositionDeleteVisitor visitor;
+ auto options =
+ delete_reader_options(&runtime_state, &profile, &scan_params,
&io_context, &meta_cache);
+ auto st = read_iceberg_position_delete_file(delete_file, options,
&visitor);
+
+ std::filesystem::remove(delete_file_path);
+ ASSERT_FALSE(st.ok());
+ EXPECT_NE(st.to_string().find("file_path contains null values"),
std::string::npos)
+ << st.to_string();
+}
+
} // namespace doris
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index 2d3f74963d9..8a10fc06a41 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -260,6 +260,19 @@ std::shared_ptr<arrow::Array> build_nullable_int64_array(
return finish_array(&builder);
}
+std::shared_ptr<arrow::Array> build_nullable_string_array(
+ const std::vector<std::optional<std::string>>& values) {
+ arrow::StringBuilder builder;
+ 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_string_array(const
std::vector<std::string>& values) {
arrow::StringBuilder builder;
for (const auto& value : values) {
@@ -405,6 +418,31 @@ void write_position_delete_parquet_file(const std::string&
file_path,
builder.build()));
}
+void write_nullable_position_delete_parquet_file(
+ const std::string& file_path,
+ const std::vector<std::optional<std::string>>& data_file_paths,
+ const std::vector<std::optional<int64_t>>& positions) {
+ ASSERT_EQ(data_file_paths.size(), positions.size());
+ auto schema = arrow::schema({
+ arrow::field("file_path", arrow::utf8(), true),
+ arrow::field("pos", arrow::int64(), true),
+ });
+ auto table = arrow::Table::Make(schema,
{build_nullable_string_array(data_file_paths),
+
build_nullable_int64_array(positions)});
+
+ 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,
+
static_cast<int64_t>(positions.size()),
+ builder.build()));
+}
+
int64_t write_iceberg_deletion_vector_file(const std::string& file_path,
const std::vector<uint64_t>&
deleted_positions) {
roaring::Roaring64Map rows;
@@ -1980,6 +2018,147 @@ TEST(IcebergV2ReaderTest,
IcebergTableReaderAppliesPositionDeleteFile) {
std::filesystem::remove_all(test_dir);
}
+TEST(IcebergV2ReaderTest,
IcebergTableReaderAppliesNullablePositionDeleteFileWithoutNulls) {
+ const auto test_dir = std::filesystem::temp_directory_path() /
+ "doris_iceberg_nullable_position_delete_file_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 /
"position-delete.parquet").string();
+ write_int_pair_parquet_file(file_path, {1, 2, 3, 4, 5}, {10, 20, 30, 40,
50},
+ {"one", "two", "three", "four", "five"});
+ write_nullable_position_delete_parquet_file(delete_file_path, {file_path,
file_path},
+ {int64_t {1}, int64_t {3}});
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+ 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;
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = io_ctx,
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ })
+ .ok());
+
+ 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_position_delete_file(delete_file_path)}));
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+
+ EXPECT_EQ(read_iceberg_ids(&reader, projected_columns),
std::vector<int32_t>({1, 3, 5}));
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
+TEST(IcebergV2ReaderTest,
IcebergTableReaderRejectsNullablePositionDeleteFileWithActualNulls) {
+ const auto test_dir = std::filesystem::temp_directory_path() /
+
"doris_iceberg_nullable_position_delete_actual_null_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 /
"position-delete.parquet").string();
+ write_int_pair_parquet_file(file_path, {1, 2, 3}, {10, 20, 30}, {"one",
"two", "three"});
+ write_nullable_position_delete_parquet_file(delete_file_path, {file_path,
std::nullopt},
+ {int64_t {1}, int64_t {2}});
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+ 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;
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = io_ctx,
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ })
+ .ok());
+
+ 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_position_delete_file(delete_file_path)}));
+ auto status = reader.prepare_split(split_options);
+ ASSERT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("file_path contains null values"),
std::string::npos)
+ << status.to_string();
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
+TEST(IcebergV2ReaderTest,
IcebergTableReaderRejectsNullablePositionDeletePosWithActualNulls) {
+ const auto test_dir = std::filesystem::temp_directory_path() /
+
"doris_iceberg_nullable_position_delete_pos_null_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 /
"position-delete.parquet").string();
+ write_int_pair_parquet_file(file_path, {1, 2, 3}, {10, 20, 30}, {"one",
"two", "three"});
+ write_nullable_position_delete_parquet_file(delete_file_path, {file_path,
file_path},
+ {int64_t {1}, std::nullopt});
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+
+ RuntimeProfile profile("test_profile");
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+ 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;
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = io_ctx,
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ })
+ .ok());
+
+ 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_position_delete_file(delete_file_path)}));
+ auto status = reader.prepare_split(split_options);
+ ASSERT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("pos contains null values"),
std::string::npos)
+ << status.to_string();
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
TEST(IcebergV2ReaderTest,
IcebergTableReaderIgnoresPositionDeleteFilesWhenDeletionVectorPresent) {
const auto test_dir =
std::filesystem::temp_directory_path() /
"doris_iceberg_delete_files_merge_test";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
