github-actions[bot] commented on code in PR #66758:
URL: https://github.com/apache/doris/pull/66758#discussion_r3802546981
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -514,6 +518,324 @@ 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;
+ // Rows published as a typed ColumnVariantV2, so no consumer decodes a
Variant header per row.
+ int64_t typed_rows = 0;
+ // One per batch that a value forced back onto canonical encoded rows.
+ int64_t typed_downgrades = 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;
+ }
+ }
+
+ // find_key() binary searches whenever the dictionary claims
sorted_strings. A dictionary
+ // that sets that flag while storing unordered keys would silently
miss a present key and
+ // turn it into SQL NULL, so validate each distinct dictionary once
here instead. The cache
+ // deduplicates metadata bytes, so this walks a handful of
dictionaries per batch rather
+ // than one per row.
+ VariantMetadataRef metadata {.data = metadata_bytes.data, .size =
metadata_bytes.size};
+ metadata.validate();
+ 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);
+ }
+ }
+ }
+ VariantSelectedValueBuilder builder(physical.size());
+ container_cache.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 = [&]() {
+ builder.append_missing();
+ 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;
+ // Direct seek assumes spec-valid scalar payloads and needs only
the basic type for
+ // dispatch. Containers still go through VariantContainerLookup
before their tables or
+ // offsets are trusted, and a selected subtree is validated at its
original depth below.
+ const VariantBasicType basic_type = current.basic_type();
+ std::optional<DirectSeekContainerCache::Lookup> container;
+ std::optional<VariantContainerLookup> uncached_container;
+ if (basic_type == VariantBasicType::OBJECT || basic_type ==
VariantBasicType::ARRAY) {
+ // Build bounded structural state even when the requested
segment has the other
+ // container kind, because the lookup will dereference this
container's table.
+ container = container_cache.find_or_build(current, position,
result,
+ uncached_container);
+ }
+ if (path[position].kind ==
VariantShreddedPathSegment::Kind::OBJECT_KEY) {
+ const int64_t field_id = object_ids[metadata_id * path.size()
+ position];
+ if (basic_type != VariantBasicType::OBJECT ||
+ !container_cache.object_find_by_id(*container, field_id,
&selected)) {
+ found = false;
+ break;
+ }
+ } else {
+ if (basic_type != VariantBasicType::ARRAY ||
+ !container->value->array_find(path[position].index,
&selected)) {
+ found = false;
+ break;
+ }
+ }
+ current = selected;
+ ++current_depth;
+ }
+ if (!found) {
+ append_missing();
+ continue;
+ }
+
+ // Traversed containers perform bounded reads. Validate the selected
subtree exactly, but
+ // intentionally do not visit unrelated siblings in the unshredded
root.
+ validate_variant_payload(current, current_depth);
+ // A homogeneous scalar projection becomes a typed column, so the
selected leaf is decoded
+ // once instead of being re-encoded here and parsed again by every
consumer. Containers and
+ // mixed kinds fall back to canonical encoded rows, which keep only
the reachable keys with
+ // remapped ids; copying the root dictionary would multiply wide
metadata across
+ // independently projected paths and composite states.
+ builder.append_selected(current);
+ add_selected_bytes(current.value.size);
Review Comment:
[P2] Keep direct-seek bytes in one documented unit
`VariantDirectResidualSeekBytes` is documented as selected encoded bytes
copied to the result, but this adds the source `current.value.size` even when
`VariantSelectedValueBuilder` decodes a homogeneous scalar into a typed Doris
column without copying that slice. Missing rows also add a synthetic one-byte
null even though they are represented by the result null map, while
container/mixed batches rebuild compact bytes whose size can differ from the
source. The counter therefore mixes incompatible units on the first successful
request, independently of the older cached-result thread. Please either
define/count source encoded bytes examined (with misses contributing zero) or
measure the actual output bytes, and add exact integer/string/missing/container
assertions.
--
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]
