github-actions[bot] commented on code in PR #66758:
URL: https://github.com/apache/doris/pull/66758#discussion_r3796685093
##########
be/src/format_v2/column_mapper.cpp:
##########
@@ -2008,17 +2008,48 @@ static Status build_scan_projection(ColumnMapping*
mapping, bool force_full_comp
return Status::OK();
}
+static void merge_variant_access_paths(VariantAccessPaths* target,
+ const VariantAccessPaths& source) {
+ DORIS_CHECK(target != nullptr);
+ target->insert(target->end(), source.begin(), source.end());
+ std::ranges::sort(*target);
+ target->erase(std::unique(target->begin(), target->end()), target->end());
+}
+
+static void register_root_variant_access_paths(FileScanRequest* file_request,
+ const ColumnMapping& mapping,
+ bool is_predicate_column) {
+ DORIS_CHECK(file_request != nullptr);
+ if (!mapping.file_local_id.has_value() || mapping.table_type == nullptr ||
+ mapping.variant_access_paths.empty() ||
Review Comment:
[P2] Carry access paths to nested Variant terminals
Nested Variant paths are preserved on recursive child mappings, but this
root-only registration returns for a surrounding STRUCT/ARRAY/MAP, and
recursive `build_variant_plan()` calls receive no path metadata. A nested
unshredded Variant therefore sees an empty plan and bypasses the new multi-path
materialization fallback, so queries such as `info.payload['x']` plus
`info.payload['deep']['name']` still seek and re-encode every row once per
expression. Please represent/pass paths per nested Variant terminal and add
STRUCT/ARRAY/MAP multi-path coverage. This is distinct from the live top-level
multi-path thread because that fix is never wired to nested terminals.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -514,6 +516,323 @@ 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;
+ }
+ }
+
+ // Parquet requires every Variant metadata/value pair to be
spec-valid. This direct-seek
+ // cache therefore only deduplicates the immutable metadata bytes
instead of eagerly
+ // walking every dictionary key. The bounded find_key()/key_at()
accessors still reject
+ // truncated layouts before any referenced key is observed.
+ VariantMetadataRef metadata {.data = metadata_bytes.data, .size =
metadata_bytes.size};
Review Comment:
[P1] Validate metadata before trusting the sorted flag
This cache stores each unique metadata blob without validation, but path
preparation immediately calls `find_key()`, which binary-searches whenever the
header merely advertises sorted strings. For example, metadata keys `['z','a']`
marked sorted plus a root object id table `[1,0]` is locally ordered as `a,z`,
yet lookup of `z` returns `-1` and this path emits SQL NULL even though the key
is present. The previous materialization path rejected the same row through
`validate_variant_metadata()`. Please validate each unique dictionary before
using its ordering/uniqueness claims and cover a falsely advertised sorted
dictionary. This is distinct from the existing root-payload/depth threads
because the wrong result comes from the metadata search invariant.
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -527,12 +527,66 @@ class CompositeVariantShreddedState final : public
VariantShreddedState {
.normalized = nullptr};
}
- auto normalized = find_normalized_value(path);
- if (!normalized.has_value()) {
- return std::nullopt;
+ const bool all_direct_values =
+ all_direct && std::ranges::all_of(matches, [](const auto&
match) {
+ const bool typed = match.column && match.type &&
!match.normalized;
+ const bool normalized = match.normalized && !match.column
&& !match.type;
+ return typed || normalized;
+ });
+ if (all_direct_values) {
+ auto values = ColumnVariantV2::create();
+ auto nulls = ColumnUInt8::create();
+ nulls->reserve(size());
+ for (size_t index = 0; index < matches.size(); ++index) {
+ const auto& match = matches[index];
+ std::optional<ColumnPtr> normalized_typed;
+ if (!match.normalized) {
+ // Typed Parquet leaves need their segment's schema-aware
normalization to
+ // preserve physical distinctions such as INT32 versus
INT64. Normalize only
+ // those segments; a normalized unshredded match already
contains exact bytes.
+ normalized_typed =
_segments[index]->find_normalized_value(path);
+ if (!normalized_typed.has_value()) {
+ return std::nullopt;
+ }
+ }
+ const ColumnPtr& matched = match.normalized ? match.normalized
: *normalized_typed;
+ const auto& nullable = assert_cast<const
ColumnNullable&>(*matched);
+ nulls->insert_range_from(nullable.get_null_map_column(), 0,
nullable.size());
+ const auto& variants =
+ assert_cast<const
ColumnVariantV2&>(nullable.get_nested_column());
+ values->insert_range_from(variants, 0, variants.size());
+ }
+ return VariantShreddedTypedValue {
+ .column = nullptr,
+ .type = nullptr,
+ .normalized = ColumnNullable::create(std::move(values),
std::move(nulls))};
+ }
+
+ auto values = ColumnVariantV2::create();
+ auto nulls = ColumnUInt8::create();
+ nulls->reserve(size());
+ for (size_t index = 0; index < _segments.size(); ++index) {
+ std::optional<ColumnPtr> normalized;
+ if (index < matches.size() && matches[index].normalized) {
+ // Keep normalized prefix matches that were completed before a
later segment
+ // requested fallback. In particular, do not seek an
unshredded segment twice.
+ normalized = matches[index].normalized;
+ } else {
+ normalized = _segments[index]->find_normalized_value(path);
Review Comment:
[P2] Count mixed residual fallbacks reached through normalization
After an earlier segment misses, this loop sends every unvisited segment
directly to `find_normalized_value()`. A later complete mixed Parquet segment
then reconstructs through `materialized_column()`, but
`VariantDirectResidualSeekFallbacks` is incremented only by that segment's
skipped `find_typed_value()` fallback. Consequently `[partial miss, complete
mixed]` reports zero while the reversed segment order reports one for the same
reconstruction work, contrary to the new counter contract. Please count the
complete-mixed-to-materialized transition in one common place and test both
segment orders. This is separate from the old cached rows/bytes comment because
it concerns the residual fallback counter and composite branch.
--
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]
