github-actions[bot] commented on code in PR #66758:
URL: https://github.com/apache/doris/pull/66758#discussion_r3781318336
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -514,6 +516,200 @@ void import_unshredded_variant_range(const
ParquetColumnSchema& schema, const Co
appender.append(std::span<const VariantRef>(encoded_rows));
}
+struct DirectResidualSeekResult {
+ ColumnPtr column;
+ int64_t selected_value_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,
+ 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);
+ }
+ }
+ }
+ DorisVector<VariantRef> selected_rows;
+ selected_rows.reserve(physical.size());
+ auto nulls = ColumnUInt8::create();
+ nulls->reserve(physical.size());
+ std::vector<uint32_t> object_offset_scratch;
+ int64_t selected_value_bytes = 0;
+
+ 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 = [&](VariantMetadataRef metadata) {
+ selected_rows.push_back({.metadata = metadata,
+ .value = {VARIANT_NULL_VALUE.data(),
VARIANT_NULL_VALUE.size()}});
+ 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(
+ {.data = VARIANT_EMPTY_METADATA.data(), .size =
VARIANT_EMPTY_METADATA.size()});
+ 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(metadata);
+ continue;
+ }
+
+ VariantRef current {.metadata = metadata, .value =
value_cell.column->get_data_at(row)};
Review Comment:
[P1] Validate the root before direct traversal
The physical cell is not validated before its type decides path existence. A
truncated or semantically invalid scalar (for example invalid UTF-8) becomes a
type-mismatch SQL NULL, while a valid container followed by one extra byte
returns its child and drops that byte. Canonical Variant validation rejects all
these roots, and `append_prevalidated` leaves no later check. Please fully
validate scalar roots and require a container's shallow extent to match the
complete cell before traversal, without recursively visiting unrelated sibling
payloads.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -514,6 +516,200 @@ void import_unshredded_variant_range(const
ParquetColumnSchema& schema, const Co
appender.append(std::span<const VariantRef>(encoded_rows));
}
+struct DirectResidualSeekResult {
+ ColumnPtr column;
+ int64_t selected_value_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,
+ 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);
+ }
+ }
+ }
+ DorisVector<VariantRef> selected_rows;
+ selected_rows.reserve(physical.size());
+ auto nulls = ColumnUInt8::create();
+ nulls->reserve(physical.size());
+ std::vector<uint32_t> object_offset_scratch;
+ int64_t selected_value_bytes = 0;
+
+ 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 = [&](VariantMetadataRef metadata) {
+ selected_rows.push_back({.metadata = metadata,
+ .value = {VARIANT_NULL_VALUE.data(),
VARIANT_NULL_VALUE.size()}});
+ 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(
+ {.data = VARIANT_EMPTY_METADATA.data(), .size =
VARIANT_EMPTY_METADATA.size()});
+ 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(metadata);
+ continue;
+ }
+
+ VariantRef current {.metadata = metadata, .value =
value_cell.column->get_data_at(row)};
+ bool found = true;
+ for (size_t position = 0; position < path.size(); ++position) {
+ VariantRef selected;
+ if (path[position].kind ==
VariantShreddedPathSegment::Kind::OBJECT_KEY) {
+ const int64_t field_id = object_ids[metadata_id * path.size()
+ position];
+ if (current.basic_type() != VariantBasicType::OBJECT ||
+ !current.object_find_by_id_untrusted(field_id, &selected,
+
object_offset_scratch)) {
+ found = false;
+ break;
+ }
+ } else {
+ if (current.basic_type() != VariantBasicType::ARRAY ||
+ !current.array_find_untrusted(path[position].index,
&selected)) {
+ found = false;
+ break;
+ }
+ }
+ current = selected;
+ }
+ if (!found) {
+ append_missing(metadata);
+ 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);
Review Comment:
[P1] Preserve ancestor depth in selected-subtree validation
The selected node is always validated as depth zero, so one ordinary
array-element lookup from a 129-level singleton-array root drops the outer
level and accepts a subtree at depth 128 even though canonical Variant
validation rejects the source at depth 129. Carry the traversed-container count
into validation (and into miss/type-mismatch exits) so this fast path preserves
the existing nesting invariant.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -851,18 +1048,32 @@ class ParquetVariantShreddedState final : public
VariantShreddedState {
std::lock_guard lock(_materialization_lock);
_materialized.reset();
_serialized.reset();
+ _unshredded_metadata_cache.reset();
return true;
}
std::optional<VariantShreddedTypedValue> find_typed_value(
std::span<const VariantShreddedPathSegment> path) const override {
+ auto residual_seek_fallback = [&]() ->
std::optional<VariantShreddedTypedValue> {
+ // Complete mixed shredded states still need canonical
reconstruction when neither a
+ // typed leaf nor the pure unshredded direct-seek path can answer
the request.
+ if (_complete && !unshredded_child_indices(*_schema).has_value() &&
+ find_child(*_schema, "value", nullptr) != nullptr) {
+
update_counter(_profile.variant_direct_residual_seek_fallbacks, 1);
+ }
+ return std::nullopt;
+ };
auto path_miss = [&]() -> std::optional<VariantShreddedTypedValue> {
update_counter(_profile.variant_direct_leaf_path_misses, 1);
- return std::nullopt;
+ return residual_seek_fallback();
};
if (path.empty()) {
return path_miss();
}
+ if (auto normalized = find_unshredded_normalized_value(path);
normalized.has_value()) {
Review Comment:
[P1] Avoid rescanning wide containers for every projected path
Every `element_at` on a complete unshredded column now takes this route, but
the cache stores only metadata identities. Each array lookup scans all K
offsets, and each object lookup scans all keys/offsets and sorts them, so E
projected paths cost Theta(E*N*K) or Theta(E*N*K log K). Previously the first
lookup cached one validation/materialization and later paths used O(1) array or
O(log K) object lookup. Please cache reusable shallow container
validation/index state for the immutable physical rows (with
append/filter/select invalidation) and cover wide multi-projection inputs;
falling back to full materialization would make unrelated-sibling validation
depend on projection count.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -851,18 +1048,32 @@ class ParquetVariantShreddedState final : public
VariantShreddedState {
std::lock_guard lock(_materialization_lock);
_materialized.reset();
_serialized.reset();
+ _unshredded_metadata_cache.reset();
return true;
}
std::optional<VariantShreddedTypedValue> find_typed_value(
std::span<const VariantShreddedPathSegment> path) const override {
+ auto residual_seek_fallback = [&]() ->
std::optional<VariantShreddedTypedValue> {
+ // Complete mixed shredded states still need canonical
reconstruction when neither a
+ // typed leaf nor the pure unshredded direct-seek path can answer
the request.
+ if (_complete && !unshredded_child_indices(*_schema).has_value() &&
+ find_child(*_schema, "value", nullptr) != nullptr) {
+
update_counter(_profile.variant_direct_residual_seek_fallbacks, 1);
+ }
+ return std::nullopt;
+ };
auto path_miss = [&]() -> std::optional<VariantShreddedTypedValue> {
update_counter(_profile.variant_direct_leaf_path_misses, 1);
- return std::nullopt;
+ return residual_seek_fallback();
};
if (path.empty()) {
return path_miss();
}
+ if (auto normalized = find_unshredded_normalized_value(path);
normalized.has_value()) {
+ return VariantShreddedTypedValue {
Review Comment:
[P1] Update the Iceberg profile contracts for this direct path
This early return removes `VariantDirectLeafPathMisses` and, for
predicate-only scans, `VariantReconstructedRows` from complete unshredded
files. The unchanged `external_table_p0` Iceberg suite still requires those old
counters at lines 1407, 1466-1469, 1481, 1573, and 1771-1774; the
multi-row-group and position-delete queries only evaluate `v['n']`, so they now
keep both old counters at zero and fail. Please update those assertions to
require `VariantDirectResidualSeekRows` and expect no reconstruction where the
root is not projected.
--
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]
