github-actions[bot] commented on code in PR #66446:
URL: https://github.com/apache/doris/pull/66446#discussion_r3717325026
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -313,6 +315,289 @@ ValidatedTypedInput validate_typed_input(ColumnPtr
column, DataTypePtr scalar_ty
"ColumnVariantV2::{} is intentionally unsupported for
Variant values", method);
}
+class CompositeVariantShreddedState final : public VariantShreddedState {
+public:
+ explicit CompositeVariantShreddedState(
+ std::vector<std::shared_ptr<VariantShreddedState>> segments)
+ : _segments(std::move(segments)) {
+ DORIS_CHECK(std::ranges::all_of(_segments, [](const auto& segment) {
+ return segment != nullptr;
+ })) << "composite Variant shredded segments must not be null";
+ }
+
+ size_t size() const override {
Review Comment:
[P2] Keep composite row counts constant-time
`size()` now walks every segment, but
`extract_shredded_typed_variant_element()` evaluates `source.size()` in its
loop condition for every output row. Alternating incompatible file/batch states
can retain one segment per row, turning normal element extraction into
Theta(rows * segments), or quadratic in the worst case, before it even returns
the leaf. Please maintain a checked cached row count in the composite (or cache
the row count at every hot caller) and keep the full sum in sanity checks.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -552,6 +563,72 @@ bool supports_direct_typed_variant_state(const
ParquetColumnSchema& schema) {
}
}
+ColumnPtr normalize_projected_primitive_leaf(const ParquetColumnSchema& schema,
+ const ColumnPtr& typed) {
+ const auto& nullable = assert_cast<const ColumnNullable&>(*typed);
+ VariantBatchBuilder builder(VariantBatchBuilder::ReserveHint {.rows =
nullable.size()});
+ for (size_t row = 0; row < nullable.size(); ++row) {
+ auto output_row = builder.begin_row();
+ if (nullable.get_null_map_data()[row] != 0) {
+ output_row.add_null();
+ } else {
+ append_typed_scalar(schema, nullable.get_nested_column(), row,
output_row);
+ }
+ output_row.finish();
+ }
+ auto values = ColumnVariantV2::create();
+ values->insert_encoded_batch(builder.finish_batch());
+ auto nulls = nullable.get_null_map_column().clone_resized(nullable.size());
+ 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) {
+ if (segment.kind == VariantShreddedPathSegment::Kind::OBJECT_KEY) {
+ if (current.basic_type() != VariantBasicType::OBJECT ||
+ !current.object_find(segment.key, ¤t)) {
+ return false;
+ }
+ continue;
+ }
+ if (current.basic_type() != VariantBasicType::ARRAY) {
+ return false;
+ }
+ const int64_t count = current.num_elements();
+ const int64_t index = segment.index < 0 ? count + segment.index :
segment.index;
+ if (index < 0 || index >= count) {
+ return false;
+ }
+ current = current.array_at(static_cast<uint32_t>(index));
+ }
+ *output = current;
+ return true;
+}
+
+ColumnPtr normalize_materialized_path(const ColumnVariantV2& materialized,
+ std::span<const
VariantShreddedPathSegment> path) {
+ VariantBatchBuilder builder(VariantBatchBuilder::ReserveHint {.rows =
materialized.size()});
+ auto nulls = ColumnUInt8::create();
+ nulls->reserve(materialized.size());
+ for (size_t row = 0; row < materialized.size(); ++row) {
+ auto output_row = builder.begin_row();
+ VariantRef value;
+ if (find_materialized_path(materialized.get_value_ref(row), path,
&value)) {
+ output_row.add_value(value);
Review Comment:
[P1] Preserve widths in the materialized-path fallback
When a complete segment has a present residual value,
`find_normalized_value()` reaches this helper. `add_value()` reimports
INT/DECIMAL VariantRefs through `VariantBatchBuilder::import_primitive()`,
which sends all integer IDs through width-inferencing `add_int()` and decimals
through width-inferencing `decimal()`. A small INT64 or DECIMAL16 value is
consequently narrowed when this segment is combined with a projected neighbor,
even though full reconstruction retained its primitive ID. This is a separate
fallback from the direct projected normalization in r3713652174; please
preserve the source physical ID here and cover complete-with-residual plus
projected segments in both orders.
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -313,6 +315,289 @@ ValidatedTypedInput validate_typed_input(ColumnPtr
column, DataTypePtr scalar_ty
"ColumnVariantV2::{} is intentionally unsupported for
Variant values", method);
}
+class CompositeVariantShreddedState final : public VariantShreddedState {
+public:
+ explicit CompositeVariantShreddedState(
+ std::vector<std::shared_ptr<VariantShreddedState>> segments)
+ : _segments(std::move(segments)) {
+ DORIS_CHECK(std::ranges::all_of(_segments, [](const auto& segment) {
+ return segment != nullptr;
+ })) << "composite Variant shredded segments must not be null";
+ }
+
+ size_t size() const override {
+ size_t rows = 0;
+ for (const auto& segment : _segments) {
+ DORIS_CHECK_LE(segment->size(), std::numeric_limits<size_t>::max()
- rows)
+ << "composite Variant shredded row count overflows size_t";
+ rows += segment->size();
+ }
+ return rows;
+ }
+
+ size_t byte_size() const override {
+ size_t bytes = 0;
+ for (const auto& segment : _segments) {
+ bytes += segment->byte_size();
+ }
+ std::lock_guard lock(_materialization_lock);
+ return bytes + (_materialized ? _materialized->byte_size() : 0) +
+ (_serialized ? _serialized->byte_size() : 0);
+ }
+
+ size_t allocated_bytes() const override {
+ size_t bytes = 0;
+ for (const auto& segment : _segments) {
+ bytes += segment->allocated_bytes();
+ }
+ std::lock_guard lock(_materialization_lock);
+ return bytes + (_materialized ? _materialized->allocated_bytes() : 0) +
+ (_serialized ? _serialized->allocated_bytes() : 0);
+ }
+
+ void sanity_check() const override {
+ for (const auto& segment : _segments) {
+ segment->sanity_check();
+ }
+ }
+
+ void for_each_subcolumn(const IColumn::ImutableColumnCallback& callback)
const override {
+ for (const auto& segment : _segments) {
+ segment->for_each_subcolumn(callback);
+ }
+ }
+
+ std::shared_ptr<VariantShreddedState> filter(const IColumn::Filter& filter,
+ ssize_t /*result_size_hint*/)
const override {
+ DORIS_CHECK_EQ(filter.size(), size())
+ << "composite Variant shredded filter size does not match row
count";
+ std::vector<std::shared_ptr<VariantShreddedState>> selected;
+ selected.reserve(_segments.size());
+ size_t offset = 0;
+ for (const auto& segment : _segments) {
+ IColumn::Filter segment_filter;
+ segment_filter.insert(filter.begin() + offset,
+ filter.begin() + offset + segment->size());
+ auto filtered = segment->filter(segment_filter, -1);
+ if (filtered->size() != 0) {
+ selected.push_back(std::move(filtered));
+ }
+ offset += segment->size();
+ }
+ return pack(std::move(selected));
+ }
+
+ std::shared_ptr<VariantShreddedState> select_range(size_t start, size_t
length) const override {
+ DORIS_CHECK_LE(start, size()) << "composite Variant range starts past
source size";
+ DORIS_CHECK_LE(length, size() - start) << "composite Variant range
exceeds source size";
+ std::vector<std::shared_ptr<VariantShreddedState>> selected;
+ if (length == 0) {
+ return pack(std::move(selected));
+ }
+ const size_t end = start + length;
+ size_t offset = 0;
+ for (const auto& segment : _segments) {
+ const size_t segment_end = offset + segment->size();
+ const size_t overlap_begin = std::max(start, offset);
+ const size_t overlap_end = std::min(end, segment_end);
+ if (overlap_begin < overlap_end) {
+ selected.push_back(
+ segment->select_range(overlap_begin - offset,
overlap_end - overlap_begin));
+ }
+ offset = segment_end;
+ if (offset >= end) {
+ break;
+ }
+ }
+ return pack(std::move(selected));
+ }
+
+ std::shared_ptr<VariantShreddedState> select_indices(
Review Comment:
[P1] Preserve projected state through local TopN row selection
This selection API is not used by local TopN: `sort_block()` calls
`permute()` on every payload column, while `ColumnVariantV2::permute()` still
calls `_shredded->materialized_column()`. With a mapper-eligible leaf such as
`projected['shared']`, a two-phase TopN therefore throws on the incomplete
Parquet state before it reaches the exchange, so the new wire serializer cannot
help. Please route shredded permutation (and truncation) through
`select_indices()`/`select_range()` or fence projection before sort, and add a
forced two-phase TopN case with only eligible top-level paths.
--
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]
