fix: optimize shared-shredding read & fix ORC read-size estimation for nested columns (#216)
diff --git a/src/paimon/common/data/shredding/map_shared_shredding_file_reader.cpp b/src/paimon/common/data/shredding/map_shared_shredding_file_reader.cpp
index b7c8409..b1cdf96 100644
--- a/src/paimon/common/data/shredding/map_shared_shredding_file_reader.cpp
+++ b/src/paimon/common/data/shredding/map_shared_shredding_file_reader.cpp
@@ -32,7 +32,6 @@
#include "paimon/common/utils/arrow/mem_utils.h"
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/checked_cast.h"
-#include "paimon/core/casting/casting_utils.h"
#include "paimon/core/utils/nested_projection_utils.h"
namespace paimon {
@@ -110,6 +109,56 @@
std::vector<SelectedKey> selected_keys_;
};
+Result<std::shared_ptr<arrow::Array>> MaskSinglePhysicalColumn(
+ const std::shared_ptr<arrow::StructArray>& physical_struct_array,
+ const std::shared_ptr<arrow::ListArray>& field_mapping_array,
+ const std::shared_ptr<arrow::Int32Array>& field_mapping_values,
+ const std::shared_ptr<arrow::Array>& physical_column_array, int32_t physical_column_id,
+ int32_t field_id, const std::string& field_name, arrow::MemoryPool* arrow_pool) {
+ int64_t row_count = physical_struct_array->length();
+ if (physical_column_array->length() != row_count) {
+ return Status::Invalid("shared-shredding physical column length does not match row count");
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Buffer> validity,
+ arrow::AllocateEmptyBitmap(row_count, arrow_pool));
+ int64_t valid_count = 0;
+ for (int64_t row = 0; row < row_count; ++row) {
+ if (physical_struct_array->IsNull(row)) {
+ continue;
+ }
+ if (field_mapping_array->IsNull(row)) {
+ return Status::Invalid(fmt::format(
+ "__field_mapping cannot be null in non-null shared-shredding row for field {}",
+ field_name));
+ }
+ int32_t mapping_offset = field_mapping_array->value_offset(row);
+ int32_t mapping_length = field_mapping_array->value_length(row);
+ if (physical_column_id < 0 || physical_column_id >= mapping_length) {
+ return Status::Invalid("physical column id is out of __field_mapping range");
+ }
+ int32_t mapping_index = mapping_offset + physical_column_id;
+ if (field_mapping_values->IsNull(mapping_index)) {
+ return Status::Invalid("__field_mapping element cannot be null");
+ }
+ if (field_mapping_values->Value(mapping_index) != field_id ||
+ physical_column_array->IsNull(row)) {
+ continue;
+ }
+ arrow::bit_util::SetBit(validity->mutable_data(), row);
+ ++valid_count;
+ }
+
+ // Replace only the top-level validity; offsets, values, and nested children stay shared.
+ std::shared_ptr<arrow::ArrayData> result_data = physical_column_array->data()->Copy();
+ if (result_data->buffers.empty()) {
+ return Status::Invalid("shared-shredding physical column has no validity buffer slot");
+ }
+ int64_t null_count = row_count - valid_count;
+ result_data->buffers[0] = null_count == 0 ? nullptr : std::move(validity);
+ result_data->SetNullCount(null_count);
+ return arrow::MakeArray(std::move(result_data));
+}
+
class DefaultSelectedKeysReadPlan : public MapFieldReadPlan {
public:
DefaultSelectedKeysReadPlan(const std::shared_ptr<arrow::Field>& logical_field,
@@ -386,12 +435,10 @@
std::shared_ptr<arrow::MapArray> overflow_array;
CollectPhysicalColumns(physical_struct_array, &physical_column_name_to_array, &overflow_array);
for (auto& [_, physical_column_array] : physical_column_name_to_array) {
- if (physical_column_array->type_id() == arrow::Type::DICTIONARY) {
- PAIMON_ASSIGN_OR_RAISE(
- physical_column_array,
- CastingUtils::Cast(physical_column_array, logical_map_type_->item_type(),
- arrow::compute::CastOptions::Safe(), arrow_pool));
- }
+ PAIMON_ASSIGN_OR_RAISE(
+ physical_column_array,
+ NestedProjectionUtils::AlignArrayToReadType(
+ physical_column_array, logical_map_type_->item_type(), arrow_pool));
}
std::shared_ptr<arrow::Int32Array> overflow_keys;
@@ -406,12 +453,9 @@
if (!overflow_items) {
return Status::Invalid("__overflow map item array is null");
}
- if (overflow_items->type_id() == arrow::Type::DICTIONARY) {
- PAIMON_ASSIGN_OR_RAISE(
- overflow_items,
- CastingUtils::Cast(overflow_items, logical_map_type_->item_type(),
- arrow::compute::CastOptions::Safe(), arrow_pool));
- }
+ PAIMON_ASSIGN_OR_RAISE(overflow_items,
+ NestedProjectionUtils::AlignArrayToReadType(
+ overflow_items, logical_map_type_->item_type(), arrow_pool));
}
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::unique_ptr<arrow::ArrayBuilder> map_builder_base,
@@ -530,12 +574,9 @@
std::shared_ptr<arrow::MapArray> overflow_array;
CollectPhysicalColumns(physical_struct_array, &physical_column_name_to_array, &overflow_array);
for (auto& [_, physical_column_array] : physical_column_name_to_array) {
- if (physical_column_array->type_id() == arrow::Type::DICTIONARY) {
- PAIMON_ASSIGN_OR_RAISE(
- physical_column_array,
- CastingUtils::Cast(physical_column_array, value_type,
- arrow::compute::CastOptions::Safe(), arrow_pool));
- }
+ PAIMON_ASSIGN_OR_RAISE(physical_column_array,
+ NestedProjectionUtils::AlignArrayToReadType(physical_column_array,
+ value_type, arrow_pool));
}
std::shared_ptr<arrow::Int32Array> overflow_keys;
@@ -550,65 +591,89 @@
if (!overflow_items) {
return Status::Invalid("__overflow map item array is null");
}
- if (overflow_items->type_id() == arrow::Type::DICTIONARY) {
- PAIMON_ASSIGN_OR_RAISE(
- overflow_items,
- CastingUtils::Cast(overflow_items, value_type, arrow::compute::CastOptions::Safe(),
- arrow_pool));
- }
+ PAIMON_ASSIGN_OR_RAISE(overflow_items, NestedProjectionUtils::AlignArrayToReadType(
+ overflow_items, value_type, arrow_pool));
}
- std::unique_ptr<arrow::ArrayBuilder> access_builder_base;
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(access_builder_base,
- arrow::MakeBuilder(LogicalField()->type(), arrow_pool));
- if (!access_builder_base || !access_builder_base->type() ||
- access_builder_base->type()->id() != arrow::Type::STRUCT) {
- return Status::Invalid(
- fmt::format("selected-key MAP field {} is not a STRUCT", LogicalField()->name()));
- }
- auto* access_builder = checked_cast<arrow::StructBuilder*>(access_builder_base.get());
- PAIMON_RETURN_NOT_OK_FROM_ARROW(access_builder->Reserve(physical_struct_array->length()));
-
- for (int64_t row = 0; row < physical_struct_array->length(); ++row) {
- if (physical_struct_array->IsNull(row)) {
- PAIMON_RETURN_NOT_OK_FROM_ARROW(access_builder->AppendNull());
+ int64_t row_count = physical_struct_array->length();
+ arrow::ArrayVector selected_key_arrays;
+ selected_key_arrays.reserve(selected_keys_.size());
+ for (int32_t key_index = 0; key_index < selected_keys_type->num_fields(); ++key_index) {
+ const SelectedKey& selected_key = selected_keys_[key_index];
+ if (selected_key.field_id < 0) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::Array> null_array,
+ arrow::MakeArrayOfNull(selected_keys_type->field(key_index)->type(), row_count,
+ arrow_pool));
+ selected_key_arrays.push_back(std::move(null_array));
continue;
}
- if (field_mapping_array->IsNull(row)) {
- return Status::Invalid(fmt::format(
- "__field_mapping cannot be null in non-null shared-shredding row for field {}",
- LogicalField()->name()));
- }
- int32_t mapping_offset = field_mapping_array->value_offset(row);
- PAIMON_RETURN_NOT_OK_FROM_ARROW(access_builder->Append());
- for (int32_t key_index = 0; key_index < selected_keys_type->num_fields(); ++key_index) {
- arrow::ArrayBuilder* value_builder = access_builder->field_builder(key_index);
- const SelectedKey& selected_key = selected_keys_[key_index];
+ if (selected_key.candidate_columns.size() == 1 && !selected_key.may_use_overflow) {
+ int32_t physical_column_id = selected_key.candidate_columns[0];
+ std::string physical_column_name =
+ MapSharedShreddingDefine::PhysicalColumnName(physical_column_id);
+ auto physical_column_iter = physical_column_name_to_array.find(physical_column_name);
+ if (physical_column_iter == physical_column_name_to_array.end()) {
+ return Status::Invalid(
+ fmt::format("cannot find selected physical column {} for field {}",
+ physical_column_name, LogicalField()->name()));
+ }
+ const std::shared_ptr<arrow::Array>& physical_column_array =
+ physical_column_iter->second;
+ if (physical_column_array->offset() == 0) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<arrow::Array> masked_array,
+ MaskSinglePhysicalColumn(physical_struct_array, field_mapping_array,
+ field_mapping_values, physical_column_array,
+ physical_column_id, selected_key.field_id,
+ LogicalField()->name(), arrow_pool));
+ selected_key_arrays.push_back(std::move(masked_array));
+ continue;
+ } else {
+ return Status::Invalid("paimon only supports arrays with zero offset");
+ }
+ }
+
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::unique_ptr<arrow::ArrayBuilder> value_builder,
+ arrow::MakeBuilder(selected_keys_type->field(key_index)->type(), arrow_pool));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder->Reserve(row_count));
+ for (int64_t row = 0; row < row_count; ++row) {
+ if (physical_struct_array->IsNull(row)) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder->AppendNull());
+ continue;
+ }
+ if (field_mapping_array->IsNull(row)) {
+ return Status::Invalid(fmt::format(
+ "__field_mapping cannot be null in non-null shared-shredding row for field {}",
+ LogicalField()->name()));
+ }
+ int32_t mapping_offset = field_mapping_array->value_offset(row);
bool appended = false;
- if (selected_key.field_id >= 0) {
- for (int32_t physical_column_id : selected_key.candidate_columns) {
- int32_t mapping_index = mapping_offset + physical_column_id;
- if (field_mapping_values->IsNull(mapping_index)) {
- return Status::Invalid("__field_mapping element cannot be null");
- }
- if (field_mapping_values->Value(mapping_index) != selected_key.field_id) {
- continue;
- }
- std::string physical_column_name =
- MapSharedShreddingDefine::PhysicalColumnName(physical_column_id);
- auto physical_column_iter =
- physical_column_name_to_array.find(physical_column_name);
- if (physical_column_iter == physical_column_name_to_array.end()) {
- return Status::Invalid(
- fmt::format("cannot find selected physical column {} for field {}",
- physical_column_name, LogicalField()->name()));
- }
- PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder->AppendArraySlice(
- *physical_column_iter->second->data(), row, 1));
- appended = true;
- break;
+ for (int32_t physical_column_id : selected_key.candidate_columns) {
+ int32_t mapping_index = mapping_offset + physical_column_id;
+ if (field_mapping_values->IsNull(mapping_index)) {
+ return Status::Invalid("__field_mapping element cannot be null");
}
+ if (field_mapping_values->Value(mapping_index) != selected_key.field_id) {
+ continue;
+ }
+ std::string physical_column_name =
+ MapSharedShreddingDefine::PhysicalColumnName(physical_column_id);
+ auto physical_column_iter =
+ physical_column_name_to_array.find(physical_column_name);
+ if (physical_column_iter == physical_column_name_to_array.end()) {
+ return Status::Invalid(
+ fmt::format("cannot find selected physical column {} for field {}",
+ physical_column_name, LogicalField()->name()));
+ }
+ const std::shared_ptr<arrow::Array>& physical_column_array =
+ physical_column_iter->second;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(
+ value_builder->AppendArraySlice(*physical_column_array->data(), row, 1));
+ appended = true;
+ break;
}
if (!appended && selected_key.may_use_overflow && overflow_array &&
@@ -630,9 +695,24 @@
PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder->AppendNull());
}
}
+ std::shared_ptr<arrow::Array> selected_key_array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder->Finish(&selected_key_array));
+ selected_key_arrays.push_back(std::move(selected_key_array));
}
- std::shared_ptr<arrow::Array> result;
- PAIMON_RETURN_NOT_OK_FROM_ARROW(access_builder->Finish(&result));
+
+ std::shared_ptr<arrow::Buffer> parent_validity;
+ int64_t parent_null_count = physical_struct_array->null_count();
+ if (parent_null_count > 0) {
+ if (physical_struct_array->offset() == 0) {
+ parent_validity = physical_struct_array->null_bitmap();
+ } else {
+ return Status::Invalid("paimon only supports arrays with zero offset");
+ }
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::StructArray> result,
+ arrow::StructArray::Make(selected_key_arrays, selected_keys_type->fields(),
+ std::move(parent_validity), parent_null_count));
return result;
}
@@ -646,14 +726,10 @@
}
auto map_array = checked_pointer_cast<arrow::MapArray>(physical_array);
auto selected_keys_type = checked_pointer_cast<arrow::StructType>(LogicalField()->type());
- auto physical_map_type = checked_pointer_cast<arrow::MapType>(PhysicalReadField()->type());
std::shared_ptr<arrow::Array> items = map_array->items();
- if (items->type_id() == arrow::Type::DICTIONARY) {
- PAIMON_ASSIGN_OR_RAISE(items,
- CastingUtils::Cast(items, physical_map_type->item_type(),
- arrow::compute::CastOptions::Safe(), arrow_pool));
- }
+ PAIMON_ASSIGN_OR_RAISE(items, NestedProjectionUtils::AlignArrayToReadType(
+ items, selected_keys_type->field(0)->type(), arrow_pool));
std::shared_ptr<arrow::Array> keys = map_array->keys();
std::unique_ptr<arrow::ArrayBuilder> access_builder_base;
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(access_builder_base,
diff --git a/src/paimon/common/data/shredding/map_shared_shredding_file_reader_test.cpp b/src/paimon/common/data/shredding/map_shared_shredding_file_reader_test.cpp
index 80f5046..b2dd937 100644
--- a/src/paimon/common/data/shredding/map_shared_shredding_file_reader_test.cpp
+++ b/src/paimon/common/data/shredding/map_shared_shredding_file_reader_test.cpp
@@ -333,6 +333,96 @@
AssertChunkedArrayEquals(expected, actual);
}
+TEST_F(MapSharedShreddingFileReaderTest, TestSelectedKeysStructProjectionSharesValueBuffers) {
+ ASSERT_OK_AND_ASSIGN(auto physical_array, PhysicalArray());
+ auto physical_root = checked_pointer_cast<arrow::StructArray>(physical_array);
+ auto physical_tags =
+ checked_pointer_cast<arrow::StructArray>(physical_root->GetFieldByName("tags"));
+ auto physical_column =
+ physical_tags->GetFieldByName(MapSharedShreddingDefine::PhysicalColumnName(1));
+
+ auto selected_type = arrow::struct_(
+ {arrow::field("a", arrow::int64()), arrow::field("b", arrow::int64()),
+ arrow::field("e", arrow::int64()), arrow::field("missing", arrow::int64())});
+ auto selected_field = arrow::field(
+ "tags", selected_type, /*nullable=*/true,
+ arrow::KeyValueMetadata::Make({DataField::MAP_SELECTED_KEYS}, {"a,b,e,missing"}));
+ ASSERT_OK_AND_ASSIGN(
+ auto field_read_plan,
+ MapFieldReadPlanFactory::CreateSharedSelectedKeysReadPlan(selected_field, TagsMeta()));
+ ASSERT_OK_AND_ASSIGN(auto result,
+ field_read_plan->Materialize(physical_tags, arrow::default_memory_pool()));
+ auto result_struct = checked_pointer_cast<arrow::StructArray>(result);
+
+ auto expected = arrow::ipc::internal::json::ArrayFromJSON(selected_type, R"([
+ [10, 20, null, null],
+ [40, null, null, null],
+ null,
+ [80, null, 70, null]
+ ])")
+ .ValueOrDie();
+ ASSERT_TRUE(expected->Equals(result)) << "Expected:\n"
+ << expected->ToString() << "\nActual:\n"
+ << result->ToString();
+ ASSERT_EQ(physical_column->data()->buffers[1], result_struct->field(1)->data()->buffers[1]);
+ ASSERT_EQ(physical_column->data()->buffers[1], result_struct->field(2)->data()->buffers[1]);
+ ASSERT_NE(physical_column->data()->buffers[0], result_struct->field(1)->data()->buffers[0]);
+ ASSERT_EQ(physical_tags->data()->buffers[0], result_struct->data()->buffers[0]);
+}
+
+TEST_F(MapSharedShreddingFileReaderTest, TestSelectedKeysStructProjectionSharesNestedValueBuffers) {
+ auto item_type = arrow::list(arrow::int64());
+ auto logical_schema =
+ arrow::schema({arrow::field("id", arrow::int32()),
+ arrow::field("tags", arrow::map(arrow::utf8(), item_type))});
+ ASSERT_OK_AND_ASSIGN(auto physical_schema, MapSharedShreddingUtils::LogicalToPhysicalSchema(
+ logical_schema, {{"tags", 1}}));
+ auto physical_array =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(physical_schema->fields()), R"([
+ [1, [[0], [1, 2], null]],
+ [2, [[1], [3, 4, 5], null]],
+ [3, null],
+ [4, [[0], null, null]]
+ ])")
+ .ValueOrDie();
+ auto physical_root = checked_pointer_cast<arrow::StructArray>(physical_array);
+ auto physical_tags =
+ checked_pointer_cast<arrow::StructArray>(physical_root->GetFieldByName("tags"));
+ auto physical_column =
+ physical_tags->GetFieldByName(MapSharedShreddingDefine::PhysicalColumnName(0));
+
+ MapSharedShreddingFieldMeta meta;
+ meta.name_to_id = {{"a", 0}, {"b", 1}};
+ meta.field_to_columns = {{0, {0}}, {1, {0}}};
+ meta.num_columns = 1;
+ meta.max_row_width = 1;
+ auto selected_type = arrow::struct_({arrow::field("b", item_type)});
+ auto selected_field =
+ arrow::field("tags", selected_type, /*nullable=*/true,
+ arrow::KeyValueMetadata::Make({DataField::MAP_SELECTED_KEYS}, {"b"}));
+ ASSERT_OK_AND_ASSIGN(
+ auto field_read_plan,
+ MapFieldReadPlanFactory::CreateSharedSelectedKeysReadPlan(selected_field, meta));
+ ASSERT_OK_AND_ASSIGN(auto result,
+ field_read_plan->Materialize(physical_tags, arrow::default_memory_pool()));
+ auto result_struct = checked_pointer_cast<arrow::StructArray>(result);
+ auto result_list = checked_pointer_cast<arrow::ListArray>(result_struct->field(0));
+
+ auto expected = arrow::ipc::internal::json::ArrayFromJSON(selected_type, R"([
+ [null],
+ [[3, 4, 5]],
+ null,
+ [null]
+ ])")
+ .ValueOrDie();
+ ASSERT_TRUE(expected->Equals(result)) << "Expected:\n"
+ << expected->ToString() << "\nActual:\n"
+ << result->ToString();
+ ASSERT_EQ(physical_column->data()->buffers[1], result_list->data()->buffers[1]);
+ ASSERT_EQ(physical_column->data()->child_data[0]->buffers[1],
+ result_list->data()->child_data[0]->buffers[1]);
+}
+
TEST_F(MapSharedShreddingFileReaderTest, TestSelectedKeysStructProjectionFromDefaultMap) {
auto map_type = checked_pointer_cast<arrow::MapType>(
arrow::map(arrow::utf8(), arrow::field("value", arrow::int64())));
@@ -674,6 +764,66 @@
AssertChunkedArrayEquals(expected, actual);
}
+TEST_F(MapSharedShreddingFileReaderTest, TestOrcDictionaryEncodedStringListValue) {
+ std::shared_ptr<arrow::Schema> logical_schema = arrow::schema({
+ arrow::field("id", arrow::int32()),
+ arrow::field("tags", arrow::map(arrow::utf8(), arrow::list(arrow::utf8()))),
+ });
+ auto options = options_;
+ std::string format = "orc";
+ options[Options::FILE_FORMAT] = format;
+ options["orc.dictionary-key-size-threshold"] = "1";
+ ASSERT_OK_AND_ASSIGN(auto table_schema,
+ TableSchema::Create(TableSchema::FIRST_SCHEMA_ID, logical_schema,
+ /*partition_keys=*/{}, /*primary_keys=*/{}, options));
+
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap(options));
+ auto path_factory = CreatePathFactory(dir->Str(), format, core_options);
+ auto compact_manager = std::make_shared<NoopCompactManager>();
+ ASSERT_OK_AND_ASSIGN(
+ auto writer,
+ CreateAppendOnlyWriter(core_options, /*schema_id=*/0, logical_schema,
+ /*write_cols=*/std::nullopt,
+ /*max_sequence_number=*/-1, path_factory, compact_manager));
+ auto batch = CreateBatch(logical_schema, R"([
+ [1, [["a", ["red", "blue"]], ["b", ["blue"]]]],
+ [2, [["c", ["green"]], ["a", ["red", null, "blue"]], ["b", ["blue"]]]],
+ [3, null],
+ [4, [["d", ["yellow"]], ["e", ["blue"]], ["c", [null]], ["a", ["red"]]]]
+ ])");
+ ASSERT_OK(writer->Write(std::move(batch)));
+ ASSERT_OK_AND_ASSIGN(auto inc, writer->PrepareCommit(/*wait_compaction=*/true));
+ ASSERT_OK(writer->Close());
+
+ std::string data_file_path =
+ path_factory->ToPath(inc.GetNewFilesIncrement().NewFiles()[0]->file_name);
+ std::map<std::string, std::string> reader_options = {{"orc.read.enable-lazy-decoding", "true"}};
+ auto reader = WrapReader(OpenFormatReader(data_file_path, format, reader_options),
+ /*selected_keys_str=*/"a,c");
+
+ auto read_metadata = std::make_shared<arrow::KeyValueMetadata>();
+ read_metadata->Append("paimon.map.selected-keys", "a,c");
+ arrow::FieldVector read_fields = logical_schema->fields();
+ read_fields[1] = read_fields[1]->WithMetadata(read_metadata);
+ auto read_schema = ExportSchema(arrow::schema(std::move(read_fields)));
+ ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt));
+ ASSERT_OK_AND_ASSIGN(auto actual, ReadResultCollector::CollectResult(reader.get()));
+ std::shared_ptr<arrow::ChunkedArray> expected;
+ ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON(
+ arrow::struct_(logical_schema->fields()), {R"([
+ [1, [["a", ["red", "blue"]]]],
+ [2, [["a", ["red", null, "blue"]], ["c", ["green"]]]],
+ [3, null],
+ [4, [["a", ["red"]], ["c", [null]]]]
+ ])"},
+ &expected)
+ .ok());
+ AssertChunkedArrayEquals(expected, actual);
+}
+
TEST_F(MapSharedShreddingFileReaderTest, TestReadsRealFormatFile) {
// TODO(lisizhuo.lsz): support other format
auto options = options_;
diff --git a/src/paimon/format/orc/orc_file_batch_reader.cpp b/src/paimon/format/orc/orc_file_batch_reader.cpp
index cd627eb..a201839 100644
--- a/src/paimon/format/orc/orc_file_batch_reader.cpp
+++ b/src/paimon/format/orc/orc_file_batch_reader.cpp
@@ -46,6 +46,16 @@
#include "paimon/format/orc/predicate_converter.h"
namespace paimon::orc {
+namespace {
+
+void CollectAllColumnIds(const ::orc::Type* type, std::vector<uint64_t>* column_ids) {
+ column_ids->push_back(type->getColumnId());
+ for (uint64_t i = 0; i < type->getSubtypeCount(); ++i) {
+ CollectAllColumnIds(type->getSubtype(i), column_ids);
+ }
+}
+
+} // namespace
OrcFileBatchReader::OrcFileBatchReader(std::unique_ptr<::orc::ReaderMetrics>&& reader_metrics,
std::unique_ptr<OrcReaderWrapper>&& reader,
@@ -228,13 +238,15 @@
}
break;
}
- // Do not support partial field recall inside list/map types.
default: {
+ // Partial field recall inside list/map types is unsupported, so the target must match
+ // the complete source subtree. Include the container and every descendant because all
+ // of their streams are recalled by the ORC reader.
if (src_type->toString() != target_type->toString()) {
return Status::Invalid(fmt::format("type mismatch: src {} vs target {}",
src_type->toString(), target_type->toString()));
}
- target_column_ids->push_back(src_type->getColumnId());
+ CollectAllColumnIds(src_type, target_column_ids);
break;
}
}
diff --git a/src/paimon/format/orc/orc_file_batch_reader_test.cpp b/src/paimon/format/orc/orc_file_batch_reader_test.cpp
index 038c93e..c39ba51 100644
--- a/src/paimon/format/orc/orc_file_batch_reader_test.cpp
+++ b/src/paimon/format/orc/orc_file_batch_reader_test.cpp
@@ -368,8 +368,9 @@
OrcFileBatchReader::CreateRowReaderOptions(
src_type.get(), target_type.get(),
/*search_arg=*/nullptr, options, &target_column_ids));
- // Struct IDs (0, 1) not included. Selected: sub1(2), sub2-list(3), sub3(6), col3-map(8).
- ASSERT_EQ(target_column_ids, (std::vector<uint64_t>{2, 3, 6, 8}));
+ // Struct IDs (0, 1) are not included. LIST/MAP containers include their complete
+ // subtrees: sub1(2), sub2(3, 4, 5), sub3(6), col3(8, 9, 10).
+ ASSERT_EQ(target_column_ids, (std::vector<uint64_t>{2, 3, 4, 5, 6, 8, 9, 10}));
}
{
// read with type mismatch in nested field
@@ -485,6 +486,98 @@
}
}
+TEST_F(OrcFileBatchReaderTest, TestCollectTargetColumnIdsPrimitiveList) {
+ std::unique_ptr<::orc::Type> src_type =
+ ::orc::Type::buildTypeFromString("struct<items:array<int>,ignored:string>");
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString("struct<items:array<int>>");
+ std::vector<uint64_t> target_column_ids;
+
+ ASSERT_OK(OrcFileBatchReader::CollectTargetColumnIds(src_type.get(), target_type.get(),
+ &target_column_ids));
+ // root struct(0), items-list(1), element(2), ignored(3)
+ ASSERT_EQ(target_column_ids, (std::vector<uint64_t>{1, 2}));
+}
+
+TEST_F(OrcFileBatchReaderTest, TestCollectTargetColumnIdsDeeplyNestedList) {
+ std::string schema = "struct<items:array<array<struct<value:double,label:string>>>>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(schema);
+ std::unique_ptr<::orc::Type> target_type = ::orc::Type::buildTypeFromString(schema);
+ std::vector<uint64_t> target_column_ids;
+
+ ASSERT_OK(OrcFileBatchReader::CollectTargetColumnIds(src_type.get(), target_type.get(),
+ &target_column_ids));
+ // root struct(0), outer list(1), inner list(2), element struct(3), value(4), label(5)
+ ASSERT_EQ(target_column_ids, (std::vector<uint64_t>{1, 2, 3, 4, 5}));
+}
+
+TEST_F(OrcFileBatchReaderTest, TestCollectTargetColumnIdsPrimitiveMap) {
+ std::unique_ptr<::orc::Type> src_type =
+ ::orc::Type::buildTypeFromString("struct<attributes:map<string,int>,ignored:double>");
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString("struct<attributes:map<string,int>>");
+ std::vector<uint64_t> target_column_ids;
+
+ ASSERT_OK(OrcFileBatchReader::CollectTargetColumnIds(src_type.get(), target_type.get(),
+ &target_column_ids));
+ // root struct(0), attributes-map(1), key(2), value(3), ignored(4)
+ ASSERT_EQ(target_column_ids, (std::vector<uint64_t>{1, 2, 3}));
+}
+
+TEST_F(OrcFileBatchReaderTest, TestCollectTargetColumnIdsDeeplyNestedMap) {
+ std::string schema =
+ "struct<attributes:map<string,array<struct<score:double,tags:array<string>>>>>";
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(schema);
+ std::unique_ptr<::orc::Type> target_type = ::orc::Type::buildTypeFromString(schema);
+ std::vector<uint64_t> target_column_ids;
+
+ ASSERT_OK(OrcFileBatchReader::CollectTargetColumnIds(src_type.get(), target_type.get(),
+ &target_column_ids));
+ // root struct(0), map(1), key(2), value-list(3), element struct(4), score(5),
+ // tags-list(6), tag element(7)
+ ASSERT_EQ(target_column_ids, (std::vector<uint64_t>{1, 2, 3, 4, 5, 6, 7}));
+}
+
+TEST_F(OrcFileBatchReaderTest, TestCollectTargetColumnIdsStructProjectionWithListAndMap) {
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(
+ "struct<outer:struct<items:array<int>,plain:double,attributes:map<string,int>>,"
+ "ignored:string>");
+ std::unique_ptr<::orc::Type> target_type = ::orc::Type::buildTypeFromString(
+ "struct<outer:struct<items:array<int>,attributes:map<string,int>>>");
+ std::vector<uint64_t> target_column_ids;
+
+ ASSERT_OK(OrcFileBatchReader::CollectTargetColumnIds(src_type.get(), target_type.get(),
+ &target_column_ids));
+ // root struct(0) and outer struct(1) are not included. Selected: items(2, 3) and
+ // attributes(5, 6, 7). plain(4) and ignored(8) are skipped.
+ ASSERT_EQ(target_column_ids, (std::vector<uint64_t>{2, 3, 5, 6, 7}));
+}
+
+TEST_F(OrcFileBatchReaderTest, TestCollectTargetColumnIdsRejectsPartialListAndMapProjection) {
+ {
+ std::unique_ptr<::orc::Type> src_type =
+ ::orc::Type::buildTypeFromString("struct<items:array<struct<a:int,b:double>>>");
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString("struct<items:array<struct<a:int>>>");
+ std::vector<uint64_t> target_column_ids;
+ ASSERT_NOK_WITH_MSG(OrcFileBatchReader::CollectTargetColumnIds(
+ src_type.get(), target_type.get(), &target_column_ids),
+ "type mismatch");
+ ASSERT_TRUE(target_column_ids.empty());
+ }
+ {
+ std::unique_ptr<::orc::Type> src_type = ::orc::Type::buildTypeFromString(
+ "struct<attributes:map<string,struct<a:int,b:double>>>");
+ std::unique_ptr<::orc::Type> target_type =
+ ::orc::Type::buildTypeFromString("struct<attributes:map<string,struct<a:int>>>");
+ std::vector<uint64_t> target_column_ids;
+ ASSERT_NOK_WITH_MSG(OrcFileBatchReader::CollectTargetColumnIds(
+ src_type.get(), target_type.get(), &target_column_ids),
+ "type mismatch");
+ ASSERT_TRUE(target_column_ids.empty());
+ }
+}
+
TEST_P(OrcFileBatchReaderTest, TestNextBatchSimple) {
std::string file_name = paimon::test::GetDataDir() +
"/orc/append_09.db/append_09/f1=10/bucket-1/"
