eldenmoon commented on code in PR #66758:
URL: https://github.com/apache/doris/pull/66758#discussion_r3793515693
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -514,6 +516,318 @@ void import_unshredded_variant_range(const
ParquetColumnSchema& schema, const Co
appender.append(std::span<const VariantRef>(encoded_rows));
}
+struct DirectResidualSeekResult {
+ ColumnPtr column;
+ int64_t rows = 0;
+ int64_t selected_value_bytes = 0;
+ int64_t container_index_builds = 0;
+ int64_t container_index_hits = 0;
+};
+
+struct DirectSeekContainerKey {
+ const char* metadata_data = nullptr;
+ size_t metadata_size = 0;
+ const char* value_data = nullptr;
+ size_t value_size = 0;
+
+ bool operator==(const DirectSeekContainerKey&) const = default;
+};
+
+struct DirectSeekContainerKeyHash {
+ size_t operator()(const DirectSeekContainerKey& key) const noexcept {
+ size_t hash = std::hash<const void*> {}(key.metadata_data);
+ auto combine = [&](size_t value) {
+ hash ^= value + 0x9e3779b97f4a7c15ULL + (hash << 6) + (hash >> 2);
+ };
+ combine(std::hash<size_t> {}(key.metadata_size));
+ combine(std::hash<const void*> {}(key.value_data));
+ combine(std::hash<size_t> {}(key.value_size));
+ return hash;
+ }
+};
+
+class DirectSeekContainerCache {
+public:
+ struct Lookup {
+ VariantContainerLookup* value;
+ bool cache_hit;
+ };
+
+ Lookup find_or_build(VariantRef value, size_t path_position,
DirectResidualSeekResult& result,
+ std::optional<VariantContainerLookup>&
uncached_lookup) {
+ const DirectSeekContainerKey key {.metadata_data = value.metadata.data,
+ .metadata_size = value.metadata.size,
+ .value_data = value.value.data,
+ .value_size = value.value.size};
+ if (auto found = _entries.find(key); found != _entries.end()) {
+ ++result.container_index_hits;
+ return {.value = &found->second, .cache_hit = true};
+ }
+ ++result.container_index_builds;
+ if (path_position >= MAX_RETAINED_PATH_DEPTH || _entries.size() >=
MAX_RETAINED_ENTRIES) {
+ // Validation is still mandatory after the high-water mark. Keep
only this traversal's
+ // lookup alive so an adversarial rows-times-depth path cannot
grow persistent state.
+ uncached_lookup.emplace(value);
+ return {.value = &*uncached_lookup, .cache_hit = false};
+ }
+ auto [inserted, was_inserted] = _entries.try_emplace(key, value);
+ DORIS_CHECK(was_inserted);
+ return {.value = &inserted->second, .cache_hit = false};
+ }
+
+ bool object_find_by_id(Lookup lookup, int64_t field_id, VariantRef* out) {
+ size_t promoted_bytes = 0;
+ const size_t promotion_budget =
+ lookup.cache_hit ? MAX_PROMOTED_OFFSET_BYTES -
_promoted_offset_bytes : 0;
+ const bool found =
+ lookup.value->object_find_by_id(field_id, out,
promotion_budget, &promoted_bytes);
+ _promoted_offset_bytes += promoted_bytes;
+ return found;
+ }
+
+ void reserve(size_t size) {
+ if (_entries.empty()) {
+ const size_t retained = size > MAX_RETAINED_ENTRIES /
MAX_RETAINED_PATH_DEPTH
+ ? MAX_RETAINED_ENTRIES
+ : size * MAX_RETAINED_PATH_DEPTH;
+ _entries.reserve(retained);
+ }
+ }
+
+ void reset() {
+ ContainerMap empty;
+ _entries.swap(empty);
+ _promoted_offset_bytes = 0;
+ }
+
+ size_t allocated_bytes() const {
+ return _entries.bucket_count() * sizeof(void*) +
+ _entries.size() * sizeof(ContainerMap::value_type) +
_promoted_offset_bytes;
+ }
+
+private:
+ // Four retained ancestors cover the common root/nested-object projections
for a default
+ // scanner batch. The independent entry and promoted-offset caps keep
state bounded for deep
+ // paths, appended columns, and noncanonical wide objects.
+ static constexpr size_t MAX_RETAINED_PATH_DEPTH = 4;
+ static constexpr size_t MAX_RETAINED_ENTRIES = 16 * 1024;
+ static constexpr size_t MAX_PROMOTED_OFFSET_BYTES = 4 * 1024 * 1024;
+
+ using ContainerMap = std::unordered_map<
+ DirectSeekContainerKey, VariantContainerLookup,
DirectSeekContainerKeyHash,
+ std::equal_to<DirectSeekContainerKey>,
+ CustomStdAllocator<std::pair<const DirectSeekContainerKey,
VariantContainerLookup>>>;
+ ContainerMap _entries;
+ size_t _promoted_offset_bytes = 0;
+};
+
+struct UnshreddedMetadataCache {
+ static constexpr uint32_t NO_METADATA =
std::numeric_limits<uint32_t>::max();
+
+ DorisVector<VariantMetadataRef> metadatas;
+ DorisVector<uint32_t> row_metadata_ids;
+
+ size_t allocated_bytes() const {
+ return metadatas.capacity() * sizeof(VariantMetadataRef) +
+ row_metadata_ids.capacity() * sizeof(uint32_t);
+ }
+};
+
+std::shared_ptr<const UnshreddedMetadataCache> build_unshredded_metadata_cache(
+ const ParquetColumnSchema& schema, const IColumn& physical, size_t
metadata_index) {
+ const auto* outer_nullable =
check_and_get_column<ColumnNullable>(physical);
+ const IColumn& wrapper =
+ outer_nullable == nullptr ? physical :
outer_nullable->get_nested_column();
+ const auto& structure = assert_cast<const ColumnStruct&>(wrapper);
+ if (structure.tuple_size() != schema.children.size()) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} physical
field count mismatch",
+ schema.name);
+ }
+
+ using MetadataIndex =
+ std::unordered_map<std::string_view, uint32_t,
std::hash<std::string_view>,
+ std::equal_to<std::string_view>,
+ CustomStdAllocator<std::pair<const
std::string_view, uint32_t>>>;
+ auto cache = std::make_shared<UnshreddedMetadataCache>();
+ cache->row_metadata_ids.resize(physical.size(),
UnshreddedMetadataCache::NO_METADATA);
+ MetadataIndex metadata_index_by_value;
+ std::optional<uint32_t> previous_metadata_id;
+
+ auto same_metadata = [&](uint32_t id, StringRef bytes) {
+ const VariantMetadataRef cached = cache->metadatas[id];
+ return StringRef(cached.data, cached.size) == bytes;
+ };
+ for (size_t row = 0; row < physical.size(); ++row) {
+ if (outer_nullable != nullptr &&
outer_nullable->get_null_map_data()[row] != 0) {
+ continue;
+ }
+ const Cell metadata_cell =
cell_at(structure.get_column(metadata_index), row);
+ if (metadata_cell.is_null) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} has
null metadata at row {}",
+ schema.name, row);
+ }
+ const StringRef metadata_bytes =
metadata_cell.column->get_data_at(row);
+ if (previous_metadata_id.has_value() &&
+ same_metadata(*previous_metadata_id, metadata_bytes)) {
+ cache->row_metadata_ids[row] = *previous_metadata_id;
+ continue;
+ }
+
+ const std::string_view key(metadata_bytes.data == nullptr ? "" :
metadata_bytes.data,
+ metadata_bytes.size);
+ if (!metadata_index_by_value.empty()) {
+ if (const auto found = metadata_index_by_value.find(key);
+ found != metadata_index_by_value.end()) {
+ cache->row_metadata_ids[row] = found->second;
+ previous_metadata_id = found->second;
+ continue;
+ }
+ }
+
+ VariantMetadataRef metadata {.data = metadata_bytes.data, .size =
metadata_bytes.size};
+ validate_variant_metadata(metadata);
Review Comment:
这里不需要validate meta
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -514,6 +516,318 @@ void import_unshredded_variant_range(const
ParquetColumnSchema& schema, const Co
appender.append(std::span<const VariantRef>(encoded_rows));
}
+struct DirectResidualSeekResult {
+ ColumnPtr column;
+ int64_t rows = 0;
+ int64_t selected_value_bytes = 0;
+ int64_t container_index_builds = 0;
+ int64_t container_index_hits = 0;
+};
+
+struct DirectSeekContainerKey {
+ const char* metadata_data = nullptr;
+ size_t metadata_size = 0;
+ const char* value_data = nullptr;
+ size_t value_size = 0;
+
+ bool operator==(const DirectSeekContainerKey&) const = default;
+};
+
+struct DirectSeekContainerKeyHash {
+ size_t operator()(const DirectSeekContainerKey& key) const noexcept {
+ size_t hash = std::hash<const void*> {}(key.metadata_data);
+ auto combine = [&](size_t value) {
+ hash ^= value + 0x9e3779b97f4a7c15ULL + (hash << 6) + (hash >> 2);
+ };
+ combine(std::hash<size_t> {}(key.metadata_size));
+ combine(std::hash<const void*> {}(key.value_data));
+ combine(std::hash<size_t> {}(key.value_size));
+ return hash;
+ }
+};
+
+class DirectSeekContainerCache {
+public:
+ struct Lookup {
+ VariantContainerLookup* value;
+ bool cache_hit;
+ };
+
+ Lookup find_or_build(VariantRef value, size_t path_position,
DirectResidualSeekResult& result,
+ std::optional<VariantContainerLookup>&
uncached_lookup) {
+ const DirectSeekContainerKey key {.metadata_data = value.metadata.data,
+ .metadata_size = value.metadata.size,
+ .value_data = value.value.data,
+ .value_size = value.value.size};
+ if (auto found = _entries.find(key); found != _entries.end()) {
+ ++result.container_index_hits;
+ return {.value = &found->second, .cache_hit = true};
+ }
+ ++result.container_index_builds;
+ if (path_position >= MAX_RETAINED_PATH_DEPTH || _entries.size() >=
MAX_RETAINED_ENTRIES) {
+ // Validation is still mandatory after the high-water mark. Keep
only this traversal's
+ // lookup alive so an adversarial rows-times-depth path cannot
grow persistent state.
+ uncached_lookup.emplace(value);
+ return {.value = &*uncached_lookup, .cache_hit = false};
+ }
+ auto [inserted, was_inserted] = _entries.try_emplace(key, value);
+ DORIS_CHECK(was_inserted);
+ return {.value = &inserted->second, .cache_hit = false};
+ }
+
+ bool object_find_by_id(Lookup lookup, int64_t field_id, VariantRef* out) {
+ size_t promoted_bytes = 0;
+ const size_t promotion_budget =
+ lookup.cache_hit ? MAX_PROMOTED_OFFSET_BYTES -
_promoted_offset_bytes : 0;
+ const bool found =
+ lookup.value->object_find_by_id(field_id, out,
promotion_budget, &promoted_bytes);
+ _promoted_offset_bytes += promoted_bytes;
+ return found;
+ }
+
+ void reserve(size_t size) {
+ if (_entries.empty()) {
+ const size_t retained = size > MAX_RETAINED_ENTRIES /
MAX_RETAINED_PATH_DEPTH
+ ? MAX_RETAINED_ENTRIES
+ : size * MAX_RETAINED_PATH_DEPTH;
+ _entries.reserve(retained);
+ }
+ }
+
+ void reset() {
+ ContainerMap empty;
+ _entries.swap(empty);
+ _promoted_offset_bytes = 0;
+ }
+
+ size_t allocated_bytes() const {
+ return _entries.bucket_count() * sizeof(void*) +
+ _entries.size() * sizeof(ContainerMap::value_type) +
_promoted_offset_bytes;
+ }
+
+private:
+ // Four retained ancestors cover the common root/nested-object projections
for a default
+ // scanner batch. The independent entry and promoted-offset caps keep
state bounded for deep
+ // paths, appended columns, and noncanonical wide objects.
+ static constexpr size_t MAX_RETAINED_PATH_DEPTH = 4;
+ static constexpr size_t MAX_RETAINED_ENTRIES = 16 * 1024;
+ static constexpr size_t MAX_PROMOTED_OFFSET_BYTES = 4 * 1024 * 1024;
+
+ using ContainerMap = std::unordered_map<
+ DirectSeekContainerKey, VariantContainerLookup,
DirectSeekContainerKeyHash,
+ std::equal_to<DirectSeekContainerKey>,
+ CustomStdAllocator<std::pair<const DirectSeekContainerKey,
VariantContainerLookup>>>;
+ ContainerMap _entries;
+ size_t _promoted_offset_bytes = 0;
+};
+
+struct UnshreddedMetadataCache {
+ static constexpr uint32_t NO_METADATA =
std::numeric_limits<uint32_t>::max();
+
+ DorisVector<VariantMetadataRef> metadatas;
+ DorisVector<uint32_t> row_metadata_ids;
+
+ size_t allocated_bytes() const {
+ return metadatas.capacity() * sizeof(VariantMetadataRef) +
+ row_metadata_ids.capacity() * sizeof(uint32_t);
+ }
+};
+
+std::shared_ptr<const UnshreddedMetadataCache> build_unshredded_metadata_cache(
+ const ParquetColumnSchema& schema, const IColumn& physical, size_t
metadata_index) {
+ const auto* outer_nullable =
check_and_get_column<ColumnNullable>(physical);
+ const IColumn& wrapper =
+ outer_nullable == nullptr ? physical :
outer_nullable->get_nested_column();
+ const auto& structure = assert_cast<const ColumnStruct&>(wrapper);
+ if (structure.tuple_size() != schema.children.size()) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} physical
field count mismatch",
+ schema.name);
+ }
+
+ using MetadataIndex =
+ std::unordered_map<std::string_view, uint32_t,
std::hash<std::string_view>,
+ std::equal_to<std::string_view>,
+ CustomStdAllocator<std::pair<const
std::string_view, uint32_t>>>;
+ auto cache = std::make_shared<UnshreddedMetadataCache>();
+ cache->row_metadata_ids.resize(physical.size(),
UnshreddedMetadataCache::NO_METADATA);
+ MetadataIndex metadata_index_by_value;
+ std::optional<uint32_t> previous_metadata_id;
+
+ auto same_metadata = [&](uint32_t id, StringRef bytes) {
+ const VariantMetadataRef cached = cache->metadatas[id];
+ return StringRef(cached.data, cached.size) == bytes;
+ };
+ for (size_t row = 0; row < physical.size(); ++row) {
+ if (outer_nullable != nullptr &&
outer_nullable->get_null_map_data()[row] != 0) {
+ continue;
+ }
+ const Cell metadata_cell =
cell_at(structure.get_column(metadata_index), row);
+ if (metadata_cell.is_null) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} has
null metadata at row {}",
+ schema.name, row);
+ }
+ const StringRef metadata_bytes =
metadata_cell.column->get_data_at(row);
+ if (previous_metadata_id.has_value() &&
+ same_metadata(*previous_metadata_id, metadata_bytes)) {
+ cache->row_metadata_ids[row] = *previous_metadata_id;
+ continue;
+ }
+
+ const std::string_view key(metadata_bytes.data == nullptr ? "" :
metadata_bytes.data,
+ metadata_bytes.size);
+ if (!metadata_index_by_value.empty()) {
+ if (const auto found = metadata_index_by_value.find(key);
+ found != metadata_index_by_value.end()) {
+ cache->row_metadata_ids[row] = found->second;
+ previous_metadata_id = found->second;
+ continue;
+ }
+ }
+
+ VariantMetadataRef metadata {.data = metadata_bytes.data, .size =
metadata_bytes.size};
+ validate_variant_metadata(metadata);
+ if (cache->metadatas.size() == std::numeric_limits<uint32_t>::max()) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Parquet Variant metadata dictionary exceeds the
uint32 id limit");
+ }
+ if (cache->metadatas.size() == 1 && metadata_index_by_value.empty()) {
+ const VariantMetadataRef first = cache->metadatas.front();
+ metadata_index_by_value.emplace(
+ std::string_view(first.data == nullptr ? "" : first.data,
first.size), 0);
+ }
+ const auto id = static_cast<uint32_t>(cache->metadatas.size());
+ cache->metadatas.push_back(metadata);
+ if (!metadata_index_by_value.empty()) {
+ metadata_index_by_value.emplace(key, id);
+ }
+ cache->row_metadata_ids[row] = id;
+ previous_metadata_id = id;
+ }
+ return cache;
+}
+
+DirectResidualSeekResult seek_unshredded_variant_path(
+ const ParquetColumnSchema& schema, const IColumn& physical, size_t
value_index,
+ const UnshreddedMetadataCache& metadata_cache,
DirectSeekContainerCache& container_cache,
+ std::span<const VariantShreddedPathSegment> path) {
+ const auto* outer_nullable =
check_and_get_column<ColumnNullable>(physical);
+ const IColumn& wrapper =
+ outer_nullable == nullptr ? physical :
outer_nullable->get_nested_column();
+ const auto& structure = assert_cast<const ColumnStruct&>(wrapper);
+ if (structure.tuple_size() != schema.children.size()) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} physical
field count mismatch",
+ schema.name);
+ }
+
+ DORIS_CHECK_EQ(metadata_cache.row_metadata_ids.size(), physical.size());
+ if (!metadata_cache.metadatas.empty() &&
+ path.size() > std::numeric_limits<size_t>::max() /
metadata_cache.metadatas.size()) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Parquet Variant direct-seek path cache size overflows
size_t");
+ }
+ DorisVector<int64_t> object_ids(metadata_cache.metadatas.size() *
path.size(), -1);
+ for (size_t metadata_id = 0; metadata_id <
metadata_cache.metadatas.size(); ++metadata_id) {
+ for (size_t position = 0; position < path.size(); ++position) {
+ if (path[position].kind ==
VariantShreddedPathSegment::Kind::OBJECT_KEY) {
+ object_ids[metadata_id * path.size() + position] =
+
metadata_cache.metadatas[metadata_id].find_key(path[position].key);
+ }
+ }
+ }
+ VariantBatchBuilder builder(VariantBatchBuilder::ReserveHint {.rows =
physical.size()});
+ container_cache.reserve(physical.size());
+ auto nulls = ColumnUInt8::create();
+ nulls->reserve(physical.size());
+ int64_t selected_value_bytes = 0;
+ DirectResidualSeekResult result;
+ DORIS_CHECK_LE(physical.size(),
static_cast<size_t>(std::numeric_limits<int64_t>::max()));
+ result.rows = static_cast<int64_t>(physical.size());
+
+ auto add_selected_bytes = [&](size_t bytes) {
+ DORIS_CHECK_LE(bytes,
static_cast<size_t>(std::numeric_limits<int64_t>::max() -
+ selected_value_bytes));
+ selected_value_bytes += static_cast<int64_t>(bytes);
+ };
+ auto append_missing = [&]() {
+ auto output_row = builder.begin_row();
+ output_row.add_null();
+ output_row.finish();
+ nulls->insert_value(1);
+ add_selected_bytes(VARIANT_NULL_VALUE.size());
+ };
+ for (size_t row = 0; row < physical.size(); ++row) {
+ if (outer_nullable != nullptr &&
outer_nullable->get_null_map_data()[row] != 0) {
+ append_missing();
+ continue;
+ }
+
+ const uint32_t metadata_id = metadata_cache.row_metadata_ids[row];
+ DORIS_CHECK_NE(metadata_id, UnshreddedMetadataCache::NO_METADATA);
+ DORIS_CHECK_LT(metadata_id, metadata_cache.metadatas.size());
+ const VariantMetadataRef metadata =
metadata_cache.metadatas[metadata_id];
+ const Cell value_cell = cell_at(structure.get_column(value_index),
row);
+ if (value_cell.is_null) {
+ append_missing();
+ continue;
+ }
+
+ VariantRef current {.metadata = metadata, .value =
value_cell.column->get_data_at(row)};
+ bool found = true;
+ uint32_t current_depth = 0;
+ for (size_t position = 0; position < path.size(); ++position) {
+ VariantRef selected;
+ const VariantBasicType basic_type =
+ validate_variant_payload_shallow(current, current_depth);
Review Comment:
不需要validate， 直接拿basic_type， validate_variant_payload_shallow 这个函数也没必要
--
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]
