github-actions[bot] commented on code in PR #67016:
URL: https://github.com/apache/doris/pull/67016#discussion_r4003508999
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -586,13 +611,105 @@ ColumnPtr normalize_projected_primitive_leaf(const
ParquetColumnSchema& schema,
return ColumnNullable::create(std::move(values), std::move(nulls));
}
-bool find_materialized_path(VariantRef current, std::span<const
VariantShreddedPathSegment> path,
- VariantRef* output) {
- DORIS_CHECK(output != nullptr);
- for (const auto& segment : path) {
+struct UnshreddedPathCacheEntry {
+ static constexpr uint32_t MISSING = std::numeric_limits<uint32_t>::max();
+
+ uint32_t value_offset = MISSING;
+ uint32_t value_size = 0;
+
+ bool present() const noexcept { return value_offset != MISSING; }
+};
+
+struct UnshreddedPathCache {
+ DorisVector<UnshreddedPathCacheEntry> entries;
+
+ size_t byte_size() const noexcept { return entries.size() *
sizeof(UnshreddedPathCacheEntry); }
+ size_t allocated_bytes() const noexcept {
+ return entries.capacity() * sizeof(UnshreddedPathCacheEntry);
+ }
+};
+
+struct UnshreddedMetadataIndex {
+ static constexpr uint32_t NULL_ROW = std::numeric_limits<uint32_t>::max();
+
+ const IColumn* physical_identity = nullptr;
+ DorisVector<VariantMetadataRef> dictionaries;
+ DorisVector<uint32_t> row_dictionary_ids;
+
+ size_t byte_size() const noexcept {
+ return dictionaries.size() * sizeof(VariantMetadataRef) +
+ row_dictionary_ids.size() * sizeof(uint32_t);
+ }
+ size_t allocated_bytes() const noexcept {
+ return dictionaries.capacity() * sizeof(VariantMetadataRef) +
+ row_dictionary_ids.capacity() * sizeof(uint32_t);
+ }
+};
+
+std::shared_ptr<const UnshreddedMetadataIndex> build_unshredded_metadata_index(
+ const ParquetColumnSchema& schema, const IColumn& physical,
+ std::pair<size_t, size_t> child_indices) {
+ 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);
+ DORIS_CHECK_EQ(structure.tuple_size(), schema.children.size());
+
+ auto index = std::make_shared<UnshreddedMetadataIndex>();
+ index->physical_identity = &physical;
+ index->row_dictionary_ids.resize(physical.size(),
UnshreddedMetadataIndex::NULL_ROW);
+ using MetadataIdMap =
+ 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>>>;
+ MetadataIdMap dictionary_ids;
+ 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_at(structure.get_column(child_indices.first), row);
+ if (metadata.is_null) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} has
null metadata at row {}",
+ schema.name, row);
+ }
+ const StringRef bytes = metadata.column->get_data_at(row);
+ uint32_t dictionary_id = 0;
+ if (index->dictionaries.empty()) {
+ index->dictionaries.push_back({.data = bytes.data, .size =
bytes.size});
+ } else if (index->dictionaries.size() == 1 && dictionary_ids.empty() &&
+ StringRef(index->dictionaries.front().data,
index->dictionaries.front().size) ==
+ bytes) {
+ // Iceberg normally repeats one metadata dictionary throughout a
decoded block. Delay
+ // the hash table until a second dictionary is actually observed.
+ } else {
+ if (dictionary_ids.empty()) {
+ const VariantMetadataRef first = index->dictionaries.front();
+ dictionary_ids.emplace(std::string_view(first.data,
first.size), 0);
+ }
+ const std::string_view key(bytes.data, bytes.size);
+ if (const auto found = dictionary_ids.find(key); found !=
dictionary_ids.end()) {
+ dictionary_id = found->second;
+ } else {
+ dictionary_id =
static_cast<uint32_t>(index->dictionaries.size());
+ index->dictionaries.push_back({.data = bytes.data, .size =
bytes.size});
+ dictionary_ids.emplace(key, dictionary_id);
+ }
+ }
+ index->row_dictionary_ids[row] = dictionary_id;
+ }
+ return index;
+}
+
+template <typename ObjectFinder>
+bool find_materialized_path_impl(VariantRef current,
+ std::span<const VariantShreddedPathSegment>
path,
+ const ObjectFinder& find_object, VariantRef*
output) {
+ DCHECK(output != nullptr);
+ for (size_t position = 0; position < path.size(); ++position) {
Review Comment:
[P1] Preserve the maximum nesting-depth check on direct traversal
This loop can descend through an arbitrarily long selected object/array
chain without carrying the absolute Variant depth. A structurally valid row
with 129 nested single-child containers followed by a canonical string/int is
therefore returned as a typed value, while the former root reconstruction
rejects it in `VariantBatchBuilder::require_import_depth()` at the 128-level
limit. Chained `element_at` calls also reset the subtree's apparent root, so
later normalization cannot recover the check. Please track the accumulated
prefix depth and enforce `VARIANT_MAX_NESTING_DEPTH`, with 128/129 boundary
tests for one-shot and chained traversal.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -611,6 +728,400 @@ bool find_materialized_path(VariantRef current,
std::span<const VariantShreddedP
return true;
}
+bool find_materialized_path(VariantRef current, std::span<const
VariantShreddedPathSegment> path,
+ VariantRef* output) {
+ return find_materialized_path_impl(
+ current, path,
+ [](VariantRef object, StringRef key, size_t, VariantRef* found) {
+ return object.object_find(key, found);
+ },
+ output);
+}
+
+bool find_materialized_path_with_index(VariantRef current, uint32_t
dictionary_id,
+ const UnshreddedMetadataIndex&
metadata_index,
+ std::span<const
VariantShreddedPathSegment> path,
+ DorisVector<int64_t>&
resolved_field_ids,
+ VariantRef* output) {
+ DCHECK_LT(dictionary_id, metadata_index.dictionaries.size());
+ DCHECK_EQ(resolved_field_ids.size(), metadata_index.dictionaries.size() *
path.size());
+ constexpr int64_t UNRESOLVED_FIELD_ID = -2;
+ return find_materialized_path_impl(
+ current, path,
+ [&](VariantRef object, StringRef key, size_t position, VariantRef*
found) {
+ int64_t& field_id = resolved_field_ids[dictionary_id *
path.size() + position];
+ bool layout_validated = false;
+ if (field_id == UNRESOLVED_FIELD_ID) {
+ // object_find() validates the object layout before
consulting metadata.
+ static_cast<void>(object.num_elements());
+ layout_validated = true;
+ field_id =
metadata_index.dictionaries[dictionary_id].find_key(key);
Review Comment:
[P1] Validate metadata before trusting binary-search ordering
This resolves a key with `find_key()` on raw Parquet metadata before
`VariantMetadataRef::validate()` has established the advertised
sorted-and-unique invariant. For example, metadata containing `["b","a"]` while
setting the sorted flag makes lookup of `"a"` return `-1`, so a present field
is emitted as a missing-path SQL NULL; the former root reconstruction rejected
the metadata. The later integer-only validation cannot help misses, strings,
JSON nulls, or fallback types. Please validate each distinct dictionary before
caching path IDs and cover false sorted/duplicate metadata.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -611,6 +728,400 @@ bool find_materialized_path(VariantRef current,
std::span<const VariantShreddedP
return true;
}
+bool find_materialized_path(VariantRef current, std::span<const
VariantShreddedPathSegment> path,
+ VariantRef* output) {
+ return find_materialized_path_impl(
+ current, path,
+ [](VariantRef object, StringRef key, size_t, VariantRef* found) {
+ return object.object_find(key, found);
+ },
+ output);
+}
+
+bool find_materialized_path_with_index(VariantRef current, uint32_t
dictionary_id,
+ const UnshreddedMetadataIndex&
metadata_index,
+ std::span<const
VariantShreddedPathSegment> path,
+ DorisVector<int64_t>&
resolved_field_ids,
+ VariantRef* output) {
+ DCHECK_LT(dictionary_id, metadata_index.dictionaries.size());
+ DCHECK_EQ(resolved_field_ids.size(), metadata_index.dictionaries.size() *
path.size());
+ constexpr int64_t UNRESOLVED_FIELD_ID = -2;
+ return find_materialized_path_impl(
+ current, path,
+ [&](VariantRef object, StringRef key, size_t position, VariantRef*
found) {
+ int64_t& field_id = resolved_field_ids[dictionary_id *
path.size() + position];
+ bool layout_validated = false;
+ if (field_id == UNRESOLVED_FIELD_ID) {
+ // object_find() validates the object layout before
consulting metadata.
+ static_cast<void>(object.num_elements());
+ layout_validated = true;
+ field_id =
metadata_index.dictionaries[dictionary_id].find_key(key);
+ }
+ if (field_id < 0) {
+ // A cached metadata miss must not hide a corrupt object
in a later row.
+ if (!layout_validated) {
+ static_cast<void>(object.num_elements());
+ }
+ return false;
+ }
+ return
object.object_find_by_id(static_cast<uint32_t>(field_id), found);
Review Comment:
[P1] Validate the raw object layout before indexed lookup
This now looks up directly in the Parquet `metadata + value` bytes, before
`VariantBatchBuilder::validate_import_object()` has checked strict key ordering
and a complete non-overlapping partition of the object value region.
`object_find_by_id()` only validates the IDs it probes and the selected child's
encoded length. For example, a two-field object with IDs `[0,1]` and both
offsets set to zero makes both keys return the first scalar, whereas the former
reconstruction path rejects the overlap. Please establish those object-layout
invariants before using the cached ID, and cover
overlapping/gapped/out-of-order entries.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -611,6 +728,400 @@ bool find_materialized_path(VariantRef current,
std::span<const VariantShreddedP
return true;
}
+bool find_materialized_path(VariantRef current, std::span<const
VariantShreddedPathSegment> path,
+ VariantRef* output) {
+ return find_materialized_path_impl(
+ current, path,
+ [](VariantRef object, StringRef key, size_t, VariantRef* found) {
+ return object.object_find(key, found);
+ },
+ output);
+}
+
+bool find_materialized_path_with_index(VariantRef current, uint32_t
dictionary_id,
+ const UnshreddedMetadataIndex&
metadata_index,
+ std::span<const
VariantShreddedPathSegment> path,
+ DorisVector<int64_t>&
resolved_field_ids,
+ VariantRef* output) {
+ DCHECK_LT(dictionary_id, metadata_index.dictionaries.size());
+ DCHECK_EQ(resolved_field_ids.size(), metadata_index.dictionaries.size() *
path.size());
+ constexpr int64_t UNRESOLVED_FIELD_ID = -2;
+ return find_materialized_path_impl(
+ current, path,
+ [&](VariantRef object, StringRef key, size_t position, VariantRef*
found) {
+ int64_t& field_id = resolved_field_ids[dictionary_id *
path.size() + position];
+ bool layout_validated = false;
+ if (field_id == UNRESOLVED_FIELD_ID) {
+ // object_find() validates the object layout before
consulting metadata.
+ static_cast<void>(object.num_elements());
+ layout_validated = true;
+ field_id =
metadata_index.dictionaries[dictionary_id].find_key(key);
+ }
+ if (field_id < 0) {
+ // A cached metadata miss must not hide a corrupt object
in a later row.
+ if (!layout_validated) {
+ static_cast<void>(object.num_elements());
+ }
+ return false;
+ }
+ return
object.object_find_by_id(static_cast<uint32_t>(field_id), found);
+ },
+ output);
+}
+
+struct UnshreddedPathScan {
+ MutableColumnPtr outer_nulls;
+ MutableColumnPtr typed_values;
+ DataTypePtr typed_type;
+ std::shared_ptr<const UnshreddedPathCache> path_cache;
+ int64_t copied_bytes = 0;
+};
+
+enum class UnshreddedTypedKind : uint8_t { UNKNOWN, STRING, INTEGER,
UNSUPPORTED };
+
+class UnshreddedTypedValueBuilder {
+public:
+ UnshreddedTypedValueBuilder(size_t rows, const UnshreddedMetadataIndex&
metadata_index)
+ : _rows(rows),
+ _metadata_index(metadata_index),
+ _inner_nulls(ColumnUInt8::create()),
+ _result_nulls(ColumnUInt8::create()),
+ _validated_metadata(metadata_index.dictionaries.size(), 0) {
+ _inner_nulls->reserve(rows);
+ _result_nulls->reserve(rows);
+ }
+
+ void append_outer_null() { append_null(1); }
+
+ void append_json_null(uint32_t dictionary_id) {
+ if (_typed_kind == UnshreddedTypedKind::UNKNOWN) {
+ _pending_json_null_dictionaries.push_back(dictionary_id);
+ } else if (_typed_kind == UnshreddedTypedKind::INTEGER &&
+ !validate_integer_metadata(dictionary_id)) {
+ mark_unsupported();
+ }
+ append_null(0);
+ }
+
+ void append_scalar(const VariantRef& found, uint32_t dictionary_id, size_t
row) {
+ if (_typed_kind == UnshreddedTypedKind::UNSUPPORTED) {
+ append_null(0);
+ return;
+ }
+
+ const VariantBasicType basic_type = found.basic_type();
+ const bool is_string = basic_type == VariantBasicType::SHORT_STRING ||
+ (basic_type == VariantBasicType::PRIMITIVE &&
+ found.primitive_id() ==
VariantPrimitiveId::STRING);
+ if (is_string) {
+ if (!prepare(UnshreddedTypedKind::STRING, row)) {
+ append_null(0);
+ return;
+ }
+ const StringRef string = found.get_string();
Review Comment:
[P1] Preserve UTF-8 validation on the direct STRING path
`VariantRef::get_string()` validates the encoded length, but it does not
validate UTF-8. Copying it straight into a typed `ColumnString` therefore
accepts a malformed requested Variant string that the former
`VariantBatchBuilder` import rejected via `require_import_utf8()` /
`VariantScalarRef::string()`. This is observable without a later encode because
the typed Variant-to-string cast returns the physical string column directly.
Please validate the selected bytes before typed promotion and add an
invalid-UTF-8 direct-extraction case.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -758,6 +1344,11 @@ class ParquetVariantShreddedState final : public
VariantShreddedState {
return path_miss();
}
+ if (unshredded_child_indices(*_schema).has_value()) {
+ return VariantShreddedTypedValue {
+ .column = nullptr, .type = nullptr, .normalized =
direct_unshredded_path(path)};
Review Comment:
[P2] Avoid executing direct seek twice for composite states
This eagerly runs `direct_unshredded_path()` while
`CompositeVariantShreddedState` is only probing whether all segments have one
homogeneous typed representation. A mixed unshredded/partial-shredded composite
fails that homogeneity check because this result is normalized, then
`CompositeVariantShreddedState::find_normalized_value()` calls this segment's
`direct_unshredded_path()` again. Such mixed file layouts therefore scan every
row and allocate the result twice for one `element_at`. Please let the
composite reuse the normalized result it already obtained, or make this probe
non-materializing.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -611,6 +728,400 @@ bool find_materialized_path(VariantRef current,
std::span<const VariantShreddedP
return true;
}
+bool find_materialized_path(VariantRef current, std::span<const
VariantShreddedPathSegment> path,
+ VariantRef* output) {
+ return find_materialized_path_impl(
+ current, path,
+ [](VariantRef object, StringRef key, size_t, VariantRef* found) {
+ return object.object_find(key, found);
+ },
+ output);
+}
+
+bool find_materialized_path_with_index(VariantRef current, uint32_t
dictionary_id,
+ const UnshreddedMetadataIndex&
metadata_index,
+ std::span<const
VariantShreddedPathSegment> path,
+ DorisVector<int64_t>&
resolved_field_ids,
+ VariantRef* output) {
+ DCHECK_LT(dictionary_id, metadata_index.dictionaries.size());
+ DCHECK_EQ(resolved_field_ids.size(), metadata_index.dictionaries.size() *
path.size());
+ constexpr int64_t UNRESOLVED_FIELD_ID = -2;
+ return find_materialized_path_impl(
+ current, path,
+ [&](VariantRef object, StringRef key, size_t position, VariantRef*
found) {
+ int64_t& field_id = resolved_field_ids[dictionary_id *
path.size() + position];
+ bool layout_validated = false;
+ if (field_id == UNRESOLVED_FIELD_ID) {
+ // object_find() validates the object layout before
consulting metadata.
+ static_cast<void>(object.num_elements());
+ layout_validated = true;
+ field_id =
metadata_index.dictionaries[dictionary_id].find_key(key);
+ }
+ if (field_id < 0) {
+ // A cached metadata miss must not hide a corrupt object
in a later row.
+ if (!layout_validated) {
+ static_cast<void>(object.num_elements());
+ }
+ return false;
+ }
+ return
object.object_find_by_id(static_cast<uint32_t>(field_id), found);
+ },
+ output);
+}
+
+struct UnshreddedPathScan {
+ MutableColumnPtr outer_nulls;
+ MutableColumnPtr typed_values;
+ DataTypePtr typed_type;
+ std::shared_ptr<const UnshreddedPathCache> path_cache;
+ int64_t copied_bytes = 0;
+};
+
+enum class UnshreddedTypedKind : uint8_t { UNKNOWN, STRING, INTEGER,
UNSUPPORTED };
+
+class UnshreddedTypedValueBuilder {
+public:
+ UnshreddedTypedValueBuilder(size_t rows, const UnshreddedMetadataIndex&
metadata_index)
+ : _rows(rows),
+ _metadata_index(metadata_index),
+ _inner_nulls(ColumnUInt8::create()),
+ _result_nulls(ColumnUInt8::create()),
+ _validated_metadata(metadata_index.dictionaries.size(), 0) {
+ _inner_nulls->reserve(rows);
+ _result_nulls->reserve(rows);
+ }
+
+ void append_outer_null() { append_null(1); }
+
+ void append_json_null(uint32_t dictionary_id) {
+ if (_typed_kind == UnshreddedTypedKind::UNKNOWN) {
+ _pending_json_null_dictionaries.push_back(dictionary_id);
+ } else if (_typed_kind == UnshreddedTypedKind::INTEGER &&
+ !validate_integer_metadata(dictionary_id)) {
+ mark_unsupported();
+ }
+ append_null(0);
+ }
+
+ void append_scalar(const VariantRef& found, uint32_t dictionary_id, size_t
row) {
+ if (_typed_kind == UnshreddedTypedKind::UNSUPPORTED) {
+ append_null(0);
+ return;
+ }
+
+ const VariantBasicType basic_type = found.basic_type();
+ const bool is_string = basic_type == VariantBasicType::SHORT_STRING ||
+ (basic_type == VariantBasicType::PRIMITIVE &&
+ found.primitive_id() ==
VariantPrimitiveId::STRING);
+ if (is_string) {
+ if (!prepare(UnshreddedTypedKind::STRING, row)) {
+ append_null(0);
+ return;
+ }
+ const StringRef string = found.get_string();
+
assert_cast<ColumnString&>(*_typed_values).insert_data(string.data,
string.size);
+ _inner_nulls->insert_value(0);
+ _result_nulls->insert_value(0);
+ DCHECK_LE(string.size,
+ static_cast<size_t>(std::numeric_limits<int64_t>::max()
- _copied_bytes));
+ _copied_bytes += static_cast<int64_t>(string.size);
+ return;
+ }
+
+ const auto primitive_id = basic_type == VariantBasicType::PRIMITIVE
+ ? found.primitive_id()
+ : VariantPrimitiveId::NULL_VALUE;
+ const bool is_integer = primitive_id == VariantPrimitiveId::INT8 ||
+ primitive_id == VariantPrimitiveId::INT16 ||
+ primitive_id == VariantPrimitiveId::INT32 ||
+ primitive_id == VariantPrimitiveId::INT64;
+ const int64_t integer = is_integer ? found.get_int() : 0;
+ // Typed Variant integers are re-encoded using the narrowest width.
Keep explicitly widened
+ // source integers on the encoded path so observable physical types
remain unchanged.
+ const bool has_canonical_width =
+ is_integer &&
VariantScalarRef::integer(integer).encoded_size() == found.value.size;
+ if (!has_canonical_width || !validate_integer_metadata(dictionary_id)
||
Review Comment:
[P1] Validate selected unsupported scalars before publishing a subtree
When the selected value is neither a string nor a canonical integer, this
branch marks the result unsupported without invoking the primitive's semantic
accessor. An exactly bounded `DECIMAL16` with scale 39 thus becomes a non-null
lazy shredded value; `element_at(...) IS NOT NULL` reads only the returned null
map and succeeds, whereas the former reconstruction called `get_decimal()` /
`import_primitive()` and rejected it. Please validate the selected scalar
before deferring its representation, and add invalid decimal scale/magnitude
tests with a null-only consumer.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -611,6 +728,400 @@ bool find_materialized_path(VariantRef current,
std::span<const VariantShreddedP
return true;
}
+bool find_materialized_path(VariantRef current, std::span<const
VariantShreddedPathSegment> path,
+ VariantRef* output) {
+ return find_materialized_path_impl(
+ current, path,
+ [](VariantRef object, StringRef key, size_t, VariantRef* found) {
+ return object.object_find(key, found);
+ },
+ output);
+}
+
+bool find_materialized_path_with_index(VariantRef current, uint32_t
dictionary_id,
+ const UnshreddedMetadataIndex&
metadata_index,
+ std::span<const
VariantShreddedPathSegment> path,
+ DorisVector<int64_t>&
resolved_field_ids,
+ VariantRef* output) {
+ DCHECK_LT(dictionary_id, metadata_index.dictionaries.size());
+ DCHECK_EQ(resolved_field_ids.size(), metadata_index.dictionaries.size() *
path.size());
+ constexpr int64_t UNRESOLVED_FIELD_ID = -2;
+ return find_materialized_path_impl(
+ current, path,
+ [&](VariantRef object, StringRef key, size_t position, VariantRef*
found) {
+ int64_t& field_id = resolved_field_ids[dictionary_id *
path.size() + position];
+ bool layout_validated = false;
+ if (field_id == UNRESOLVED_FIELD_ID) {
+ // object_find() validates the object layout before
consulting metadata.
+ static_cast<void>(object.num_elements());
+ layout_validated = true;
+ field_id =
metadata_index.dictionaries[dictionary_id].find_key(key);
+ }
+ if (field_id < 0) {
+ // A cached metadata miss must not hide a corrupt object
in a later row.
+ if (!layout_validated) {
+ static_cast<void>(object.num_elements());
+ }
+ return false;
+ }
+ return
object.object_find_by_id(static_cast<uint32_t>(field_id), found);
+ },
+ output);
+}
+
+struct UnshreddedPathScan {
+ MutableColumnPtr outer_nulls;
+ MutableColumnPtr typed_values;
+ DataTypePtr typed_type;
+ std::shared_ptr<const UnshreddedPathCache> path_cache;
+ int64_t copied_bytes = 0;
+};
+
+enum class UnshreddedTypedKind : uint8_t { UNKNOWN, STRING, INTEGER,
UNSUPPORTED };
+
+class UnshreddedTypedValueBuilder {
+public:
+ UnshreddedTypedValueBuilder(size_t rows, const UnshreddedMetadataIndex&
metadata_index)
+ : _rows(rows),
+ _metadata_index(metadata_index),
+ _inner_nulls(ColumnUInt8::create()),
+ _result_nulls(ColumnUInt8::create()),
+ _validated_metadata(metadata_index.dictionaries.size(), 0) {
+ _inner_nulls->reserve(rows);
+ _result_nulls->reserve(rows);
+ }
+
+ void append_outer_null() { append_null(1); }
+
+ void append_json_null(uint32_t dictionary_id) {
+ if (_typed_kind == UnshreddedTypedKind::UNKNOWN) {
+ _pending_json_null_dictionaries.push_back(dictionary_id);
+ } else if (_typed_kind == UnshreddedTypedKind::INTEGER &&
+ !validate_integer_metadata(dictionary_id)) {
+ mark_unsupported();
+ }
+ append_null(0);
+ }
+
+ void append_scalar(const VariantRef& found, uint32_t dictionary_id, size_t
row) {
+ if (_typed_kind == UnshreddedTypedKind::UNSUPPORTED) {
+ append_null(0);
+ return;
+ }
+
+ const VariantBasicType basic_type = found.basic_type();
+ const bool is_string = basic_type == VariantBasicType::SHORT_STRING ||
+ (basic_type == VariantBasicType::PRIMITIVE &&
+ found.primitive_id() ==
VariantPrimitiveId::STRING);
+ if (is_string) {
+ if (!prepare(UnshreddedTypedKind::STRING, row)) {
+ append_null(0);
+ return;
+ }
+ const StringRef string = found.get_string();
+
assert_cast<ColumnString&>(*_typed_values).insert_data(string.data,
string.size);
+ _inner_nulls->insert_value(0);
+ _result_nulls->insert_value(0);
+ DCHECK_LE(string.size,
+ static_cast<size_t>(std::numeric_limits<int64_t>::max()
- _copied_bytes));
+ _copied_bytes += static_cast<int64_t>(string.size);
+ return;
+ }
+
+ const auto primitive_id = basic_type == VariantBasicType::PRIMITIVE
+ ? found.primitive_id()
+ : VariantPrimitiveId::NULL_VALUE;
+ const bool is_integer = primitive_id == VariantPrimitiveId::INT8 ||
+ primitive_id == VariantPrimitiveId::INT16 ||
+ primitive_id == VariantPrimitiveId::INT32 ||
+ primitive_id == VariantPrimitiveId::INT64;
+ const int64_t integer = is_integer ? found.get_int() : 0;
+ // Typed Variant integers are re-encoded using the narrowest width.
Keep explicitly widened
+ // source integers on the encoded path so observable physical types
remain unchanged.
+ const bool has_canonical_width =
+ is_integer &&
VariantScalarRef::integer(integer).encoded_size() == found.value.size;
+ if (!has_canonical_width || !validate_integer_metadata(dictionary_id)
||
+ !prepare(UnshreddedTypedKind::INTEGER, row)) {
+ mark_unsupported();
Review Comment:
[P1] Validate selected containers before returning a lazy subtree
This unsupported fallback can publish a selected array before validating its
child boundaries. For example, an otherwise valid parent can contain a
two-element array with offsets `[0,1,1]` and one NULL byte: its declared outer
size is exact, but the second `array_at()` is invalid. Selecting that array
produces a non-null shredded value, so `element_at(root, 'array') IS NOT NULL`
returns true; the former `VariantBatchBuilder::import_array()` rejected it.
Please validate the selected container before deferring it, and cover malformed
interior offsets with a null-only consumer.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -902,6 +1515,103 @@ class ParquetVariantShreddedState final : public
VariantShreddedState {
}
private:
+ template <typename PathCacheSelector>
+ std::shared_ptr<ParquetVariantShreddedState> select_state(
+ ColumnPtr selected_physical, const PathCacheSelector&
select_path_cache) const {
+ std::shared_ptr<const UnshreddedPathCache> path_cache;
+ {
+ std::lock_guard lock(_materialization_lock);
+ path_cache = _unshredded_path_cache;
+ }
+
+ std::shared_ptr<const UnshreddedPathCache> selected_path_cache;
+ if (path_cache) {
+ selected_path_cache = select_path_cache(*path_cache);
+ }
+ return std::make_shared<ParquetVariantShreddedState>(
+ _schema, std::move(selected_physical), _complete, _profile,
_unshredded_prefix,
+ std::move(selected_path_cache));
+ }
+
+ std::vector<OwnedShreddedPathSegment> combined_unshredded_path(
+ std::span<const VariantShreddedPathSegment> suffix) const {
+ std::vector<OwnedShreddedPathSegment> combined = _unshredded_prefix;
+ combined.reserve(combined.size() + suffix.size());
+ for (const auto& segment : suffix) {
+ combined.push_back(
+ {.kind = segment.kind,
+ .key = segment.kind ==
VariantShreddedPathSegment::Kind::OBJECT_KEY
+ ? (segment.key.size == 0
+ ? std::string()
+ : std::string(segment.key.data,
segment.key.size))
+ : std::string(),
+ .index = segment.index});
+ }
+ return combined;
+ }
+
+ static std::vector<VariantShreddedPathSegment> borrow_unshredded_path(
+ const std::vector<OwnedShreddedPathSegment>& owned) {
+ std::vector<VariantShreddedPathSegment> borrowed;
+ borrowed.reserve(owned.size());
+ for (const auto& segment : owned) {
+ borrowed.push_back({.kind = segment.kind,
+ .key = {segment.key.data(),
segment.key.size()},
+ .index = segment.index});
+ }
+ return borrowed;
+ }
+
+ std::shared_ptr<const UnshreddedMetadataIndex>
get_unshredded_metadata_index(
+ std::pair<size_t, size_t> child_indices) const {
+ std::lock_guard lock(_materialization_lock);
+ if (!_unshredded_metadata_index) {
+ _unshredded_metadata_index =
+ build_unshredded_metadata_index(*_schema, *_physical,
child_indices);
+ }
+ return _unshredded_metadata_index;
+ }
+
+ ColumnPtr direct_unshredded_path(std::span<const
VariantShreddedPathSegment> suffix) const {
+ const auto child_indices = unshredded_child_indices(*_schema);
+ DORIS_CHECK(child_indices.has_value());
+ std::vector<OwnedShreddedPathSegment> combined =
combined_unshredded_path(suffix);
+ const auto borrowed_combined = borrow_unshredded_path(combined);
+ std::shared_ptr<const UnshreddedMetadataIndex> metadata_index;
+ UnshreddedPathScan scan;
+ {
+ SCOPED_TIMER(_profile.variant_unshredded_direct_seek_time.get());
+ metadata_index = get_unshredded_metadata_index(*child_indices);
+ scan = scan_unshredded_path(
+ *_schema, *_physical, *child_indices, *metadata_index,
+ _unshredded_path_cache
+ ? suffix
+ : std::span<const
VariantShreddedPathSegment>(borrowed_combined),
+ _unshredded_path_cache.get());
+ }
+
+ const auto rows = static_cast<int64_t>(_physical->size());
+ update_counter(_profile.variant_unshredded_direct_seek_rows, rows);
+ if (_unshredded_path_cache) {
+ update_counter(_profile.variant_unshredded_prefix_reuse_rows,
rows);
+ }
+ MutableColumnPtr values;
+ std::shared_ptr<ParquetVariantShreddedState> subtree_state;
+ if (scan.typed_values) {
+ values =
ColumnVariantV2::create_typed(std::move(scan.typed_values),
+ std::move(scan.typed_type));
+ update_counter(_profile.variant_unshredded_direct_seek_bytes,
scan.copied_bytes);
+ update_counter(_profile.variant_direct_leaf_rows, rows);
+ } else {
+ subtree_state = std::make_shared<ParquetVariantShreddedState>(
Review Comment:
[P2] Reuse the metadata index across chained subtree states
This subtree retains the identical immutable `_physical` column and already
carries the prefix-offset cache, but it drops the metadata index that was just
built for the scan. The next chained `element_at` therefore walks every row and
compares/hashes the complete metadata bytes again before it can use the prefix,
making a depth-D chain O(D*R*M) in rows and metadata bytes and allocating a
fresh row-ID vector at each step. Please pass the shared index into states that
keep the same physical identity; filter/cut/gather can still rebuild it and
append can reset it.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -878,6 +1473,24 @@ class ParquetVariantShreddedState final : public
VariantShreddedState {
ErrorCode::INTERNAL_ERROR,
"A projected Parquet Variant can only serve its validated
shredded leaves");
}
+ if (!_unshredded_prefix.empty()) {
+ if (!_normalized_prefix) {
+ const auto child_indices = unshredded_child_indices(*_schema);
+ DORIS_CHECK(child_indices.has_value());
+ const auto borrowed =
borrow_unshredded_path(_unshredded_prefix);
+ int64_t copied_bytes = 0;
+ {
+
SCOPED_TIMER(_profile.variant_unshredded_direct_seek_time.get());
+ _normalized_prefix = normalize_unshredded_path(
Review Comment:
[P2] Consume the cached offsets when materializing this subtree
`direct_unshredded_path()` already computed and retained each selected
value's exact root-relative offset and size. This call ignores that cache and
starts from every raw root again, re-walking the complete accumulated prefix
before it can copy the same values. A depth-D chain ending in a
BINARY/BOOL/DECIMAL/container/mixed value therefore adds redundant O(R*D)
lookup work when a cast, serialization, hash, or `get_value_ref()` forces
materialization. Please normalize directly from `_unshredded_path_cache` (while
applying the selected-value validators) and add a test or counter proving the
prefix is not resolved twice.
--
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]
