github-actions[bot] commented on code in PR #69723:
URL: https://github.com/apache/doris/pull/69723#discussion_r4422958987
##########
be/src/format/table/iceberg_delete_file_reader_helper.cpp:
##########
@@ -285,54 +238,43 @@ Status read_iceberg_position_delete_file(const
TIcebergDeleteFileDesc& delete_fi
TFileFormatType::type file_format;
RETURN_IF_ERROR(validate_position_delete_file_format(delete_file,
&file_format));
- if (file_format == TFileFormatType::FORMAT_PARQUET) {
- ParquetReader reader(options.profile, *options.scan_params,
delete_range,
- options.batch_size,
&options.state->timezone_obj(), options.io_ctx,
- options.state, options.meta_cache);
- bool dictionary_coded = false;
- RETURN_IF_ERROR(init_parquet_delete_reader(&reader,
&dictionary_coded));
-
- bool eof = false;
- while (!eof) {
- Block block;
- if (dictionary_coded) {
- block.insert(ColumnWithTypeAndName(
- ColumnNullable::create(ColumnDictI32::create(),
ColumnUInt8::create()),
- make_nullable(std::make_shared<DataTypeString>()),
ICEBERG_FILE_PATH));
- } else {
- block.insert(ColumnWithTypeAndName(
- make_nullable(std::make_shared<DataTypeString>()),
ICEBERG_FILE_PATH));
- }
-
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));
- }
- return Status::OK();
- }
-
- if (file_format == TFileFormatType::FORMAT_ORC) {
- OrcReader reader(options.profile, options.state, *options.scan_params,
delete_range,
- options.batch_size, options.state->timezone(),
options.io_ctx,
- options.meta_cache);
- RETURN_IF_ERROR(init_orc_delete_reader(&reader));
-
- bool eof = false;
- while (!eof) {
- Block block;
- block.insert(ColumnWithTypeAndName(
- ColumnString::create(),
std::make_shared<DataTypeString>(), ICEBERG_FILE_PATH));
- block.insert(ColumnWithTypeAndName(ColumnInt64::create(),
-
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));
- }
- return Status::OK();
+ const auto physical_format = file_format == TFileFormatType::FORMAT_PARQUET
+ ? format::FileFormat::PARQUET
+ : format::FileFormat::ORC;
+ auto params = *options.scan_params;
+ params.__set_format_type(file_format);
+ // Delete files have their own schema and row numbering; data-file
predicates and table
+ // wrappers must not filter delete records before the visitor resolves
their target files.
+ std::vector<format::ColumnDefinition> columns(2);
+ columns[0].identifier =
Field::create_field<TYPE_STRING>(ICEBERG_FILE_PATH);
+ columns[0].name = ICEBERG_FILE_PATH;
+ columns[0].type = make_nullable(std::make_shared<DataTypeString>());
+ columns[1].identifier = Field::create_field<TYPE_STRING>(ICEBERG_ROW_POS);
+ columns[1].name = ICEBERG_ROW_POS;
+ columns[1].type = make_nullable(std::make_shared<DataTypeInt64>());
+ auto io_ctx = std::shared_ptr<io::IOContext>(options.io_ctx,
[](io::IOContext*) {});
+ format::TableReader reader;
+ RETURN_IF_ERROR(reader.init({.projected_columns = columns,
+ .conjuncts = {},
+ .format = physical_format,
+ .scan_params = ¶ms,
+ .io_ctx = std::move(io_ctx),
+ .runtime_state = options.state,
+ .scanner_profile = options.profile}));
+ reader.set_batch_size(options.batch_size);
Review Comment:
[P1] Clamp position-delete batches to the native Parquet limit
`load_rewritable_delete_rows()` in `viceberg_delete_sink.cpp` passes
`options.batch_size = 102400` when an Iceberg v3 DELETE/MERGE merges existing
position deletes. This adapter forwards it unchanged to the V2 Parquet
scheduler, whose first physical batch must satisfy `batch_rows <= UINT16_MAX`.
A valid Parquet position-delete Row Group with at least 65,536 rows therefore
raises a FatalError and fails the operation; the former standalone reader
accepted this cap. Clamp the adapter/native batch size and cover a 65,536+ row
Row Group with the production 102400 option, checking every visited position.
##########
be/src/exec/operator/file_scan_operator.cpp:
##########
@@ -245,41 +138,12 @@ Status
FileScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
scan_params = _split_source->get_params();
}
RETURN_IF_ERROR(validate_file_scan_support(*scan_params));
- const bool is_load =
-
state()->desc_tbl().get_tuple_descriptor(scan_params->src_tuple_id) != nullptr;
- // TODO: Use scanner v2 for all queries.
- const bool use_file_scanner_v2 =
- should_use_file_scanner_v2(state()->query_options(), is_load,
*scan_params);
- _operator_profile->add_info_string("UseScannerV2", use_file_scanner_v2 ?
"true" : "false");
- const auto* output_tuple_desc =
state()->desc_tbl().get_tuple_descriptor(_output_tuple_id);
- DORIS_CHECK(output_tuple_desc != nullptr);
- const bool metadata_only_count =
- is_count_star_pushdown() &&
_split_source->all_ranges_have_table_level_row_count();
- if (!is_load && !use_file_scanner_v2 && !metadata_only_count &&
- std::ranges::any_of(output_tuple_desc->slots(), [](const
SlotDescriptor* slot) {
- return contains_variant_type(slot->type());
- })) {
- // A syntactic COUNT(*) alone is insufficient: every assigned range
must prove that the
- // legacy scanner will emit metadata counts without decoding a Variant
carrier.
- return Status::NotSupported(
- "External VARIANT columns require FileScannerV2; the legacy
file scanner does "
- "not support VARIANT");
- }
- if (use_file_scanner_v2) {
- _parquet_profile_cache =
-
std::make_unique<format::parquet::ParquetProfileCache>(_scanner_profile.get());
- }
+ _parquet_profile_cache =
+
std::make_unique<format::parquet::ParquetProfileCache>(_scanner_profile.get());
for (int i = 0; i < _max_scanners; ++i) {
- ScannerSPtr scanner;
- if (use_file_scanner_v2) {
- scanner = FileScannerV2::create_shared(state(), this, p._limit,
_split_source,
- _scanner_profile.get(),
p._kv_cache.get(),
- &p._colname_to_slot_id);
- } else {
- scanner = FileScanner::create_shared(state(), this, p._limit,
_split_source,
- _scanner_profile.get(),
p._kv_cache.get(),
- &p._colname_to_slot_id);
- }
+ ScannerSPtr scanner = FileScannerV2::create_shared(
+ state(), this, p._limit, _split_source,
_scanner_profile.get(), p._kv_cache.get(),
Review Comment:
[P2] Preserve ORC footer caching on the unconditional V2 path
With the metadata cache enabled, V1 passed it into OrcReader and reused
`setSerializedFileTail()` on warm opens. V2 `OrcReader::init()` always creates
fresh SDK ReaderOptions and calls `createReader()` without a cached tail;
neither TableReader nor its input stream replaces that shared cache. Repeated
scans of 100 unchanged S3 ORC files (larger than the in-memory threshold, byte
FileCache disabled, default metadata capacity 1000) now repeat footer network
reads that V1 avoided. Preserve stable-identity ORC tail caching and cold/warm
read-count coverage before removing the cached path.
##########
be/src/format_v2/orc/orc_reader.cpp:
##########
@@ -2127,6 +2332,33 @@ Status
OrcReader::_filter_orc_batch(::orc::ColumnVectorBatch& data, uint16_t* se
_state->root_type->getFieldName(file_column_id.value())});
}
+ return Status::OK();
+}
+
+Status OrcReader::_filter_orc_batch(::orc::ColumnVectorBatch& data, uint16_t*
sel, uint16_t size,
+ void* /*arg*/) {
+ if (!_state->orc_lazy_read_enabled || sel == nullptr || size == 0) {
+ data.numElements = size;
+ return Status::OK();
+ }
+ if (size > MAX_ORC_READ_BATCH_SIZE) {
+ return Status::InvalidArgument("ORC lazy filter batch size {} exceeds
{}", size,
+ MAX_ORC_READ_BATCH_SIZE);
+ }
+
+ auto* struct_batch = dynamic_cast<::orc::StructVectorBatch*>(&data);
+ if (struct_batch == nullptr) {
+ return Status::InternalError("New ORC lazy filter expects struct row
batch");
+ }
+
+ // Reuse the callback's column layout and buffers. A fully rejected SDK
batch can
+ // invoke this callback again before get_block() returns, so clear its
prior rows first.
+ auto& filter_block = _state->orc_lazy_predicate_block;
+ if (filter_block.columns() == 0) {
+ RETURN_IF_ERROR(_init_lazy_predicate_block());
+ }
+ filter_block.clear_column_data(_request->local_positions.size());
+
std::set<format::LocalColumnId> decoded_columns;
RETURN_IF_ERROR(_decode_columns(*struct_batch,
_request->predicate_columns, size, &filter_block,
Review Comment:
[P2] Preserve dictionary-ID filtering for matching ORC stripes
For `SELECT payload FROM t WHERE s = 'hit'` with a two-value STRING
dictionary, rare hits in every row group and a repeated 64-KiB nonmatching
value, metadata and the restored stripe callback must retain the stripe. V1
rewrote this predicate to matching dictionary IDs, filtered integer codes, then
expanded only survivors. This path instead copies every dictionary string in
`_decode_columns()` before `_build_keep_filter()`: a 4096-row batch can expand
nearly 256 MiB of rejected strings even with one SerDe call. Queries previously
selecting V1 now lose that optimization. Retain safe per-stripe ID matches and
filter encoded rows before string expansion, with a selective
matching-dictionary test. This is distinct from the existing absent-value
stripe-pruning and duplicate-conversion findings.
##########
be/test/format/table/iceberg/iceberg_position_delete_sys_table_reader_test.cpp:
##########
@@ -1,1054 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-#include "format/table/iceberg_position_delete_sys_table_reader.h"
-
-#include <arrow/api.h>
-#include <arrow/io/api.h>
-#include <gtest/gtest.h>
-#include <parquet/arrow/writer.h>
-
-#include <filesystem>
-#include <fstream>
-#include <memory>
-#include <optional>
-#include <string>
-#include <unordered_map>
-#include <utility>
-#include <vector>
-
-#include "common/object_pool.h"
-#include "core/column/column_nullable.h"
-#include "core/column/column_string.h"
-#include "core/column/column_struct.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 "core/data_type/data_type_struct.h"
-#include "core/data_type/data_type_varbinary.h"
-#include "exprs/vexpr.h"
-#include "format/orc/orc_memory_stream_test.h"
-#include "format/table/iceberg_scan_semantics.h"
-#include "format/table/parquet_utils.h"
-#include "format_v2/table/iceberg_position_delete_sys_table_reader.h"
-#include "io/io_common.h"
-#include "runtime/runtime_profile.h"
-#include "runtime/runtime_state.h"
-
-namespace doris {
-
-namespace {
-
-TFileRangeDesc range_with_delete_file(const TIcebergDeleteFileDesc&
delete_file);
-
-class ProfileTrackingReader final : public GenericReader {
-public:
- int collect_calls = 0;
-
-protected:
- Status _do_get_next_block(Block* /*block*/, size_t* /*read_rows*/, bool*
/*eof*/) override {
- return Status::OK();
- }
-
- void _collect_profile_before_close() override { ++collect_calls; }
-};
-
-SlotDescriptor* make_slot(ObjectPool* pool, int id, std::string name,
DataTypePtr type) {
- TSlotDescriptor slot_desc;
- slot_desc.__set_id(id);
- slot_desc.__set_parent(0);
- slot_desc.__set_slotType(type->to_thrift());
- slot_desc.__set_columnPos(id);
- slot_desc.__set_byteOffset(0);
- slot_desc.__set_nullIndicatorByte(id / 8);
- slot_desc.__set_nullIndicatorBit(id % 8);
- slot_desc.__set_slotIdx(id);
- slot_desc.__set_isMaterialized(true);
- slot_desc.__set_colName(std::move(name));
- return pool->add(new SlotDescriptor(slot_desc));
-}
-
-Block make_output_block(const std::vector<SlotDescriptor*>& slots) {
- Block block;
- for (const auto* slot : slots) {
- auto type = slot->type();
- block.insert(ColumnWithTypeAndName(type->create_column(), type,
slot->col_name()));
- }
- return block;
-}
-
-const IColumn& nested_column(const Block& block, const std::string& name) {
- const auto position = block.get_position_by_name(name);
- DORIS_CHECK(position >= 0);
- const auto& column = *block.get_by_position(position).column;
- if (const auto* nullable = check_and_get_column<ColumnNullable>(&column)) {
- return nullable->get_nested_column();
- }
- return column;
-}
-
-bool is_null_at(const Block& block, const std::string& name, size_t row) {
- const auto position = block.get_position_by_name(name);
- DORIS_CHECK(position >= 0);
- const auto* nullable =
-
check_and_get_column<ColumnNullable>(block.get_by_position(position).column.get());
- DORIS_CHECK(nullable != nullptr);
- return nullable->is_null_at(row);
-}
-
-std::string string_at(const Block& block, const std::string& name, size_t row)
{
- const auto* column =
check_and_get_column<ColumnString>(&nested_column(block, name));
- DORIS_CHECK(column != nullptr);
- return column->get_data_at(row).to_string();
-}
-
-Int64 int_at(const Block& block, const std::string& name, size_t row) {
- return nested_column(block, name).get_int(row);
-}
-
-Int64 struct_int_at(const Block& block, const std::string& name, size_t
child_index, size_t row) {
- const auto& struct_column = assert_cast<const
ColumnStruct&>(nested_column(block, name));
- const auto& child = assert_cast<const
ColumnNullable&>(struct_column.get_column(child_index));
- DORIS_CHECK(!child.is_null_at(row));
- return child.get_nested_column().get_int(row);
-}
-
-bool struct_child_is_null_at(const Block& block, const std::string& name,
size_t child_index,
- size_t row) {
- const auto& struct_column = assert_cast<const
ColumnStruct&>(nested_column(block, name));
- const auto& child = assert_cast<const
ColumnNullable&>(struct_column.get_column(child_index));
- return child.is_null_at(row);
-}
-
-std::shared_ptr<arrow::Array> int32_array(int32_t value) {
- arrow::Int32Builder builder;
- EXPECT_TRUE(builder.Append(value).ok());
- auto result = builder.Finish();
- EXPECT_TRUE(result.ok()) << result.status();
- return result.ok() ? *result : nullptr;
-}
-
-void write_mixed_id_position_delete_parquet(const std::string& path,
- std::optional<int32_t> row_id =
2147483544,
- bool include_child_id = true) {
- auto id = [](int32_t value) {
- return arrow::key_value_metadata({"PARQUET:field_id"},
{std::to_string(value)});
- };
- auto legacy_a = arrow::field("legacy_a", arrow::int32(), false);
- if (include_child_id) {
- legacy_a = legacy_a->WithMetadata(id(1));
- }
- auto idless_b = arrow::field("b", arrow::int32(), false);
- auto row_array =
- arrow::StructArray::Make({int32_array(42), int32_array(99)},
{legacy_a, idless_b});
- ASSERT_TRUE(row_array.ok()) << row_array.status();
-
- arrow::StringBuilder path_builder;
- ASSERT_TRUE(path_builder.Append("s3://bucket/data.parquet").ok());
- auto path_array = path_builder.Finish();
- ASSERT_TRUE(path_array.ok()) << path_array.status();
- arrow::Int64Builder pos_builder;
- ASSERT_TRUE(pos_builder.Append(5).ok());
- auto pos_array = pos_builder.Finish();
- ASSERT_TRUE(pos_array.ok()) << pos_array.status();
-
- auto row_field = arrow::field("row", arrow::struct_({legacy_a, idless_b}),
false);
- if (row_id.has_value()) {
- row_field = row_field->WithMetadata(id(*row_id));
- }
- auto schema = arrow::schema({
- arrow::field("file_path", arrow::utf8(),
false)->WithMetadata(id(2147483546)),
- arrow::field("pos", arrow::int64(),
false)->WithMetadata(id(2147483545)),
- row_field,
- });
- auto table = arrow::Table::Make(schema, {*path_array, *pos_array,
*row_array});
- auto output = arrow::io::FileOutputStream::Open(path);
- ASSERT_TRUE(output.ok()) << output.status();
- ::parquet::WriterProperties::Builder properties;
- properties.compression(::parquet::Compression::UNCOMPRESSED);
- PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), *output,
- 1, properties.build()));
-}
-
-void write_nested_wrapper_position_delete_parquet(const std::string& path) {
- auto id = [](int32_t value) {
- return arrow::key_value_metadata({"PARQUET:field_id"},
{std::to_string(value)});
- };
- auto legacy_a = arrow::field("legacy_a", arrow::int32(),
false)->WithMetadata(id(1));
- auto s_array = arrow::StructArray::Make({int32_array(42)}, {legacy_a});
- ASSERT_TRUE(s_array.ok()) << s_array.status();
- auto idless_s = arrow::field("s", arrow::struct_({legacy_a}), false);
- auto row_array = arrow::StructArray::Make({*s_array}, {idless_s});
- ASSERT_TRUE(row_array.ok()) << row_array.status();
-
- arrow::StringBuilder path_builder;
- ASSERT_TRUE(path_builder.Append("s3://bucket/data.parquet").ok());
- auto path_array = path_builder.Finish();
- ASSERT_TRUE(path_array.ok()) << path_array.status();
- arrow::Int64Builder pos_builder;
- ASSERT_TRUE(pos_builder.Append(5).ok());
- auto pos_array = pos_builder.Finish();
- ASSERT_TRUE(pos_array.ok()) << pos_array.status();
-
- auto schema = arrow::schema({
- arrow::field("file_path", arrow::utf8(),
false)->WithMetadata(id(2147483546)),
- arrow::field("pos", arrow::int64(),
false)->WithMetadata(id(2147483545)),
- arrow::field("row", arrow::struct_({idless_s}),
false)->WithMetadata(id(2147483544)),
- });
- auto table = arrow::Table::Make(schema, {*path_array, *pos_array,
*row_array});
- auto output = arrow::io::FileOutputStream::Open(path);
- ASSERT_TRUE(output.ok()) << output.status();
- ::parquet::WriterProperties::Builder properties;
- properties.compression(::parquet::Compression::UNCOMPRESSED);
- PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), *output,
- 1, properties.build()));
-}
-
-void write_mixed_id_position_delete_orc(const std::string& path,
- std::optional<int32_t> row_id =
2147483544,
- bool include_child_id = true) {
- auto type = std::unique_ptr<::orc::Type>(::orc::Type::buildTypeFromString(
-
"struct<file_path:string,pos:bigint,row:struct<legacy_a:int,b:int>>"));
- type->getSubtype(0)->setAttribute("iceberg.id", "2147483546");
- type->getSubtype(1)->setAttribute("iceberg.id", "2147483545");
- if (row_id.has_value()) {
- type->getSubtype(2)->setAttribute("iceberg.id",
std::to_string(*row_id));
- }
- if (include_child_id) {
- type->getSubtype(2)->getSubtype(0)->setAttribute("iceberg.id", "1");
- }
-
- MemoryOutputStream memory_stream(1024 * 1024);
- ::orc::WriterOptions options;
- options.setCompression(::orc::CompressionKind_NONE);
- options.setMemoryPool(::orc::getDefaultPool());
- auto writer = ::orc::createWriter(*type, &memory_stream, options);
- auto batch = writer->createRowBatch(1);
- auto& root = dynamic_cast<::orc::StructVectorBatch&>(*batch);
- auto& file_path = dynamic_cast<::orc::StringVectorBatch&>(*root.fields[0]);
- auto& pos = dynamic_cast<::orc::LongVectorBatch&>(*root.fields[1]);
- auto& row = dynamic_cast<::orc::StructVectorBatch&>(*root.fields[2]);
- auto& legacy_a = dynamic_cast<::orc::LongVectorBatch&>(*row.fields[0]);
- auto& idless_b = dynamic_cast<::orc::LongVectorBatch&>(*row.fields[1]);
- std::string data_path = "s3://bucket/data.orc";
- file_path.data[0] = data_path.data();
- file_path.length[0] = data_path.size();
- pos.data[0] = 5;
- legacy_a.data[0] = 42;
- idless_b.data[0] = 99;
- root.numElements = file_path.numElements = pos.numElements =
row.numElements =
- legacy_a.numElements = idless_b.numElements = 1;
- writer->add(*batch);
- writer->close();
-
- std::ofstream output(path, std::ios::binary);
- output.write(memory_stream.getData(),
static_cast<std::streamsize>(memory_stream.getLength()));
-}
-
-format::ColumnDefinition table_column(int32_t id, std::string name,
DataTypePtr type) {
- format::ColumnDefinition column;
- column.identifier = Field::create_field<TYPE_INT>(id);
- column.name = std::move(name);
- column.type = std::move(type);
- return column;
-}
-
-schema::external::TFieldPtr external_scalar_field(const std::string& name,
int32_t id,
- TPrimitiveType::type type) {
- auto field = std::make_shared<schema::external::TField>();
- field->__set_name(name);
- field->__set_id(id);
- TColumnType column_type;
- column_type.__set_type(type);
- field->__set_type(column_type);
- schema::external::TFieldPtr ptr;
- ptr.__set_field_ptr(std::move(field));
- return ptr;
-}
-
-schema::external::TStructField position_delete_table_schema() {
- schema::external::TStructField row_children;
- row_children.__set_fields({external_scalar_field("a", 1,
TPrimitiveType::INT),
- external_scalar_field("b", 2,
TPrimitiveType::INT)});
- auto row = std::make_shared<schema::external::TField>();
- row->__set_name("row");
- row->__set_id(2147483544);
- TColumnType row_type;
- row_type.__set_type(TPrimitiveType::STRUCT);
- row->__set_type(row_type);
- row->nestedField.__set_struct_field(std::move(row_children));
- row->__isset.nestedField = true;
- schema::external::TFieldPtr row_ptr;
- row_ptr.__set_field_ptr(std::move(row));
-
- schema::external::TStructField root;
- root.__set_fields({external_scalar_field("file_path", 2147483546,
TPrimitiveType::STRING),
- external_scalar_field("pos", 2147483545,
TPrimitiveType::BIGINT),
- std::move(row_ptr)});
- return root;
-}
-
-void run_v1_idless_row_position_delete_test(TFileFormatType::type file_format,
- const std::string& extension) {
- const auto test_dir = std::filesystem::temp_directory_path() /
- ("doris_v1_position_delete_idless_row_" + extension);
- std::filesystem::remove_all(test_dir);
- std::filesystem::create_directories(test_dir);
- const auto path = (test_dir / ("delete." + extension)).string();
- if (file_format == TFileFormatType::FORMAT_PARQUET) {
- write_mixed_id_position_delete_parquet(path, std::nullopt, false);
- } else {
- write_mixed_id_position_delete_orc(path, std::nullopt, false);
- }
-
- const auto nullable_int32 =
make_nullable(std::make_shared<DataTypeInt32>());
- const auto row_type = make_nullable(std::make_shared<DataTypeStruct>(
- DataTypes {nullable_int32, nullable_int32}, Strings {"a", "b"}));
- ObjectPool pool;
- std::vector<SlotDescriptor*> slots {make_slot(&pool, 0, "row", row_type)};
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- state.set_timezone("UTC");
- RuntimeProfile profile("test_profile");
- TFileScanRangeParams scan_params;
-
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_1);
- scan_params.__set_file_type(TFileType::FILE_LOCAL);
- scan_params.__set_format_type(file_format);
- scan_params.__set_current_schema_id(-1);
- schema::external::TSchema current_schema;
- current_schema.__set_schema_id(-1);
- current_schema.__set_root_field(position_delete_table_schema());
- scan_params.__set_history_schema_info({std::move(current_schema)});
- auto io_ctx = std::make_shared<io::IOContext>();
- io::FileReaderStats file_reader_stats;
- io_ctx->file_reader_stats = &file_reader_stats;
-
- TIcebergDeleteFileDesc delete_file;
- delete_file.__set_content(1);
- delete_file.__set_path(path);
- delete_file.__set_file_format(file_format);
- auto range = range_with_delete_file(delete_file);
- range.__set_path(path);
- range.__set_start_offset(0);
- range.__set_size(static_cast<int64_t>(std::filesystem::file_size(path)));
-
range.__set_file_size(static_cast<int64_t>(std::filesystem::file_size(path)));
-
- IcebergPositionDeleteSysTableReader reader(slots, &state, &profile, range,
&scan_params, io_ctx,
- nullptr);
- ReaderInitContext context;
- ASSERT_TRUE(reader.init_reader(&context).ok());
- Block block = make_output_block(slots);
- size_t read_rows = 0;
- bool eof = false;
- ASSERT_TRUE(reader.get_next_block(&block, &read_rows, &eof).ok());
- ASSERT_EQ(1, read_rows);
- ASSERT_EQ(1, block.rows());
- // Top-level delete fields carry IDs, so an ID-less physical row must not
bind by name.
- EXPECT_TRUE(is_null_at(block, "row", 0));
-
- ASSERT_TRUE(reader.close().ok());
- std::filesystem::remove_all(test_dir);
-}
-
-void run_mixed_id_position_delete_test(format::FileFormat file_format,
- TFileFormatType::type thrift_format,
- const std::string& extension) {
- const auto test_dir = std::filesystem::temp_directory_path() /
- ("doris_iceberg_position_delete_mixed_id_" +
extension);
- std::filesystem::remove_all(test_dir);
- std::filesystem::create_directories(test_dir);
- const auto path = (test_dir / ("delete." + extension)).string();
- if (file_format == format::FileFormat::PARQUET) {
- write_mixed_id_position_delete_parquet(path);
- } else {
- write_mixed_id_position_delete_orc(path);
- }
-
- const auto nullable_int32 =
make_nullable(std::make_shared<DataTypeInt32>());
- const auto row_type = make_nullable(std::make_shared<DataTypeStruct>(
- DataTypes {nullable_int32, nullable_int32}, Strings {"a", "b"}));
- auto a = table_column(1, "a", nullable_int32);
- a.name_mapping = {"legacy_a"};
- a.has_name_mapping = true;
- auto b = table_column(2, "b", nullable_int32);
- auto row = table_column(2147483544, "row", row_type);
- row.children = {a, b};
- std::vector<format::ColumnDefinition> projected_columns {row};
-
- ObjectPool pool;
- std::vector<SlotDescriptor*> slots {make_slot(&pool, 0, "row", row_type)};
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- RuntimeProfile profile("test_profile");
- TFileScanRangeParams scan_params;
-
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_1);
- scan_params.__set_file_type(TFileType::FILE_LOCAL);
- scan_params.__set_format_type(thrift_format);
- io::FileReaderStats file_reader_stats;
- auto io_ctx = std::make_shared<io::IOContext>();
- io_ctx->file_reader_stats = &file_reader_stats;
-
- format::iceberg::IcebergPositionDeleteSysTableV2Reader reader;
- ASSERT_TRUE(reader.init({
- .projected_columns = projected_columns,
- .conjuncts = {},
- .format = file_format,
- .scan_params = &scan_params,
- .io_ctx = io_ctx,
- .runtime_state = &state,
- .scanner_profile = &profile,
- .file_slot_descs = &slots,
- })
- .ok());
- TIcebergDeleteFileDesc delete_file;
- delete_file.__set_content(1);
- delete_file.__set_path(path);
- delete_file.__set_file_format(thrift_format);
- auto range = range_with_delete_file(delete_file);
- range.__set_path(path);
- range.__set_start_offset(0);
-
range.__set_file_size(static_cast<int64_t>(std::filesystem::file_size(path)));
- format::SplitReadOptions split_options;
- split_options.current_range = std::move(range);
- split_options.current_split_format = file_format;
- ASSERT_TRUE(reader.prepare_split(split_options).ok());
-
- Block block = make_output_block(slots);
- bool eos = false;
- ASSERT_TRUE(reader.get_block(&block, &eos).ok());
- ASSERT_FALSE(eos);
- ASSERT_EQ(1, block.rows());
- ASSERT_FALSE(struct_child_is_null_at(block, "row", 0, 0));
- EXPECT_EQ(42, struct_int_at(block, "row", 0, 0));
- // Once any file field has an Iceberg id, the id-less sibling is absent
rather than name-bound.
- EXPECT_TRUE(struct_child_is_null_at(block, "row", 1, 0));
-
- ASSERT_TRUE(reader.close().ok());
- std::filesystem::remove_all(test_dir);
-}
-
-void run_v2_nested_wrapper_position_delete_test() {
- const auto test_dir = std::filesystem::temp_directory_path() /
- "doris_v2_position_delete_nested_wrapper_parquet";
- std::filesystem::remove_all(test_dir);
- std::filesystem::create_directories(test_dir);
- const auto path = (test_dir / "delete.parquet").string();
- write_nested_wrapper_position_delete_parquet(path);
-
- const auto nullable_int32 =
make_nullable(std::make_shared<DataTypeInt32>());
- const auto s_type = make_nullable(
- std::make_shared<DataTypeStruct>(DataTypes {nullable_int32},
Strings {"a"}));
- const auto row_type =
- make_nullable(std::make_shared<DataTypeStruct>(DataTypes {s_type},
Strings {"s"}));
- auto a = table_column(1, "a", nullable_int32);
- auto s = table_column(10, "s", s_type);
- s.children = {a};
- auto row = table_column(2147483544, "row", row_type);
- row.children = {s};
- std::vector<format::ColumnDefinition> projected_columns {row};
-
- ObjectPool pool;
- std::vector<SlotDescriptor*> slots {make_slot(&pool, 0, "row", row_type)};
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- RuntimeProfile profile("test_profile");
- TFileScanRangeParams scan_params;
-
scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_1);
- scan_params.__set_file_type(TFileType::FILE_LOCAL);
- scan_params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
- io::FileReaderStats file_reader_stats;
- auto io_ctx = std::make_shared<io::IOContext>();
- io_ctx->file_reader_stats = &file_reader_stats;
-
- format::iceberg::IcebergPositionDeleteSysTableV2Reader reader;
- ASSERT_TRUE(reader.init({
- .projected_columns = projected_columns,
- .conjuncts = {},
- .format = format::FileFormat::PARQUET,
- .scan_params = &scan_params,
- .io_ctx = io_ctx,
- .runtime_state = &state,
- .scanner_profile = &profile,
- .file_slot_descs = &slots,
- })
- .ok());
- TIcebergDeleteFileDesc delete_file;
- delete_file.__set_content(1);
- delete_file.__set_path(path);
- delete_file.__set_file_format(TFileFormatType::FORMAT_PARQUET);
- auto range = range_with_delete_file(delete_file);
- range.__set_path(path);
- range.__set_start_offset(0);
-
range.__set_file_size(static_cast<int64_t>(std::filesystem::file_size(path)));
- format::SplitReadOptions split_options;
- split_options.current_range = std::move(range);
- split_options.current_split_format = format::FileFormat::PARQUET;
- ASSERT_TRUE(reader.prepare_split(split_options).ok());
-
- Block block = make_output_block(slots);
- bool eos = false;
- ASSERT_TRUE(reader.get_block(&block, &eos).ok());
- ASSERT_FALSE(eos);
- ASSERT_EQ(1, block.rows());
- EXPECT_EQ("{\"s\":{\"a\":42}}",
row_type->to_string(*block.get_by_position(0).column, 0));
-
- ASSERT_TRUE(reader.close().ok());
- std::filesystem::remove_all(test_dir);
-}
-
-TFileRangeDesc range_with_delete_file(const TIcebergDeleteFileDesc&
delete_file) {
- TIcebergFileDesc iceberg_desc;
- iceberg_desc.__set_delete_files({delete_file});
- TTableFormatFileDesc table_format_desc;
- table_format_desc.__set_iceberg_params(std::move(iceberg_desc));
- TFileRangeDesc range;
- range.__set_table_format_params(std::move(table_format_desc));
- return range;
-}
-
-} // namespace
-
-TEST(IcebergPositionDeleteSysTableReaderTest, UsesScannerIOContext) {
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- RuntimeProfile profile("test_profile");
- TFileRangeDesc range;
- TFileScanRangeParams params;
- std::vector<SlotDescriptor*> file_slot_descs;
- auto scanner_io_ctx = std::make_shared<io::IOContext>();
- io::FileReaderStats file_reader_stats;
- scanner_io_ctx->file_reader_stats = &file_reader_stats;
-
- IcebergPositionDeleteSysTableReader reader(file_slot_descs, &state,
&profile, range, ¶ms,
- scanner_io_ctx, nullptr);
-
- EXPECT_EQ(scanner_io_ctx.get(), reader._io_ctx.get());
- EXPECT_EQ(&file_reader_stats, reader._io_ctx->file_reader_stats);
- scanner_io_ctx->should_stop = true;
- EXPECT_TRUE(reader._io_ctx->should_stop);
-
- EXPECT_TRUE(reader.count_read_rows());
- reader._delete_file_kind =
IcebergPositionDeleteSysTableReader::DeleteFileKind::DELETION_VECTOR;
- EXPECT_FALSE(reader.count_read_rows());
-}
-
-TEST(IcebergPositionDeleteSysTableReaderTest,
ForwardsProfileCollectionToNestedReader) {
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- RuntimeProfile profile("test_profile");
- TFileRangeDesc range;
- TFileScanRangeParams params;
- std::vector<SlotDescriptor*> file_slot_descs;
- auto scanner_io_ctx = std::make_shared<io::IOContext>();
-
- IcebergPositionDeleteSysTableReader reader(file_slot_descs, &state,
&profile, range, ¶ms,
- scanner_io_ctx, nullptr);
- auto nested_reader = std::make_unique<ProfileTrackingReader>();
- auto* nested_reader_ptr = nested_reader.get();
- reader._position_reader = std::move(nested_reader);
-
- reader.collect_profile_before_close();
- reader.collect_profile_before_close();
-
- EXPECT_EQ(1, nested_reader_ptr->collect_calls);
- reader._dv_positions.add(uint64_t {1});
- reader._next_dv_position.emplace(reader._dv_positions.begin());
- reader._partition_value =
std::make_shared<DataTypeInt32>()->create_column();
- ASSERT_TRUE(reader.close().ok());
- EXPECT_TRUE(reader._dv_positions.isEmpty());
- EXPECT_FALSE(reader._next_dv_position.has_value());
- EXPECT_EQ(nullptr, reader._partition_value.get());
-}
-
-TEST(IcebergPositionDeleteSysTableReaderTest,
StopsBeforeExpandingDeletionVector) {
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- RuntimeProfile profile("test_profile");
- TFileRangeDesc range;
- TFileScanRangeParams params;
- std::vector<SlotDescriptor*> file_slot_descs;
- auto scanner_io_ctx = std::make_shared<io::IOContext>();
- scanner_io_ctx->should_stop = true;
-
- IcebergPositionDeleteSysTableReader reader(file_slot_descs, &state,
&profile, range, ¶ms,
- scanner_io_ctx, nullptr);
- Block block;
- size_t read_rows = 1;
- bool eof = false;
-
- ASSERT_TRUE(reader._do_get_next_block(&block, &read_rows, &eof).ok());
- EXPECT_EQ(0, read_rows);
- EXPECT_TRUE(eof);
-}
-
-TEST(IcebergPositionDeleteSysTableReaderTest,
ValidatesRangeAndDeleteFileMetadata) {
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- RuntimeProfile profile("test_profile");
- TFileScanRangeParams params;
- std::vector<SlotDescriptor*> file_slot_descs;
- auto scanner_io_ctx = std::make_shared<io::IOContext>();
-
- TFileRangeDesc empty_range;
- IcebergPositionDeleteSysTableReader invalid_context_reader(
- file_slot_descs, &state, &profile, empty_range, ¶ms, nullptr,
nullptr);
- auto status = invalid_context_reader._do_init_reader(nullptr);
- EXPECT_TRUE(status.is<ErrorCode::INVALID_ARGUMENT>()) << status;
-
- IcebergPositionDeleteSysTableReader missing_params_reader(
- file_slot_descs, &state, &profile, empty_range, ¶ms,
scanner_io_ctx, nullptr);
- status = missing_params_reader._do_init_reader(nullptr);
- EXPECT_NE(std::string::npos, status.to_string().find("range misses
params"));
-
- TTableFormatFileDesc table_format_desc;
- table_format_desc.__set_iceberg_params(TIcebergFileDesc());
- TFileRangeDesc no_delete_file_range;
-
no_delete_file_range.__set_table_format_params(std::move(table_format_desc));
- IcebergPositionDeleteSysTableReader no_delete_file_reader(file_slot_descs,
&state, &profile,
-
no_delete_file_range, ¶ms,
- scanner_io_ctx,
nullptr);
- status = no_delete_file_reader._do_init_reader(nullptr);
- EXPECT_NE(std::string::npos, status.to_string().find("exactly one delete
file"));
-
- TIcebergDeleteFileDesc missing_content;
- auto missing_content_range = range_with_delete_file(missing_content);
- IcebergPositionDeleteSysTableReader
missing_content_reader(file_slot_descs, &state, &profile,
-
missing_content_range, ¶ms,
- scanner_io_ctx,
nullptr);
- status = missing_content_reader._do_init_reader(nullptr);
- EXPECT_NE(std::string::npos, status.to_string().find("misses content"));
-
- TIcebergDeleteFileDesc equality_delete;
- equality_delete.__set_content(2);
- auto equality_delete_range = range_with_delete_file(equality_delete);
- IcebergPositionDeleteSysTableReader
equality_delete_reader(file_slot_descs, &state, &profile,
-
equality_delete_range, ¶ms,
- scanner_io_ctx,
nullptr);
- status = equality_delete_reader._do_init_reader(nullptr);
- EXPECT_NE(std::string::npos, status.to_string().find("does not support
delete file content 2"));
-
- TIcebergDeleteFileDesc missing_format;
- missing_format.__set_content(1);
- auto missing_format_range = range_with_delete_file(missing_format);
- IcebergPositionDeleteSysTableReader missing_format_reader(file_slot_descs,
&state, &profile,
-
missing_format_range, ¶ms,
- scanner_io_ctx,
nullptr);
- status = missing_format_reader._do_init_reader(nullptr);
- EXPECT_NE(std::string::npos, status.to_string().find("misses file
format"));
-
- TIcebergDeleteFileDesc unsupported_format;
- unsupported_format.__set_content(1);
- unsupported_format.__set_file_format(TFileFormatType::FORMAT_CSV_PLAIN);
- auto unsupported_format_range = range_with_delete_file(unsupported_format);
- IcebergPositionDeleteSysTableReader
unsupported_format_reader(file_slot_descs, &state, &profile,
-
unsupported_format_range, ¶ms,
-
scanner_io_ctx, nullptr);
- status = unsupported_format_reader._do_init_reader(nullptr);
- EXPECT_TRUE(status.is<ErrorCode::NOT_IMPLEMENTED_ERROR>()) << status;
-
- TIcebergDeleteFileDesc invalid_dv;
- invalid_dv.__set_content(3);
- auto invalid_dv_range = range_with_delete_file(invalid_dv);
- IcebergPositionDeleteSysTableReader invalid_dv_reader(
- file_slot_descs, &state, &profile, invalid_dv_range, ¶ms,
scanner_io_ctx, nullptr);
- status = invalid_dv_reader._do_init_reader(nullptr);
- EXPECT_NE(std::string::npos, status.to_string().find("misses referenced
data file path"));
-}
-
-TEST(IcebergPositionDeleteSysTableReaderTest,
AppendsDeletionVectorMetadataAndCachesPartition) {
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- RuntimeProfile profile("test_profile");
- TFileRangeDesc range;
- TFileScanRangeParams params;
- ObjectPool pool;
- const auto nullable_string =
make_nullable(std::make_shared<DataTypeString>());
- const auto nullable_int32 =
make_nullable(std::make_shared<DataTypeInt32>());
- const auto nullable_int64 =
make_nullable(std::make_shared<DataTypeInt64>());
- const auto partition_type = make_nullable(
- std::make_shared<DataTypeStruct>(DataTypes {nullable_int32},
Strings {"p"}));
- std::vector<SlotDescriptor*> slots {
- make_slot(&pool, 0, "file_path", nullable_string),
- make_slot(&pool, 1, "pos", nullable_int64),
- make_slot(&pool, 2, "row", nullable_int32),
- make_slot(&pool, 3, "partition", partition_type),
- make_slot(&pool, 4, "spec_id", nullable_int32),
- make_slot(&pool, 5, "delete_file_path", nullable_string),
- make_slot(&pool, 6, "content_offset", nullable_int64),
- make_slot(&pool, 7, "content_size_in_bytes", nullable_int64),
- };
- auto scanner_io_ctx = std::make_shared<io::IOContext>();
- IcebergPositionDeleteSysTableReader reader(slots, &state, &profile, range,
¶ms,
- scanner_io_ctx, nullptr);
-
- TIcebergFileDesc iceberg_desc;
- iceberg_desc.__set_partition_spec_id(7);
- iceberg_desc.__set_partition_data_json(R"({"p":42})");
- TIcebergDeleteFileDesc delete_file;
- delete_file.__set_path("/physical-delete.puffin");
- delete_file.__set_original_path("s3://bucket/delete.puffin");
- delete_file.__set_referenced_data_file_path("s3://bucket/data.parquet");
- delete_file.__set_content_offset(12);
- delete_file.__set_content_size_in_bytes(34);
- reader._iceberg_file_desc = &iceberg_desc;
- reader._delete_file_desc = &delete_file;
- reader._delete_file_kind =
IcebergPositionDeleteSysTableReader::DeleteFileKind::DELETION_VECTOR;
- reader._batch_size = 1;
- reader._dv_positions.add(uint64_t {5});
- reader._dv_positions.add(uint64_t {9});
- reader._next_dv_position.emplace(reader._dv_positions.begin());
-
- Block block = make_output_block(slots);
- size_t read_rows = 0;
- bool eof = false;
- ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows,
&eof).ok());
- EXPECT_EQ(1, read_rows);
- EXPECT_FALSE(eof);
- ASSERT_EQ(1, block.rows());
- ColumnPtr cached_partition = reader._partition_value;
- ASSERT_NE(nullptr, cached_partition.get());
- EXPECT_EQ("s3://bucket/data.parquet", string_at(block, "file_path", 0));
- EXPECT_EQ(5, int_at(block, "pos", 0));
- EXPECT_TRUE(is_null_at(block, "row", 0));
- EXPECT_FALSE(is_null_at(block, "partition", 0));
- EXPECT_EQ(42, struct_int_at(block, "partition", 0, 0));
- EXPECT_EQ(7, int_at(block, "spec_id", 0));
- EXPECT_EQ("s3://bucket/delete.puffin", string_at(block,
"delete_file_path", 0));
- EXPECT_EQ(12, int_at(block, "content_offset", 0));
- EXPECT_EQ(34, int_at(block, "content_size_in_bytes", 0));
-
- ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows,
&eof).ok());
- EXPECT_EQ(1, read_rows);
- EXPECT_TRUE(eof);
- ASSERT_EQ(2, block.rows());
- EXPECT_EQ(cached_partition.get(), reader._partition_value.get());
- EXPECT_EQ(9, int_at(block, "pos", 1));
- EXPECT_FALSE(is_null_at(block, "partition", 1));
- EXPECT_EQ(42, struct_int_at(block, "partition", 0, 1));
-
- ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows,
&eof).ok());
- EXPECT_EQ(0, read_rows);
- EXPECT_TRUE(eof);
-
- reader._dv_positions = roaring::Roaring64Map();
- reader._dv_positions.add(uint64_t {13});
- reader._next_dv_position.emplace(reader._dv_positions.begin());
- std::vector<SlotDescriptor*> partial_slots {slots[0], slots[1]};
- Block partial_block = make_output_block(partial_slots);
- ASSERT_TRUE(reader._append_deletion_vector_block(&partial_block,
&read_rows, &eof).ok());
- EXPECT_EQ(1, read_rows);
- EXPECT_TRUE(eof);
- EXPECT_EQ(13, int_at(partial_block, "pos", 0));
-}
-
-TEST(IcebergPositionDeleteSysTableReaderTest,
AppendsPositionDeleteRowsAndValidatesColumns) {
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- RuntimeProfile profile("test_profile");
- TFileRangeDesc range;
- TFileScanRangeParams params;
- ObjectPool pool;
- const auto nullable_string =
make_nullable(std::make_shared<DataTypeString>());
- const auto nullable_int32 =
make_nullable(std::make_shared<DataTypeInt32>());
- const auto nullable_int64 =
make_nullable(std::make_shared<DataTypeInt64>());
- std::vector<SlotDescriptor*> slots {
- make_slot(&pool, 0, "file_path", nullable_string),
- make_slot(&pool, 1, "pos", nullable_int64),
- make_slot(&pool, 2, "row", nullable_int32),
- };
- auto scanner_io_ctx = std::make_shared<io::IOContext>();
- IcebergPositionDeleteSysTableReader reader(slots, &state, &profile, range,
¶ms,
- scanner_io_ctx, nullptr);
- TIcebergFileDesc iceberg_desc;
- TIcebergDeleteFileDesc delete_file;
- delete_file.__set_path("/delete.parquet");
- reader._iceberg_file_desc = &iceberg_desc;
- reader._delete_file_desc = &delete_file;
- reader._delete_file_kind =
IcebergPositionDeleteSysTableReader::DeleteFileKind::POSITION_DELETE;
-
- reader.set_batch_size(17);
- EXPECT_EQ(17, reader.get_batch_size());
- EXPECT_TRUE(reader._output_column_requested("row"));
- EXPECT_FALSE(reader._output_column_requested("missing"));
- reader._init_read_columns(true);
- ASSERT_EQ(3, reader._read_columns.size());
- EXPECT_EQ("row", reader._read_columns.back().name);
-
- std::unordered_map<std::string, DataTypePtr> name_to_type;
- ASSERT_TRUE(reader._get_columns_impl(&name_to_type).ok());
- EXPECT_EQ(3, name_to_type.size());
-
- Block delete_block = reader._create_delete_block();
- {
- auto columns_guard = delete_block.mutate_columns_scoped();
- auto& columns = columns_guard.mutable_columns();
- parquet_utils::insert_string(columns[0], "s3://bucket/data.parquet");
- parquet_utils::insert_int64(columns[1], 19);
- parquet_utils::insert_int32(columns[2], 23);
- }
- Block output_block = make_output_block(slots);
- size_t appended_rows = 0;
- ASSERT_TRUE(reader._append_position_delete_block(&output_block,
delete_block, 1, &appended_rows)
- .ok());
- EXPECT_EQ(1, appended_rows);
- ASSERT_EQ(1, output_block.rows());
- EXPECT_EQ("s3://bucket/data.parquet", string_at(output_block, "file_path",
0));
- EXPECT_EQ(19, int_at(output_block, "pos", 0));
- EXPECT_EQ(23, int_at(output_block, "row", 0));
-
- Block empty_delete_block = reader._create_delete_block();
- auto path_column = nullable_string->create_column();
- auto status = reader._append_sys_column(path_column, *slots[0],
&empty_delete_block, 0, 0);
- EXPECT_NE(std::string::npos, status.to_string().find("file_path column is
missing"));
- auto pos_column = nullable_int64->create_column();
- status = reader._append_sys_column(pos_column, *slots[1],
&empty_delete_block, 0, 0);
- EXPECT_NE(std::string::npos, status.to_string().find("pos column is
missing"));
- auto row_column = nullable_int32->create_column();
- ASSERT_TRUE(reader._append_sys_column(row_column, *slots[2],
&empty_delete_block, 0, 0).ok());
- ASSERT_EQ(1, row_column->size());
-
- Block missing_columns;
- path_column = nullable_string->create_column();
- status = reader._append_sys_column(path_column, *slots[0],
&missing_columns, 0, 0);
- EXPECT_NE(std::string::npos, status.to_string().find("file_path column is
missing"));
- pos_column = nullable_int64->create_column();
- status = reader._append_sys_column(pos_column, *slots[1],
&missing_columns, 0, 0);
- EXPECT_NE(std::string::npos, status.to_string().find("pos column is
missing"));
-
- std::vector<SlotDescriptor*> partial_slots {slots[0], slots[1]};
- Block partial_output_block = make_output_block(partial_slots);
- appended_rows = 0;
- ASSERT_TRUE(reader._append_position_delete_block(&partial_output_block,
delete_block, 1,
- &appended_rows)
- .ok());
- EXPECT_EQ(1, appended_rows);
- EXPECT_EQ(1, partial_output_block.rows());
-
- auto* unknown_slot = make_slot(&pool, 3, "unknown", nullable_string);
- auto unknown_column = nullable_string->create_column();
- status = reader._append_sys_column(unknown_column, *unknown_slot,
&delete_block, 0, 0);
- EXPECT_NE(std::string::npos, status.to_string().find("Unknown Iceberg"));
-
- Block unused_block;
- size_t read_rows = 0;
- bool eof = false;
- status = reader._do_get_next_block(&unused_block, &read_rows, &eof);
- EXPECT_NE(std::string::npos, status.to_string().find("reader is not
initialized"));
-}
-
-TEST(IcebergPositionDeleteSysTableReaderTest,
AppendsNullMetadataAndUsesDeletePathFallback) {
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- RuntimeProfile profile("test_profile");
- TFileRangeDesc range;
- TFileScanRangeParams params;
- ObjectPool pool;
- const auto nullable_string =
make_nullable(std::make_shared<DataTypeString>());
- const auto nullable_int32 =
make_nullable(std::make_shared<DataTypeInt32>());
- const auto nullable_int64 =
make_nullable(std::make_shared<DataTypeInt64>());
- const auto partition_type = make_nullable(
- std::make_shared<DataTypeStruct>(DataTypes {nullable_int32},
Strings {"p"}));
- std::vector<SlotDescriptor*> slots {
- make_slot(&pool, 0, "partition", partition_type),
- make_slot(&pool, 1, "spec_id", nullable_int32),
- make_slot(&pool, 2, "delete_file_path", nullable_string),
- make_slot(&pool, 3, "content_offset", nullable_int64),
- make_slot(&pool, 4, "content_size_in_bytes", nullable_int64),
- };
- auto scanner_io_ctx = std::make_shared<io::IOContext>();
- IcebergPositionDeleteSysTableReader reader(slots, &state, &profile, range,
¶ms,
- scanner_io_ctx, nullptr);
- TIcebergFileDesc iceberg_desc;
- TIcebergDeleteFileDesc delete_file;
- delete_file.__set_path("/fallback-delete-file");
- reader._iceberg_file_desc = &iceberg_desc;
- reader._delete_file_desc = &delete_file;
- reader._delete_file_kind =
IcebergPositionDeleteSysTableReader::DeleteFileKind::DELETION_VECTOR;
- reader._dv_positions.add(uint64_t {1});
- reader._next_dv_position.emplace(reader._dv_positions.begin());
-
- Block block = make_output_block(slots);
- size_t read_rows = 0;
- bool eof = false;
- ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows,
&eof).ok());
- EXPECT_EQ(1, read_rows);
- EXPECT_TRUE(eof);
- EXPECT_TRUE(is_null_at(block, "partition", 0));
- EXPECT_TRUE(is_null_at(block, "spec_id", 0));
- EXPECT_EQ("/fallback-delete-file", string_at(block, "delete_file_path",
0));
- EXPECT_TRUE(is_null_at(block, "content_offset", 0));
- EXPECT_TRUE(is_null_at(block, "content_size_in_bytes", 0));
-
- std::vector<SlotDescriptor*> empty_slots;
- IcebergPositionDeleteSysTableReader empty_reader(empty_slots, &state,
&profile, range, ¶ms,
- scanner_io_ctx, nullptr);
- empty_reader._dv_positions.add(uint64_t {2});
- empty_reader._next_dv_position.emplace(empty_reader._dv_positions.begin());
- Block empty_block;
- ASSERT_TRUE(empty_reader._append_deletion_vector_block(&empty_block,
&read_rows, &eof).ok());
- EXPECT_EQ(1, read_rows);
- EXPECT_TRUE(eof);
-}
-
-TEST(IcebergPositionDeleteSysTableReaderTest,
ParquetUsesFullFileIdModeForIdlessRow) {
- run_v1_idless_row_position_delete_test(TFileFormatType::FORMAT_PARQUET,
"parquet");
-}
-
-TEST(IcebergPositionDeleteSysTableReaderTest,
OrcUsesFullFileIdModeForIdlessRow) {
- run_v1_idless_row_position_delete_test(TFileFormatType::FORMAT_ORC, "orc");
-}
-
-TEST(IcebergPositionDeleteSysTableV2ReaderTest, RecordsDeletionVectorRows) {
- io::FileReaderStats file_reader_stats;
- auto scanner_io_ctx = std::make_shared<io::IOContext>();
- scanner_io_ctx->file_reader_stats = &file_reader_stats;
- std::vector<SlotDescriptor*> file_slot_descs;
-
- format::iceberg::IcebergPositionDeleteSysTableV2Reader reader;
- reader._io_ctx = scanner_io_ctx;
- reader._file_slot_descs = &file_slot_descs;
- reader._batch_size = 2;
- reader._dv_positions.add(uint64_t {7});
- reader._dv_positions.add(uint64_t {9});
- reader._dv_positions.add(uint64_t {11});
- reader._next_dv_position.emplace(reader._dv_positions.begin());
-
- Block block;
- size_t read_rows = 0;
- bool eof = true;
- ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows,
&eof).ok());
- EXPECT_EQ(2, read_rows);
- EXPECT_FALSE(eof);
- EXPECT_EQ(2, file_reader_stats.read_rows);
-
- ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows,
&eof).ok());
- EXPECT_EQ(1, read_rows);
- EXPECT_FALSE(eof);
- EXPECT_EQ(3, file_reader_stats.read_rows);
-}
-
-TEST(IcebergPositionDeleteSysTableV2ReaderTest, PreservesUuidPartitionBytes) {
- ObjectPool pool;
- const auto partition_type = make_nullable(std::make_shared<DataTypeStruct>(
- DataTypes {make_nullable(std::make_shared<DataTypeVarbinary>())},
Strings {"p"}));
- auto* slot = make_slot(&pool, 0, "partition", partition_type);
- TIcebergFileDesc descriptor;
-
descriptor.__set_partition_data_json(R"({"p":"0x123E4567E89B12D3A456426614174000"})");
- auto verify = [&](auto& reader) {
- reader._iceberg_file_desc = &descriptor;
- auto column = partition_type->create_column();
- ASSERT_TRUE(reader._append_partition_column(column, *slot).ok());
- ASSERT_TRUE(reader._append_partition_column(column, *slot).ok());
- const auto& structure = assert_cast<const ColumnStruct&>(
- assert_cast<const
ColumnNullable&>(*column).get_nested_column());
- const auto& field = assert_cast<const
ColumnNullable&>(structure.get_column(0));
- for (size_t row = 0; row < 2; ++row) {
- ASSERT_FALSE(field.is_null_at(row));
- EXPECT_EQ(
-
std::string("\x12\x3e\x45\x67\xe8\x9b\x12\xd3\xa4\x56\x42\x66\x14\x17\x40\x00",
- 16),
- field.get_nested_column().get_data_at(row).to_string());
- }
- };
- format::iceberg::IcebergPositionDeleteSysTableV2Reader reader;
- verify(reader);
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- RuntimeProfile profile("test_profile");
- TFileRangeDesc range;
- TFileScanRangeParams params;
- std::vector<SlotDescriptor*> slots {slot};
- IcebergPositionDeleteSysTableReader legacy_reader(slots, &state, &profile,
range, ¶ms,
-
std::make_shared<io::IOContext>(), nullptr);
- verify(legacy_reader);
-}
-
-TEST(IcebergPositionDeleteSysTableV2ReaderTest, CachesAndClearsPartitionValue)
{
- ObjectPool pool;
- const auto nullable_int32 =
make_nullable(std::make_shared<DataTypeInt32>());
- const auto partition_type = make_nullable(
- std::make_shared<DataTypeStruct>(DataTypes {nullable_int32},
Strings {"p"}));
- auto* partition_slot = make_slot(&pool, 0, "partition", partition_type);
-
- TIcebergFileDesc iceberg_desc;
- iceberg_desc.__set_partition_data_json(R"({"p":42})");
-
- format::iceberg::IcebergPositionDeleteSysTableV2Reader reader;
- reader._iceberg_file_desc = &iceberg_desc;
- auto partition_column = partition_type->create_column();
- ASSERT_TRUE(reader._append_partition_column(partition_column,
*partition_slot).ok());
- ColumnPtr cached_partition = reader._partition_value;
- ASSERT_NE(nullptr, cached_partition.get());
-
- ASSERT_TRUE(reader._append_partition_column(partition_column,
*partition_slot).ok());
- EXPECT_EQ(cached_partition.get(), reader._partition_value.get());
-
- Block block;
- block.insert(ColumnWithTypeAndName(std::move(partition_column),
partition_type, "partition"));
- ASSERT_EQ(2, block.rows());
- EXPECT_FALSE(is_null_at(block, "partition", 0));
- EXPECT_EQ(42, struct_int_at(block, "partition", 0, 0));
- EXPECT_FALSE(is_null_at(block, "partition", 1));
- EXPECT_EQ(42, struct_int_at(block, "partition", 0, 1));
-
- reader._has_split = true;
- reader._dv_positions.add(uint64_t {1});
- reader._next_dv_position.emplace(reader._dv_positions.begin());
- ASSERT_TRUE(reader.close().ok());
- EXPECT_EQ(nullptr, reader._iceberg_file_desc);
- EXPECT_EQ(nullptr, reader._partition_value.get());
- EXPECT_TRUE(reader._dv_positions.isEmpty());
- EXPECT_FALSE(reader._next_dv_position.has_value());
- EXPECT_FALSE(reader._has_split);
-}
-
-TEST(IcebergPositionDeleteSysTableV2ReaderTest,
ValidatesDeleteFileContentAfterBindingDescriptor) {
- RuntimeState state {TQueryOptions(), TQueryGlobals()};
- RuntimeProfile profile("test_profile");
- TFileScanRangeParams params;
- std::vector<SlotDescriptor*> file_slot_descs;
- TIcebergDeleteFileDesc equality_delete;
- equality_delete.__set_content(2);
-
- format::iceberg::IcebergPositionDeleteSysTableV2Reader reader;
- reader._runtime_state = &state;
- reader._scanner_profile = &profile;
- reader._scan_params = ¶ms;
- reader._file_slot_descs = &file_slot_descs;
- reader._current_range = range_with_delete_file(equality_delete);
-
- auto status = reader._init_split();
- EXPECT_NE(std::string::npos, status.to_string().find("does not support
delete file content 2"));
- ASSERT_NE(nullptr, reader._iceberg_file_desc);
- ASSERT_NE(nullptr, reader._delete_file_desc);
- EXPECT_EQ(reader._iceberg_file_desc->delete_files.data(),
reader._delete_file_desc);
-}
-
-TEST(IcebergPositionDeleteSysTableV2ReaderTest,
StopsBeforeExpandingDeletionVector) {
- format::iceberg::IcebergPositionDeleteSysTableV2Reader reader;
- reader._io_ctx = std::make_shared<io::IOContext>();
- reader._io_ctx->should_stop = true;
-
- Block block;
- bool eof = false;
- ASSERT_TRUE(reader.get_block(&block, &eof).ok());
- EXPECT_TRUE(eof);
-}
-
-TEST(IcebergPositionDeleteSysTableV2ReaderTest,
ParquetRowUsesAnyFieldIdMapping) {
Review Comment:
[P2] Keep the V2 regressions when deleting the legacy suite
This file also contains eight tests of the surviving
`IcebergPositionDeleteSysTableV2Reader`, including
`ParquetRowUsesAnyFieldIdMapping`, `OrcRowUsesAnyFieldIdMapping` and
`ParquetReadsNestedIdlessWrapper`. Their fixtures read non-null row payloads
and assert mapped values and missing-field NULLs; other cases check UUID
partition bytes and partition-cache cleanup. These active cases are removed
rather than relocated: the maintained V2 suite only keeps
profile/default-preparation tests, and the external system-table suite's
absent/null row payloads do not replace the projection checks. Move the V2
tests and their fixtures into the maintained suite before deleting this file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
