github-actions[bot] commented on code in PR #66446:
URL: https://github.com/apache/doris/pull/66446#discussion_r3713652175
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -897,16 +1167,33 @@ void ColumnVariantV2::insert_indices_from( //
NOLINT(readability-function-size)
return;
}
- if (_shredded) {
- ensure_encoded();
- }
if (!_typed && empty() && _metadatas->empty() && source._shredded) {
// Gather into the native shredded representation for the same reason
as range selection:
// row selection does not require, and may not have, a complete
logical Variant value.
_shredded = source._shredded->select_indices(indices_begin,
indices_end);
_check_invariants();
return;
}
+ if (_shredded && source._shredded) {
+ auto selected_source = source._shredded->select_indices(indices_begin,
indices_end);
+ if (_shredded.use_count() != 1) {
+ _shredded = _shredded->select_range(0, size());
+ }
+ if (_shredded->try_append(*selected_source)) {
+ _check_invariants();
+ return;
+ }
+ if (!_shredded->can_materialize() &&
!selected_source->can_materialize()) {
+ // Exchange channels can gather projected rows from files whose
shredded leaf types
Review Comment:
[P1] Serialize a valid representation for remote exchange
This branch preserves a projected `_shredded` column inside the exchange
channel's `MutableBlock`, but non-local channels immediately pass that block to
the Variant binary SerDe. The SerDe only distinguishes typed from encoded
state; for a shredded column it writes the still-empty metadata/id/value
buffers while saving the nonzero `_shredded->size()`. Deserialization
reconstructs zero rows and throws `saved row count ... does not match decoded
row count ...`. This is reachable with an outer element access above a subquery
`ORDER BY ... LIMIT`: access-path collection still projects the leaf at the
file scan, no project-pushdown rule moves the expression through `LogicalTopN`,
and one-phase TopN requires a remote gather first. Please add an explicit
shredded wire representation, materialize when complete, or fence incomplete
projection from remote exchanges, and cover a multi-backend gather.
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -897,16 +1167,33 @@ void ColumnVariantV2::insert_indices_from( //
NOLINT(readability-function-size)
return;
}
- if (_shredded) {
- ensure_encoded();
- }
if (!_typed && empty() && _metadatas->empty() && source._shredded) {
// Gather into the native shredded representation for the same reason
as range selection:
// row selection does not require, and may not have, a complete
logical Variant value.
_shredded = source._shredded->select_indices(indices_begin,
indices_end);
_check_invariants();
return;
}
+ if (_shredded && source._shredded) {
+ auto selected_source = source._shredded->select_indices(indices_begin,
indices_end);
+ if (_shredded.use_count() != 1) {
+ _shredded = _shredded->select_range(0, size());
+ }
+ if (_shredded->try_append(*selected_source)) {
+ _check_invariants();
+ return;
+ }
+ if (!_shredded->can_materialize() &&
!selected_source->can_materialize()) {
Review Comment:
[P1] Handle complete and incomplete file states together
Projection completeness is decided per file: the mapper keeps a full Variant
when the key/type cannot be leaf-projected, and the Parquet finalizer also
restores a full projection when residual all-null statistics are unavailable. A
gather can therefore adopt an incomplete state from file A and then receive a
complete state from file B. `try_append()` rejects the `_complete` mismatch,
this condition is false, and the fallback calls
`ensure_encoded()`/`materialized_column()` on the incomplete side, which throws
instead of returning the requested rows (the reverse order fails too). The new
binary test manually forces a partial binary plan, whereas the production
mapper makes that file complete, so it misses this path. Please preserve mixed
complete/incomplete segments with a per-segment direct/canonical fallback
(including missing-path nulls), or make the projection decision uniform across
files, and add production-path coverage.
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -313,6 +315,266 @@ 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);
+ }
+
+ 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);
+ }
+
+ 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(
+ const uint32_t* indices_begin, const uint32_t* indices_end) const
override {
+ if (indices_begin == indices_end) {
+ return pack({});
+ }
+ DORIS_CHECK(indices_begin != nullptr && indices_end != nullptr &&
+ indices_begin < indices_end)
+ << "composite Variant indices are invalid";
+
+ std::vector<size_t> segment_ends;
+ segment_ends.reserve(_segments.size());
+ size_t rows = 0;
+ for (const auto& segment : _segments) {
+ rows += segment->size();
+ segment_ends.push_back(rows);
+ }
+
+ std::vector<std::shared_ptr<VariantShreddedState>> selected;
+ const uint32_t* cursor = indices_begin;
+ while (cursor != indices_end) {
+ DORIS_CHECK_LT(*cursor, rows) << "composite Variant source index
is out of range";
+ const size_t segment_index =
+ std::upper_bound(segment_ends.begin(), segment_ends.end(),
*cursor) -
+ segment_ends.begin();
+ const size_t segment_begin = segment_index == 0 ? 0 :
segment_ends[segment_index - 1];
+ DorisVector<uint32_t> local_indices;
+ while (cursor != indices_end && *cursor >= segment_begin &&
+ *cursor < segment_ends[segment_index]) {
+ local_indices.push_back(static_cast<uint32_t>(*cursor -
segment_begin));
+ ++cursor;
+ }
+ selected.push_back(_segments[segment_index]->select_indices(
+ local_indices.data(), local_indices.data() +
local_indices.size()));
+ }
+ return pack(std::move(selected));
+ }
+
+ bool can_materialize() const override {
+ return std::ranges::all_of(_segments,
+ [](const auto& segment) { return
segment->can_materialize(); });
+ }
+
+ bool try_append(const VariantShreddedState& source) override {
+ if (const auto* composite = dynamic_cast<const
CompositeVariantShreddedState*>(&source)) {
+ for (const auto& segment : composite->_segments) {
+ append(segment);
+ }
+ } else {
+ append(source.select_range(0, source.size()));
+ }
+ std::lock_guard lock(_materialization_lock);
+ _materialized.reset();
+ return true;
+ }
+
+ std::optional<VariantShreddedTypedValue> find_typed_value(
+ std::span<const VariantShreddedPathSegment> path) const override {
+ if (_segments.empty()) {
+ return std::nullopt;
+ }
+ std::vector<VariantShreddedTypedValue> matches;
+ matches.reserve(_segments.size());
+ for (const auto& segment : _segments) {
+ auto match = segment->find_typed_value(path);
+ if (!match.has_value()) {
+ return std::nullopt;
+ }
+ matches.push_back(std::move(*match));
+ }
+
+ const bool homogeneous = matches.front().column &&
matches.front().type &&
+ std::ranges::all_of(matches, [&](const auto&
match) {
+ return match.column && match.type &&
!match.normalized &&
+
exact_typed_identity(matches.front().type, match.type);
+ });
+ if (homogeneous) {
+ MutableColumnPtr combined = matches.front().column->clone_empty();
+ for (const auto& match : matches) {
+ combined->insert_range_from(*match.column, 0,
match.column->size());
+ }
+ return VariantShreddedTypedValue {.column = std::move(combined),
+ .type = matches.front().type,
+ .normalized = nullptr};
+ }
+
+ auto values = ColumnVariantV2::create();
+ auto nulls = ColumnUInt8::create();
+ nulls->reserve(size());
+ for (const auto& match : matches) {
+ if (match.normalized) {
+ const auto& nullable = assert_cast<const
ColumnNullable&>(*match.normalized);
+ const auto& variants =
+ assert_cast<const
ColumnVariantV2&>(nullable.get_nested_column());
+ values->insert_range_from(variants, 0, variants.size());
+ nulls->insert_range_from(nullable.get_null_map_column(), 0,
nullable.size());
+ continue;
+ }
+ auto typed = ColumnVariantV2::create_typed(match.column,
match.type);
Review Comment:
[P1] Preserve the Parquet primitive width when normalizing mixed leaves
For two mapper-eligible partial files with different leaf types (for example
INT64 `7` and INT32 `8`), this heterogeneous branch wraps each match as a
generic typed Variant and immediately encodes it. That encoder calls
`VariantScalarRef::integer(value)` without the Parquet width, so both values
become INT8; `TYPE_DECIMAL128I` is likewise always emitted as 16 bytes instead
of the precision-derived Parquet width. The normal Parquet reconstruction path
explicitly passes `integer_width(schema, type)` / `decimal_width(...)`, and the
existing baseline asserts that an INT64 leaf remains
`VariantPrimitiveId::INT64`. Carry a schema-aware normalized value/physical
identity into this branch and add a production-eligible heterogeneous test that
asserts the primitive IDs, not only `get_int()`.
--
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]
