lxy-9602 commented on code in PR #195:
URL: https://github.com/apache/paimon-cpp/pull/195#discussion_r3762961242
##########
src/paimon/common/data/shredding/map_shared_shredding_file_reader.cpp:
##########
@@ -34,14 +35,254 @@
#include "paimon/core/casting/casting_utils.h"
namespace paimon {
+namespace {
+
+Result<std::string_view> GetStringMapKey(const std::shared_ptr<arrow::Array>&
keys, int64_t index) {
+ if (keys->IsNull(index)) {
+ return Status::Invalid("selected-key MAP read found a null MAP key");
+ }
+ if (keys->type_id() == arrow::Type::STRING) {
+ return
arrow::internal::checked_pointer_cast<arrow::StringArray>(keys)->GetView(index);
+ } else if (keys->type_id() == arrow::Type::DICTIONARY) {
+ auto dictionary =
arrow::internal::checked_pointer_cast<arrow::DictionaryArray>(keys);
+ int64_t dictionary_index = dictionary->GetValueIndex(index);
+ const auto& values = dictionary->dictionary();
+ if (values->IsNull(dictionary_index)) {
+ return Status::Invalid("selected-key MAP read found a null
dictionary MAP key");
+ }
+ if (values->type_id() == arrow::Type::STRING) {
+ return
arrow::internal::checked_pointer_cast<arrow::StringArray>(values)->GetView(
+ dictionary_index);
+ }
+ }
+ return Status::Invalid(
+ fmt::format("selected-key MAP read only supports string or dictionary
key array"));
+}
+
+std::vector<std::pair<std::string, int32_t>> ResolveSelectedKeyIds(
+ const MapSharedShreddingFieldMeta& meta, const std::vector<std::string>&
selected_keys) {
+ std::vector<std::pair<std::string, int32_t>> selected_key_ids;
+ selected_key_ids.reserve(selected_keys.size());
+ for (const auto& selected_key : selected_keys) {
+ auto id_iter = meta.name_to_id.find(selected_key);
+ if (id_iter != meta.name_to_id.end()) {
+ selected_key_ids.emplace_back(selected_key, id_iter->second);
+ }
+ }
+ return selected_key_ids;
+}
+
+void CollectPhysicalColumns(
+ const std::shared_ptr<arrow::StructArray>& physical_struct_array,
+ std::map<std::string, std::shared_ptr<arrow::Array>>*
physical_column_name_to_array,
+ std::shared_ptr<arrow::MapArray>* overflow_array) {
+ const auto& struct_type = physical_struct_array->struct_type();
+ for (int32_t i = 0; i < struct_type->num_fields(); ++i) {
+ const auto& sub_field = struct_type->field(i);
+ if (sub_field->name() == MapSharedShreddingDefine::kFieldMapping) {
+ continue;
+ }
+ if (sub_field->name() == MapSharedShreddingDefine::kOverflow) {
+ *overflow_array =
arrow::internal::checked_pointer_cast<arrow::MapArray>(
+ physical_struct_array->field(i));
+ continue;
+ }
+ (*physical_column_name_to_array)[sub_field->name()] =
physical_struct_array->field(i);
+ }
+}
+
+class FullMapReadPlan : public MapFieldReadPlan {
+ public:
+ FullMapReadPlan(const std::shared_ptr<arrow::Field>& logical_field,
+ const std::shared_ptr<arrow::Field>& physical_read_field,
+ std::vector<std::pair<std::string, int32_t>>&&
selected_key_ids)
+ : MapFieldReadPlan(logical_field, physical_read_field),
+ selected_key_ids_(std::move(selected_key_ids)),
+ logical_map_type_(
+
arrow::internal::checked_pointer_cast<arrow::MapType>(logical_field->type())) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<std::pair<std::string, int32_t>> selected_key_ids_;
+ std::shared_ptr<arrow::MapType> logical_map_type_;
+};
+
+class SharedSelectedKeysReadPlan : public MapFieldReadPlan {
+ public:
+ struct SelectedKey {
+ int32_t field_id = -1;
+ std::vector<int32_t> candidate_columns;
+ bool may_use_overflow = false;
+ };
+
+ SharedSelectedKeysReadPlan(const std::shared_ptr<arrow::Field>&
logical_field,
+ const std::shared_ptr<arrow::Field>&
physical_read_field,
+ std::vector<SelectedKey>&& selected_keys)
+ : MapFieldReadPlan(logical_field, physical_read_field),
+ selected_keys_(std::move(selected_keys)) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<SelectedKey> selected_keys_;
+};
+
+class DefaultSelectedKeysReadPlan : public MapFieldReadPlan {
+ public:
+ DefaultSelectedKeysReadPlan(const std::shared_ptr<arrow::Field>&
logical_field,
+ const std::shared_ptr<arrow::Field>&
physical_read_field,
+ const std::vector<std::string>& selected_keys)
+ : MapFieldReadPlan(logical_field, physical_read_field),
selected_keys_(selected_keys) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<std::string> selected_keys_;
+};
+
+} // namespace
+
+Result<std::unique_ptr<MapFieldReadPlan>>
MapFieldReadPlanFactory::CreateFullMapReadPlan(
+ const std::shared_ptr<arrow::Field>& logical_map_field, const
MapSharedShreddingFieldMeta& meta,
+ const std::vector<std::string>& selected_keys) {
+ if (logical_map_field->type()->id() != arrow::Type::MAP) {
+ return Status::Invalid(fmt::format("full MAP read plan requires MAP
field {}, got {}",
+ logical_map_field->name(),
+
logical_map_field->type()->ToString()));
+ }
+ auto logical_map_type =
+
arrow::internal::checked_pointer_cast<arrow::MapType>(logical_map_field->type());
+ std::set<int32_t> selected_physical_column_ids;
+ bool include_overflow = false;
+ for (const auto& selected_key : selected_keys) {
+ auto field_id_iter = meta.name_to_id.find(selected_key);
+ if (field_id_iter == meta.name_to_id.end()) {
+ continue;
+ }
+ int32_t field_id = field_id_iter->second;
+ include_overflow = include_overflow ||
meta.overflow_field_set.count(field_id) > 0;
+ auto columns_iter = meta.field_to_columns.find(field_id);
+ if (columns_iter != meta.field_to_columns.end()) {
+ selected_physical_column_ids.insert(columns_iter->second.begin(),
+ columns_iter->second.end());
+ }
+ }
+ std::shared_ptr<arrow::DataType> physical_type =
+ MapSharedShreddingUtils::BuildSpecificPhysicalStructType(
+ logical_map_type->item_type(), selected_physical_column_ids,
+ logical_map_type->item_field()->nullable(), include_overflow);
+ auto physical_read_field = logical_map_field->WithType(physical_type);
+ std::unique_ptr<MapFieldReadPlan> read_plan =
std::make_unique<FullMapReadPlan>(
+ logical_map_field, physical_read_field, ResolveSelectedKeyIds(meta,
selected_keys));
+ return read_plan;
+}
+
+Result<std::unique_ptr<MapFieldReadPlan>>
MapFieldReadPlanFactory::CreateSharedSelectedKeysReadPlan(
+ const std::shared_ptr<arrow::Field>& selected_keys_field,
+ const MapSharedShreddingFieldMeta& meta, const std::vector<std::string>&
selected_keys) {
+ if (selected_keys_field->type()->id() != arrow::Type::STRUCT) {
+ return Status::Invalid(
+ fmt::format("selected-key MAP field {} is not a STRUCT",
selected_keys_field->name()));
+ }
+ auto selected_keys_type =
+
arrow::internal::checked_pointer_cast<arrow::StructType>(selected_keys_field->type());
+ if (selected_keys_type->num_fields() == 0 ||
+ selected_keys.size() !=
static_cast<size_t>(selected_keys_type->num_fields())) {
+ return Status::Invalid(fmt::format(
+ "selected-key metadata size {} does not match STRUCT field count
{} for {}",
+ selected_keys.size(), selected_keys_type->num_fields(),
selected_keys_field->name()));
+ }
+ const auto& value_field = selected_keys_type->field(0);
+ for (int32_t i = 1; i < selected_keys_type->num_fields(); ++i) {
+ if
(!selected_keys_type->field(i)->type()->Equals(value_field->type())) {
+ return Status::Invalid(fmt::format(
+ "selected-key MAP fields must have the same value type, but {}
and {} differ",
+ value_field->type()->ToString(),
selected_keys_type->field(i)->type()->ToString()));
+ }
+ }
+
Review Comment:
Why does `selected_keys` need to be passed in as a parameter? My
understanding is that this information should already be contained in
`selected_keys_field`.
##########
src/paimon/common/data/shredding/map_shared_shredding_file_reader.cpp:
##########
@@ -34,14 +35,254 @@
#include "paimon/core/casting/casting_utils.h"
namespace paimon {
+namespace {
+
+Result<std::string_view> GetStringMapKey(const std::shared_ptr<arrow::Array>&
keys, int64_t index) {
+ if (keys->IsNull(index)) {
+ return Status::Invalid("selected-key MAP read found a null MAP key");
+ }
+ if (keys->type_id() == arrow::Type::STRING) {
+ return
arrow::internal::checked_pointer_cast<arrow::StringArray>(keys)->GetView(index);
+ } else if (keys->type_id() == arrow::Type::DICTIONARY) {
+ auto dictionary =
arrow::internal::checked_pointer_cast<arrow::DictionaryArray>(keys);
+ int64_t dictionary_index = dictionary->GetValueIndex(index);
+ const auto& values = dictionary->dictionary();
+ if (values->IsNull(dictionary_index)) {
+ return Status::Invalid("selected-key MAP read found a null
dictionary MAP key");
+ }
+ if (values->type_id() == arrow::Type::STRING) {
+ return
arrow::internal::checked_pointer_cast<arrow::StringArray>(values)->GetView(
+ dictionary_index);
+ }
+ }
+ return Status::Invalid(
+ fmt::format("selected-key MAP read only supports string or dictionary
key array"));
+}
Review Comment:
This function is very similar to `GetMapKeyViewAt` in
`nested_projection_utils.cpp`. Could we refactor this a bit and reuse the
existing logic?
##########
src/paimon/common/data/shredding/map_shared_shredding_file_reader.cpp:
##########
@@ -34,14 +35,254 @@
#include "paimon/core/casting/casting_utils.h"
namespace paimon {
+namespace {
+
+Result<std::string_view> GetStringMapKey(const std::shared_ptr<arrow::Array>&
keys, int64_t index) {
+ if (keys->IsNull(index)) {
+ return Status::Invalid("selected-key MAP read found a null MAP key");
+ }
+ if (keys->type_id() == arrow::Type::STRING) {
+ return
arrow::internal::checked_pointer_cast<arrow::StringArray>(keys)->GetView(index);
+ } else if (keys->type_id() == arrow::Type::DICTIONARY) {
+ auto dictionary =
arrow::internal::checked_pointer_cast<arrow::DictionaryArray>(keys);
+ int64_t dictionary_index = dictionary->GetValueIndex(index);
+ const auto& values = dictionary->dictionary();
+ if (values->IsNull(dictionary_index)) {
+ return Status::Invalid("selected-key MAP read found a null
dictionary MAP key");
+ }
+ if (values->type_id() == arrow::Type::STRING) {
+ return
arrow::internal::checked_pointer_cast<arrow::StringArray>(values)->GetView(
+ dictionary_index);
+ }
+ }
+ return Status::Invalid(
+ fmt::format("selected-key MAP read only supports string or dictionary
key array"));
+}
+
+std::vector<std::pair<std::string, int32_t>> ResolveSelectedKeyIds(
+ const MapSharedShreddingFieldMeta& meta, const std::vector<std::string>&
selected_keys) {
+ std::vector<std::pair<std::string, int32_t>> selected_key_ids;
+ selected_key_ids.reserve(selected_keys.size());
+ for (const auto& selected_key : selected_keys) {
+ auto id_iter = meta.name_to_id.find(selected_key);
+ if (id_iter != meta.name_to_id.end()) {
+ selected_key_ids.emplace_back(selected_key, id_iter->second);
+ }
+ }
+ return selected_key_ids;
+}
+
+void CollectPhysicalColumns(
+ const std::shared_ptr<arrow::StructArray>& physical_struct_array,
+ std::map<std::string, std::shared_ptr<arrow::Array>>*
physical_column_name_to_array,
+ std::shared_ptr<arrow::MapArray>* overflow_array) {
+ const auto& struct_type = physical_struct_array->struct_type();
+ for (int32_t i = 0; i < struct_type->num_fields(); ++i) {
+ const auto& sub_field = struct_type->field(i);
+ if (sub_field->name() == MapSharedShreddingDefine::kFieldMapping) {
+ continue;
+ }
+ if (sub_field->name() == MapSharedShreddingDefine::kOverflow) {
+ *overflow_array =
arrow::internal::checked_pointer_cast<arrow::MapArray>(
+ physical_struct_array->field(i));
+ continue;
+ }
+ (*physical_column_name_to_array)[sub_field->name()] =
physical_struct_array->field(i);
+ }
+}
+
+class FullMapReadPlan : public MapFieldReadPlan {
+ public:
+ FullMapReadPlan(const std::shared_ptr<arrow::Field>& logical_field,
+ const std::shared_ptr<arrow::Field>& physical_read_field,
+ std::vector<std::pair<std::string, int32_t>>&&
selected_key_ids)
+ : MapFieldReadPlan(logical_field, physical_read_field),
+ selected_key_ids_(std::move(selected_key_ids)),
+ logical_map_type_(
+
arrow::internal::checked_pointer_cast<arrow::MapType>(logical_field->type())) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<std::pair<std::string, int32_t>> selected_key_ids_;
+ std::shared_ptr<arrow::MapType> logical_map_type_;
+};
+
+class SharedSelectedKeysReadPlan : public MapFieldReadPlan {
+ public:
+ struct SelectedKey {
+ int32_t field_id = -1;
+ std::vector<int32_t> candidate_columns;
+ bool may_use_overflow = false;
+ };
+
+ SharedSelectedKeysReadPlan(const std::shared_ptr<arrow::Field>&
logical_field,
+ const std::shared_ptr<arrow::Field>&
physical_read_field,
+ std::vector<SelectedKey>&& selected_keys)
+ : MapFieldReadPlan(logical_field, physical_read_field),
+ selected_keys_(std::move(selected_keys)) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<SelectedKey> selected_keys_;
+};
+
+class DefaultSelectedKeysReadPlan : public MapFieldReadPlan {
+ public:
+ DefaultSelectedKeysReadPlan(const std::shared_ptr<arrow::Field>&
logical_field,
+ const std::shared_ptr<arrow::Field>&
physical_read_field,
+ const std::vector<std::string>& selected_keys)
+ : MapFieldReadPlan(logical_field, physical_read_field),
selected_keys_(selected_keys) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<std::string> selected_keys_;
+};
+
+} // namespace
+
+Result<std::unique_ptr<MapFieldReadPlan>>
MapFieldReadPlanFactory::CreateFullMapReadPlan(
+ const std::shared_ptr<arrow::Field>& logical_map_field, const
MapSharedShreddingFieldMeta& meta,
+ const std::vector<std::string>& selected_keys) {
+ if (logical_map_field->type()->id() != arrow::Type::MAP) {
Review Comment:
Why does fullmap support `selected_keys`? Does Java have similar
functionality as well?
##########
src/paimon/common/data/shredding/map_shared_shredding_file_reader_test.cpp:
##########
@@ -299,6 +300,94 @@ TEST_F(MapSharedShreddingFileReaderTest,
TestAllExistSelectedKeysWithOverflow) {
AssertChunkedArrayEquals(expected, actual);
}
+TEST_F(MapSharedShreddingFileReaderTest, TestSelectedKeysStructProjection) {
+ ASSERT_OK_AND_ASSIGN(auto physical_schema, PhysicalSchemaWithMetadata());
+ ASSERT_OK_AND_ASSIGN(auto physical_array, PhysicalArray());
+ auto mock_reader = std::make_unique<MockFileBatchReader>(
+ physical_array, arrow::struct_(physical_schema->fields()),
/*read_batch_size=*/10);
+ mock_reader->EnableRandomizeBatchSize(false);
+
+ auto selected_type =
+ arrow::struct_({arrow::field("0", arrow::int64()), arrow::field("1",
arrow::int64()),
+ arrow::field("2", arrow::int64())});
+ auto selected_field = arrow::field(
Review Comment:
Could we use field names like `a, c, and missing` in `selected_type` here?
Using `0, 1, and 2` is a bit hard to follow.
##########
src/paimon/common/data/shredding/map_shared_shredding_schema_utils.cpp:
##########
@@ -19,15 +19,92 @@
#include "paimon/data/shredding/map_shared_shredding_schema_utils.h"
+#include <unordered_set>
+#include <utility>
+#include <vector>
+
#include "arrow/c/bridge.h"
#include "arrow/type.h"
#include "arrow/util/key_value_metadata.h"
#include "fmt/format.h"
#include "paimon/common/data/shredding/map_shared_shredding_utils.h"
+#include "paimon/common/types/data_field.h"
#include "paimon/common/utils/arrow/status_utils.h"
namespace paimon {
+class MapSharedShreddingAccessBuilder::Impl {
+ public:
+ Impl(const std::shared_ptr<arrow::Field>& _map_field,
+ const std::shared_ptr<arrow::MapType>& _map_type)
+ : map_field(_map_field), map_type(_map_type) {}
+
+ std::shared_ptr<arrow::Field> map_field;
+ std::shared_ptr<arrow::MapType> map_type;
+ std::vector<std::string> keys;
+ std::unordered_set<std::string> unique_keys;
+};
+
+MapSharedShreddingAccessBuilder::~MapSharedShreddingAccessBuilder() = default;
+
+MapSharedShreddingAccessBuilder::MapSharedShreddingAccessBuilder(std::unique_ptr<Impl>&&
impl)
+ : impl_(std::move(impl)) {}
+
+Result<std::unique_ptr<MapSharedShreddingAccessBuilder>>
MapSharedShreddingAccessBuilder::Create(
+ struct ArrowSchema* map_field) {
+ if (!map_field) {
+ return Status::Invalid("MAP field is null");
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Field> field,
+ arrow::ImportField(map_field));
+ if (field->type()->id() != arrow::Type::MAP) {
+ return Status::Invalid(
+ fmt::format("MapSharedShreddingAccessBuilder requires MAP field,
got {}",
+ field->type()->ToString()));
+ }
+ auto map_type =
arrow::internal::checked_pointer_cast<arrow::MapType>(field->type());
+ if (map_type->key_type()->id() != arrow::Type::STRING) {
+ return Status::Invalid(fmt::format(
+ "MapSharedShreddingAccessBuilder only supports MAP with STRING
keys, got {}",
+ map_type->key_type()->ToString()));
+ }
+ auto impl = std::make_unique<Impl>(field, map_type);
+ return std::unique_ptr<MapSharedShreddingAccessBuilder>(
+ new MapSharedShreddingAccessBuilder(std::move(impl)));
+}
+
+Status MapSharedShreddingAccessBuilder::AddKey(const std::string& key) {
+ if (!impl_->unique_keys.insert(key).second) {
+ return Status::Invalid(fmt::format("selected MAP key must not be
duplicated: {}", key));
+ }
+ impl_->keys.push_back(key);
+ return Status::OK();
+}
+
+Result<std::unique_ptr<struct ArrowSchema>>
MapSharedShreddingAccessBuilder::Build() const {
+ if (impl_->keys.empty()) {
+ return Status::Invalid(
+ "shared shredding MAP selected-key projection needs at least one
key");
+ }
+ arrow::FieldVector fields;
+ fields.reserve(impl_->keys.size());
+ std::string encoded_keys;
+ for (size_t i = 0; i < impl_->keys.size(); ++i) {
+ if (i != 0) {
+ encoded_keys.push_back(',');
+ }
+ encoded_keys.append(impl_->keys[i]);
+ fields.push_back(arrow::field(std::to_string(i),
impl_->map_type->item_type(),
+ /*nullable=*/true));
Review Comment:
`nullable` must be true?
##########
src/paimon/common/data/shredding/map_shared_shredding_file_reader.cpp:
##########
@@ -353,37 +546,196 @@ Result<std::shared_ptr<arrow::Array>>
MapSharedShreddingFileReader::RebuildLogic
return map_array;
}
-std::vector<std::pair<std::string, int32_t>>
MapSharedShreddingFileReader::ResolveSelectedKeyIds(
- const MapSharedShreddingFieldMeta& meta, const std::vector<std::string>&
selected_keys) {
- std::vector<std::pair<std::string, int32_t>> selected_key_ids;
- selected_key_ids.reserve(selected_keys.size());
- for (const auto& selected_key : selected_keys) {
- auto id_iter = meta.name_to_id.find(selected_key);
- if (id_iter == meta.name_to_id.end()) {
+Result<std::shared_ptr<arrow::Array>> SharedSelectedKeysReadPlan::Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array, arrow::MemoryPool*
arrow_pool) const {
+ auto physical_struct_array =
+
arrow::internal::checked_pointer_cast<arrow::StructArray>(physical_array);
+ if (!physical_struct_array) {
+ return Status::Invalid(fmt::format("cannot cast physical shredding
field {} to StructArray",
+ LogicalField()->name()));
+ }
+ auto selected_keys_type =
+
arrow::internal::checked_pointer_cast<arrow::StructType>(LogicalField()->type());
+
+ auto field_mapping_array =
arrow::internal::checked_pointer_cast<arrow::ListArray>(
+
physical_struct_array->GetFieldByName(MapSharedShreddingDefine::kFieldMapping));
+ if (!field_mapping_array) {
+ return Status::Invalid(
+ fmt::format("cannot find __field_mapping for field {}",
LogicalField()->name()));
+ }
+ auto field_mapping_values =
+
arrow::internal::checked_pointer_cast<arrow::Int32Array>(field_mapping_array->values());
+ if (!field_mapping_values) {
+ return Status::Invalid("__field_mapping values is not an Int32Array");
+ }
+
+ std::shared_ptr<arrow::DataType> value_type =
selected_keys_type->field(0)->type();
+ std::map<std::string, std::shared_ptr<arrow::Array>>
physical_column_name_to_array;
+ 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));
+ }
+ }
+
+ std::shared_ptr<arrow::Int32Array> overflow_keys;
+ std::shared_ptr<arrow::Array> overflow_items;
+ if (overflow_array) {
+ overflow_keys =
+
arrow::internal::checked_pointer_cast<arrow::Int32Array>(overflow_array->keys());
+ overflow_items = overflow_array->items();
+ if (!overflow_keys || !overflow_items) {
+ return Status::Invalid("__overflow map has invalid key or item
array");
+ }
+ 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));
+ }
+ }
+
+ std::unique_ptr<arrow::ArrayBuilder> access_builder_base;
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(access_builder_base,
+
arrow::MakeBuilder(LogicalField()->type(), arrow_pool));
+ auto* access_builder =
dynamic_cast<arrow::StructBuilder*>(access_builder_base.get());
+ if (!access_builder) {
+ return Status::Invalid(
+ fmt::format("selected-key MAP field {} is not a STRUCT",
LogicalField()->name()));
+ }
+
PAIMON_RETURN_NOT_OK_FROM_ARROW(access_builder->Reserve(physical_struct_array->length()));
+
Review Comment:
As a follow-up, we could consider short-circuiting to a shallow copy when a
given key exists in only one column.
##########
src/paimon/common/data/shredding/map_shared_shredding_file_reader.cpp:
##########
@@ -34,14 +35,254 @@
#include "paimon/core/casting/casting_utils.h"
namespace paimon {
+namespace {
+
+Result<std::string_view> GetStringMapKey(const std::shared_ptr<arrow::Array>&
keys, int64_t index) {
+ if (keys->IsNull(index)) {
+ return Status::Invalid("selected-key MAP read found a null MAP key");
+ }
+ if (keys->type_id() == arrow::Type::STRING) {
+ return
arrow::internal::checked_pointer_cast<arrow::StringArray>(keys)->GetView(index);
+ } else if (keys->type_id() == arrow::Type::DICTIONARY) {
+ auto dictionary =
arrow::internal::checked_pointer_cast<arrow::DictionaryArray>(keys);
+ int64_t dictionary_index = dictionary->GetValueIndex(index);
+ const auto& values = dictionary->dictionary();
+ if (values->IsNull(dictionary_index)) {
+ return Status::Invalid("selected-key MAP read found a null
dictionary MAP key");
+ }
+ if (values->type_id() == arrow::Type::STRING) {
+ return
arrow::internal::checked_pointer_cast<arrow::StringArray>(values)->GetView(
+ dictionary_index);
+ }
+ }
+ return Status::Invalid(
+ fmt::format("selected-key MAP read only supports string or dictionary
key array"));
+}
+
+std::vector<std::pair<std::string, int32_t>> ResolveSelectedKeyIds(
+ const MapSharedShreddingFieldMeta& meta, const std::vector<std::string>&
selected_keys) {
+ std::vector<std::pair<std::string, int32_t>> selected_key_ids;
+ selected_key_ids.reserve(selected_keys.size());
+ for (const auto& selected_key : selected_keys) {
+ auto id_iter = meta.name_to_id.find(selected_key);
+ if (id_iter != meta.name_to_id.end()) {
+ selected_key_ids.emplace_back(selected_key, id_iter->second);
+ }
+ }
+ return selected_key_ids;
+}
+
+void CollectPhysicalColumns(
+ const std::shared_ptr<arrow::StructArray>& physical_struct_array,
+ std::map<std::string, std::shared_ptr<arrow::Array>>*
physical_column_name_to_array,
+ std::shared_ptr<arrow::MapArray>* overflow_array) {
+ const auto& struct_type = physical_struct_array->struct_type();
+ for (int32_t i = 0; i < struct_type->num_fields(); ++i) {
+ const auto& sub_field = struct_type->field(i);
+ if (sub_field->name() == MapSharedShreddingDefine::kFieldMapping) {
+ continue;
+ }
+ if (sub_field->name() == MapSharedShreddingDefine::kOverflow) {
+ *overflow_array =
arrow::internal::checked_pointer_cast<arrow::MapArray>(
+ physical_struct_array->field(i));
+ continue;
+ }
+ (*physical_column_name_to_array)[sub_field->name()] =
physical_struct_array->field(i);
+ }
+}
+
+class FullMapReadPlan : public MapFieldReadPlan {
+ public:
+ FullMapReadPlan(const std::shared_ptr<arrow::Field>& logical_field,
+ const std::shared_ptr<arrow::Field>& physical_read_field,
+ std::vector<std::pair<std::string, int32_t>>&&
selected_key_ids)
+ : MapFieldReadPlan(logical_field, physical_read_field),
+ selected_key_ids_(std::move(selected_key_ids)),
+ logical_map_type_(
+
arrow::internal::checked_pointer_cast<arrow::MapType>(logical_field->type())) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<std::pair<std::string, int32_t>> selected_key_ids_;
+ std::shared_ptr<arrow::MapType> logical_map_type_;
+};
+
+class SharedSelectedKeysReadPlan : public MapFieldReadPlan {
+ public:
+ struct SelectedKey {
+ int32_t field_id = -1;
+ std::vector<int32_t> candidate_columns;
+ bool may_use_overflow = false;
+ };
+
+ SharedSelectedKeysReadPlan(const std::shared_ptr<arrow::Field>&
logical_field,
+ const std::shared_ptr<arrow::Field>&
physical_read_field,
+ std::vector<SelectedKey>&& selected_keys)
+ : MapFieldReadPlan(logical_field, physical_read_field),
+ selected_keys_(std::move(selected_keys)) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<SelectedKey> selected_keys_;
+};
+
+class DefaultSelectedKeysReadPlan : public MapFieldReadPlan {
+ public:
+ DefaultSelectedKeysReadPlan(const std::shared_ptr<arrow::Field>&
logical_field,
+ const std::shared_ptr<arrow::Field>&
physical_read_field,
+ const std::vector<std::string>& selected_keys)
+ : MapFieldReadPlan(logical_field, physical_read_field),
selected_keys_(selected_keys) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<std::string> selected_keys_;
+};
+
+} // namespace
+
+Result<std::unique_ptr<MapFieldReadPlan>>
MapFieldReadPlanFactory::CreateFullMapReadPlan(
+ const std::shared_ptr<arrow::Field>& logical_map_field, const
MapSharedShreddingFieldMeta& meta,
+ const std::vector<std::string>& selected_keys) {
+ if (logical_map_field->type()->id() != arrow::Type::MAP) {
Review Comment:
Previously, tests for reading back only a subset of sub-keys should all have
gone through `CreateSharedSelectedKeysReadPlan`. It now seems they are still
going through `CreateFullMapReadPlan` instead?
##########
src/paimon/core/operation/internal_read_context.cpp:
##########
@@ -50,6 +51,37 @@ Result<std::shared_ptr<arrow::Field>>
InternalReadContext::AlignReadFieldWithTab
return table_field->WithType(read_field->type());
}
+ if (table_field->type()->id() == arrow::Type::MAP &&
+ NestedProjectionUtils::IsMapSharedShreddingAccessField(read_field)) {
+ auto table_map =
arrow::internal::checked_pointer_cast<arrow::MapType>(table_field->type());
+ if (table_map->key_type()->id() != arrow::Type::STRING) {
+ return Status::Invalid(fmt::format(
+ "Selected-key MAP pushdown only supports string MAP keys for
field '{}'",
+ table_field->name()));
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> selected_keys,
+
NestedProjectionUtils::GetMapSelectedKeys(read_field));
+ auto read_struct =
+
arrow::internal::checked_pointer_cast<arrow::StructType>(read_field->type());
+ if (selected_keys.size() !=
static_cast<size_t>(read_struct->num_fields())) {
+ return Status::Invalid(fmt::format(
+ "Selected-key metadata size {} does not match STRUCT field
count {} for '{}'",
+ selected_keys.size(), read_struct->num_fields(),
table_field->name()));
+ }
+ for (const auto& selected_field : read_struct->fields()) {
+ if (!selected_field->type()->Equals(table_map->item_type())) {
+ return Status::Invalid(fmt::format(
+ "Selected-key MAP pushdown does not support pruning MAP
value fields for "
+ "'{}': selected type {} vs MAP value type {}",
+ table_field->name(), selected_field->type()->ToString(),
+ table_map->item_type()->ToString()));
+ }
+ }
+ auto aligned_field = table_field->WithType(read_field->type());
Review Comment:
These validations are repeated several times in both
`internal_read_context.cpp` and `map_shared_shredding_file_reader.cpp`. Could
we either factor them into a helper function or keep the checks in just one
place?
##########
src/paimon/common/data/shredding/map_shared_shredding_file_reader_test.cpp:
##########
@@ -299,6 +300,94 @@ TEST_F(MapSharedShreddingFileReaderTest,
TestAllExistSelectedKeysWithOverflow) {
AssertChunkedArrayEquals(expected, actual);
}
+TEST_F(MapSharedShreddingFileReaderTest, TestSelectedKeysStructProjection) {
+ ASSERT_OK_AND_ASSIGN(auto physical_schema, PhysicalSchemaWithMetadata());
+ ASSERT_OK_AND_ASSIGN(auto physical_array, PhysicalArray());
+ auto mock_reader = std::make_unique<MockFileBatchReader>(
+ physical_array, arrow::struct_(physical_schema->fields()),
/*read_batch_size=*/10);
+ mock_reader->EnableRandomizeBatchSize(false);
+
+ auto selected_type =
+ arrow::struct_({arrow::field("0", arrow::int64()), arrow::field("1",
arrow::int64()),
+ arrow::field("2", arrow::int64())});
+ auto selected_field = arrow::field(
+ "tags", selected_type, /*nullable=*/true,
+ arrow::KeyValueMetadata::Make({DataField::MAP_SELECTED_KEYS},
{"a,c,missing"}));
+ ASSERT_OK_AND_ASSIGN(auto field_read_plan,
+
MapFieldReadPlanFactory::CreateSharedSelectedKeysReadPlan(
+ selected_field, TagsMeta(), {"a", "c",
"missing"}));
+ std::map<std::string, std::unique_ptr<MapFieldReadPlan>> contexts;
+ contexts.emplace("tags", std::move(field_read_plan));
+ auto reader =
std::make_unique<MapSharedShreddingFileReader>(std::move(mock_reader),
+
std::move(contexts), pool_);
+
+ auto read_schema =
+ ExportSchema(arrow::schema({arrow::field("id", arrow::int32()),
selected_field}));
+ ASSERT_OK(reader->SetReadSchema(read_schema.get(), /*predicate=*/nullptr,
+ /*selection_bitmap=*/std::nullopt));
+ ASSERT_OK_AND_ASSIGN(auto actual,
ReadResultCollector::CollectResult(reader.get()));
+
+ auto expected_type = arrow::struct_({arrow::field("id", arrow::int32()),
selected_field});
+ std::shared_ptr<arrow::ChunkedArray> expected;
+
ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON(expected_type,
{R"([
+ [1, [10, null, null]],
+ [2, [40, 30, null]],
+ [3, null],
+ [4, [80, null, null]]
+ ])"},
+ &expected)
+ .ok());
+ AssertChunkedArrayEquals(expected, actual);
+}
+
+TEST_F(MapSharedShreddingFileReaderTest,
TestSelectedKeysStructProjectionFromLegacyMap) {
Review Comment:
FromLegacyMap -> FromDefaultMap
##########
src/paimon/core/io/field_mapping_reader.cpp:
##########
@@ -75,6 +87,11 @@ Result<std::shared_ptr<arrow::Array>>
FieldMappingReader::FilterMapSelectedKeysR
}
auto type_id = read_field->type()->id();
+ if (NestedProjectionUtils::IsMapSharedShreddingAccessField(read_field)) {
+ // The shared-shredding wrapper (including its legacy MAP fallback)
has already
+ // materialized this projection as a STRUCT.
Review Comment:
`legacy` and `default` seem to be used interchangeably in the code. Could we
unify the terminology?
##########
test/inte/write_and_read_inte_test.cpp:
##########
@@ -3556,6 +3586,30 @@ TEST_P(WriteAndReadInteTest,
TestMapStorageLayoutDefaultToSharedShreddingPartial
[0, 4, []]
])"));
ASSERT_TRUE(success);
+
+ auto c_map_field = std::make_unique<ArrowSchema>();
+ ASSERT_TRUE(arrow::ExportField(*arrow::field("tags", map_type),
c_map_field.get()).ok());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<MapSharedShreddingAccessBuilder>
access_builder,
+
MapSharedShreddingAccessBuilder::Create(c_map_field.get()));
+ ASSERT_OK(access_builder->AddKey("a"));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ArrowSchema> c_access_field,
access_builder->Build());
+ auto imported_access_field = arrow::ImportField(c_access_field.get());
+ ASSERT_TRUE(imported_access_field.ok());
+ std::shared_ptr<arrow::Field> access_field =
imported_access_field.ValueOrDie();
+ read_schema = arrow::schema({arrow::field("id", arrow::int32()),
access_field});
+ expected_type = arrow::struct_({
+ arrow::field("_VALUE_KIND", arrow::int8()),
+ arrow::field("id", arrow::int32()),
+ access_field,
+ });
+ ASSERT_OK_AND_ASSIGN(success, ReadAndCheckWithReadSchema(options_v1,
read_schema, expected_type,
+ R"([
+ [0, 1, [10]],
+ [0, 2, null],
+ [0, 3, [30]],
+ [0, 4, [null]]
+ ])"));
+ ASSERT_TRUE(success);
}
Review Comment:
Could you add similar tests with struct selected-key read scenarios (like
previous map selected-key), especially where the map value is a nested type or
with dictionary?
##########
src/paimon/core/utils/nested_projection_utils.h:
##########
@@ -79,6 +79,16 @@ class PAIMON_EXPORT NestedProjectionUtils {
static Result<std::vector<std::string>> GetMapSelectedKeys(
const std::shared_ptr<arrow::Field>& field);
+ /// @return true when `field` is a selected-key MAP projection: a STRUCT
carrying
+ /// `paimon.map.selected-keys` metadata.
+ static bool IsMapSharedShreddingAccessField(const
std::shared_ptr<arrow::Field>& field);
+
+ /// Rewrites a selected-key STRUCT projection to use the data file MAP
value type for every
+ /// child. This mirrors schema evolution mapping before the selected
values are materialized.
+ static Result<std::shared_ptr<arrow::DataType>>
BuildMapSharedShreddingAccessDataType(
Review Comment:
Do we support schema evolution within map? If not, should we add a comment
here to make that explicit?
##########
src/paimon/common/data/shredding/map_shared_shredding_file_reader.cpp:
##########
@@ -34,14 +35,254 @@
#include "paimon/core/casting/casting_utils.h"
namespace paimon {
+namespace {
+
+Result<std::string_view> GetStringMapKey(const std::shared_ptr<arrow::Array>&
keys, int64_t index) {
+ if (keys->IsNull(index)) {
+ return Status::Invalid("selected-key MAP read found a null MAP key");
+ }
+ if (keys->type_id() == arrow::Type::STRING) {
+ return
arrow::internal::checked_pointer_cast<arrow::StringArray>(keys)->GetView(index);
+ } else if (keys->type_id() == arrow::Type::DICTIONARY) {
+ auto dictionary =
arrow::internal::checked_pointer_cast<arrow::DictionaryArray>(keys);
+ int64_t dictionary_index = dictionary->GetValueIndex(index);
+ const auto& values = dictionary->dictionary();
+ if (values->IsNull(dictionary_index)) {
+ return Status::Invalid("selected-key MAP read found a null
dictionary MAP key");
+ }
+ if (values->type_id() == arrow::Type::STRING) {
+ return
arrow::internal::checked_pointer_cast<arrow::StringArray>(values)->GetView(
+ dictionary_index);
+ }
+ }
+ return Status::Invalid(
+ fmt::format("selected-key MAP read only supports string or dictionary
key array"));
+}
+
+std::vector<std::pair<std::string, int32_t>> ResolveSelectedKeyIds(
+ const MapSharedShreddingFieldMeta& meta, const std::vector<std::string>&
selected_keys) {
+ std::vector<std::pair<std::string, int32_t>> selected_key_ids;
+ selected_key_ids.reserve(selected_keys.size());
+ for (const auto& selected_key : selected_keys) {
+ auto id_iter = meta.name_to_id.find(selected_key);
+ if (id_iter != meta.name_to_id.end()) {
+ selected_key_ids.emplace_back(selected_key, id_iter->second);
+ }
+ }
+ return selected_key_ids;
+}
+
+void CollectPhysicalColumns(
+ const std::shared_ptr<arrow::StructArray>& physical_struct_array,
+ std::map<std::string, std::shared_ptr<arrow::Array>>*
physical_column_name_to_array,
+ std::shared_ptr<arrow::MapArray>* overflow_array) {
+ const auto& struct_type = physical_struct_array->struct_type();
+ for (int32_t i = 0; i < struct_type->num_fields(); ++i) {
+ const auto& sub_field = struct_type->field(i);
+ if (sub_field->name() == MapSharedShreddingDefine::kFieldMapping) {
+ continue;
+ }
+ if (sub_field->name() == MapSharedShreddingDefine::kOverflow) {
+ *overflow_array =
arrow::internal::checked_pointer_cast<arrow::MapArray>(
+ physical_struct_array->field(i));
+ continue;
+ }
+ (*physical_column_name_to_array)[sub_field->name()] =
physical_struct_array->field(i);
+ }
+}
+
+class FullMapReadPlan : public MapFieldReadPlan {
+ public:
+ FullMapReadPlan(const std::shared_ptr<arrow::Field>& logical_field,
+ const std::shared_ptr<arrow::Field>& physical_read_field,
+ std::vector<std::pair<std::string, int32_t>>&&
selected_key_ids)
+ : MapFieldReadPlan(logical_field, physical_read_field),
+ selected_key_ids_(std::move(selected_key_ids)),
+ logical_map_type_(
+
arrow::internal::checked_pointer_cast<arrow::MapType>(logical_field->type())) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<std::pair<std::string, int32_t>> selected_key_ids_;
+ std::shared_ptr<arrow::MapType> logical_map_type_;
+};
+
+class SharedSelectedKeysReadPlan : public MapFieldReadPlan {
+ public:
+ struct SelectedKey {
+ int32_t field_id = -1;
+ std::vector<int32_t> candidate_columns;
+ bool may_use_overflow = false;
+ };
+
+ SharedSelectedKeysReadPlan(const std::shared_ptr<arrow::Field>&
logical_field,
+ const std::shared_ptr<arrow::Field>&
physical_read_field,
+ std::vector<SelectedKey>&& selected_keys)
+ : MapFieldReadPlan(logical_field, physical_read_field),
+ selected_keys_(std::move(selected_keys)) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<SelectedKey> selected_keys_;
+};
+
+class DefaultSelectedKeysReadPlan : public MapFieldReadPlan {
+ public:
+ DefaultSelectedKeysReadPlan(const std::shared_ptr<arrow::Field>&
logical_field,
+ const std::shared_ptr<arrow::Field>&
physical_read_field,
+ const std::vector<std::string>& selected_keys)
+ : MapFieldReadPlan(logical_field, physical_read_field),
selected_keys_(selected_keys) {}
+
+ Result<std::shared_ptr<arrow::Array>> Materialize(
+ const std::shared_ptr<arrow::Array>& physical_array,
+ arrow::MemoryPool* arrow_pool) const override;
+
+ private:
+ std::vector<std::string> selected_keys_;
+};
+
+} // namespace
+
+Result<std::unique_ptr<MapFieldReadPlan>>
MapFieldReadPlanFactory::CreateFullMapReadPlan(
+ const std::shared_ptr<arrow::Field>& logical_map_field, const
MapSharedShreddingFieldMeta& meta,
+ const std::vector<std::string>& selected_keys) {
+ if (logical_map_field->type()->id() != arrow::Type::MAP) {
+ return Status::Invalid(fmt::format("full MAP read plan requires MAP
field {}, got {}",
+ logical_map_field->name(),
+
logical_map_field->type()->ToString()));
+ }
+ auto logical_map_type =
+
arrow::internal::checked_pointer_cast<arrow::MapType>(logical_map_field->type());
+ std::set<int32_t> selected_physical_column_ids;
+ bool include_overflow = false;
+ for (const auto& selected_key : selected_keys) {
+ auto field_id_iter = meta.name_to_id.find(selected_key);
+ if (field_id_iter == meta.name_to_id.end()) {
+ continue;
+ }
+ int32_t field_id = field_id_iter->second;
+ include_overflow = include_overflow ||
meta.overflow_field_set.count(field_id) > 0;
+ auto columns_iter = meta.field_to_columns.find(field_id);
+ if (columns_iter != meta.field_to_columns.end()) {
+ selected_physical_column_ids.insert(columns_iter->second.begin(),
+ columns_iter->second.end());
+ }
+ }
+ std::shared_ptr<arrow::DataType> physical_type =
+ MapSharedShreddingUtils::BuildSpecificPhysicalStructType(
+ logical_map_type->item_type(), selected_physical_column_ids,
+ logical_map_type->item_field()->nullable(), include_overflow);
+ auto physical_read_field = logical_map_field->WithType(physical_type);
+ std::unique_ptr<MapFieldReadPlan> read_plan =
std::make_unique<FullMapReadPlan>(
+ logical_map_field, physical_read_field, ResolveSelectedKeyIds(meta,
selected_keys));
+ return read_plan;
+}
+
+Result<std::unique_ptr<MapFieldReadPlan>>
MapFieldReadPlanFactory::CreateSharedSelectedKeysReadPlan(
+ const std::shared_ptr<arrow::Field>& selected_keys_field,
+ const MapSharedShreddingFieldMeta& meta, const std::vector<std::string>&
selected_keys) {
+ if (selected_keys_field->type()->id() != arrow::Type::STRUCT) {
+ return Status::Invalid(
+ fmt::format("selected-key MAP field {} is not a STRUCT",
selected_keys_field->name()));
+ }
+ auto selected_keys_type =
+
arrow::internal::checked_pointer_cast<arrow::StructType>(selected_keys_field->type());
+ if (selected_keys_type->num_fields() == 0 ||
+ selected_keys.size() !=
static_cast<size_t>(selected_keys_type->num_fields())) {
+ return Status::Invalid(fmt::format(
+ "selected-key metadata size {} does not match STRUCT field count
{} for {}",
+ selected_keys.size(), selected_keys_type->num_fields(),
selected_keys_field->name()));
+ }
+ const auto& value_field = selected_keys_type->field(0);
+ for (int32_t i = 1; i < selected_keys_type->num_fields(); ++i) {
+ if
(!selected_keys_type->field(i)->type()->Equals(value_field->type())) {
+ return Status::Invalid(fmt::format(
+ "selected-key MAP fields must have the same value type, but {}
and {} differ",
+ value_field->type()->ToString(),
selected_keys_type->field(i)->type()->ToString()));
+ }
+ }
+
+ std::set<int32_t> selected_physical_column_ids;
+ bool include_overflow = false;
+ std::vector<SharedSelectedKeysReadPlan::SelectedKey> selected_key_plans;
+ selected_key_plans.reserve(selected_keys.size());
+ for (const auto& selected_key : selected_keys) {
+ SharedSelectedKeysReadPlan::SelectedKey selected_key_plan;
+ auto field_id_iter = meta.name_to_id.find(selected_key);
+ if (field_id_iter != meta.name_to_id.end()) {
+ selected_key_plan.field_id = field_id_iter->second;
+ auto columns_iter =
meta.field_to_columns.find(selected_key_plan.field_id);
+ if (columns_iter != meta.field_to_columns.end()) {
+ selected_key_plan.candidate_columns = columns_iter->second;
+
selected_physical_column_ids.insert(columns_iter->second.begin(),
+
columns_iter->second.end());
+ }
+ selected_key_plan.may_use_overflow =
+ meta.overflow_field_set.count(selected_key_plan.field_id) > 0;
+ include_overflow = include_overflow ||
selected_key_plan.may_use_overflow;
+ }
+ selected_key_plans.push_back(std::move(selected_key_plan));
+ }
+ std::shared_ptr<arrow::DataType> physical_type =
+ MapSharedShreddingUtils::BuildSpecificPhysicalStructType(
+ value_field->type(), selected_physical_column_ids,
value_field->nullable(),
+ include_overflow);
+ auto physical_read_field = selected_keys_field->WithType(physical_type);
+ std::unique_ptr<MapFieldReadPlan> read_plan =
std::make_unique<SharedSelectedKeysReadPlan>(
+ selected_keys_field, physical_read_field,
std::move(selected_key_plans));
+ return read_plan;
+}
+
+Result<std::unique_ptr<MapFieldReadPlan>>
+MapFieldReadPlanFactory::CreateDefaultSelectedKeysReadPlan(
+ const std::shared_ptr<arrow::Field>& file_map_field,
+ const std::shared_ptr<arrow::Field>& selected_keys_field,
+ const std::vector<std::string>& selected_keys) {
+ if (file_map_field->type()->id() != arrow::Type::MAP) {
+ return Status::Invalid(
+ fmt::format("selected-key MAP projection {} requires MAP file
field, "
+ "got {}",
+ selected_keys_field->name(),
file_map_field->type()->ToString()));
+ }
+ if (selected_keys_field->type()->id() != arrow::Type::STRUCT) {
+ return Status::Invalid(
+ fmt::format("selected-key MAP field {} is not a STRUCT",
selected_keys_field->name()));
+ }
+ auto selected_keys_type =
+
arrow::internal::checked_pointer_cast<arrow::StructType>(selected_keys_field->type());
+ if (selected_keys_type->num_fields() == 0 ||
+ selected_keys.size() !=
static_cast<size_t>(selected_keys_type->num_fields())) {
+ return Status::Invalid(fmt::format(
+ "selected-key metadata size {} does not match STRUCT field count
{} for {}",
+ selected_keys.size(), selected_keys_type->num_fields(),
selected_keys_field->name()));
+ }
+ const auto& value_field = selected_keys_type->field(0);
+ for (int32_t i = 1; i < selected_keys_type->num_fields(); ++i) {
+ if
(!selected_keys_type->field(i)->type()->Equals(value_field->type())) {
+ return Status::Invalid(fmt::format(
+ "selected-key MAP fields must have the same value type, but {}
and {} differ",
+ value_field->type()->ToString(),
selected_keys_type->field(i)->type()->ToString()));
+ }
+ }
Review Comment:
These validations seem to be the same as the ones in
`CreateSharedSelectedKeysReadPlan`. Could we extract them into a shared helper
function?
##########
src/paimon/core/operation/internal_read_context.cpp:
##########
@@ -198,6 +230,16 @@ Result<std::unique_ptr<InternalReadContext>>
InternalReadContext::Create(
}
PAIMON_ASSIGN_OR_RAISE(DataField table_field,
table_schema->GetField(read_field->name()));
+ if
(NestedProjectionUtils::IsMapSharedShreddingAccessField(read_field)) {
+ PAIMON_ASSIGN_OR_RAISE(MapStorageLayout layout,
+
core_options.GetMapStorageLayout(table_field.Name()));
+ if (layout != MapStorageLayout::SHARED_SHREDDING) {
+ return Status::Invalid(fmt::format(
+ "Selected-key MAP pushdown only supports top-level
shared-shredding MAP "
+ "field: {}",
+ table_field.Name()));
+ }
+ }
PAIMON_ASSIGN_OR_RAISE(
Review Comment:
Please add an invalid case for this if.
--
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]
