github-actions[bot] commented on code in PR #66674:
URL: https://github.com/apache/doris/pull/66674#discussion_r3770471052
##########
be/src/core/data_type_serde/data_type_variant_v2_serde.cpp:
##########
@@ -208,71 +687,260 @@ char* DataTypeVariantV2SerDe::serialize(const IColumn&
column, char* buf, int be
buf = serialize_const_flag_and_row_num(&physical, buf, &saved_rows);
const auto& variant = assert_cast<const ColumnVariantV2&>(*physical);
DCHECK_EQ(variant.size(), saved_rows);
- unaligned_store<bool>(buf, variant.is_typed());
- buf += sizeof(bool);
- if (variant.is_typed()) {
- write_variant_v2_type(variant._typed_type, buf);
- return make_nullable(variant._typed_type)->serialize(*variant._typed,
buf, be_exec_version);
+
+ if (!variant.is_shredded()) {
+ return serialize_non_shredded_payload(variant, buf, be_exec_version);
}
- const DataTypeString string_type;
- buf = string_type.serialize(*variant._metadatas, buf, be_exec_version);
- buf = serialize_meta_ids(*variant._meta_ids, buf);
- return string_type.serialize(*variant._values, buf, be_exec_version);
+
+ unaligned_store<uint8_t>(buf,
static_cast<uint8_t>(VariantV2WireRepresentation::SHREDDED));
Review Comment:
[P1] Preserve the legacy wire format for older BEs. This branch emits the
new SHREDDED tag for every `be_exec_version`, but whole-root Variant V2 scans
can produce S and send it through remote exchange or tablet-writer RPCs during
a rolling upgrade. Both base and head advertise exec version 11, and the base
reader treats this byte as the old `bool typed` discriminator, so tag 2 makes
it misparse the following S header. `ShreddedWireDoesNotDependOnBeExecVersion`
uses patched code on both sides and actually locks tag 2 in for version 10.
Please introduce a new execution-version capability, serialize an encoded/tag-0
snapshot below it, and cover the legacy reader (or an equivalent golden
fixture).
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -315,20 +691,463 @@ ValidatedTypedInput validate_typed_input(ColumnPtr
column, DataTypePtr scalar_ty
} // namespace
+template <typename Selection>
+void ColumnVariantV2::_append_source_fields_to_residual(
+ ColumnVariantV2& residual, const ColumnVariantV2& source,
+ const DorisVector<const ShreddedField*>& fields_to_residual, const
Selection& selection) {
+ DORIS_CHECK(residual.is_encoded());
+ DorisVector<uint8_t> rows_needing_merge(selection.size(), 0);
+ for (size_t output_row = 0; output_row < selection.size(); ++output_row) {
+ const size_t source_row = selection.source_row(output_row);
+ for (const auto* field : fields_to_residual) {
+ if (field->presence->get_data()[source_row] != 0) {
+ rows_needing_merge[output_row] = 1;
+ break;
+ }
+ }
+ }
+
+ const auto append_direct_run = [&](size_t run_begin, size_t run_end) {
+ if constexpr (std::is_same_v<Selection, ShreddedRangeSelection>) {
+ residual._append_encoded_range(source, selection.start + run_begin,
+ run_end - run_begin);
+ } else {
+ static_assert(std::is_same_v<Selection, ShreddedIndicesSelection>);
+ residual._append_encoded_indices(source, selection.begin +
run_begin,
+ selection.begin + run_end);
+ }
+ };
+
+#ifdef BE_TEST
+ size_t slow_rows = 0;
+#endif
+ ShreddedFieldRefs present_fields;
+ present_fields.reserve(fields_to_residual.size());
+ const auto view = source.read_view();
+ for (size_t run_begin = 0; run_begin < selection.size();) {
+ const bool needs_residual_merge = rows_needing_merge[run_begin] != 0;
+ size_t run_end = run_begin + 1;
+ while (run_end < selection.size() &&
+ (rows_needing_merge[run_end] != 0) == needs_residual_merge) {
+ ++run_end;
+ }
+ if (!needs_residual_merge) {
+ append_direct_run(run_begin, run_end);
+ run_begin = run_end;
+ continue;
+ }
+
+ VariantBatchBuilder builder({.rows = run_end - run_begin});
+ for (size_t output_row = run_begin; output_row < run_end;
++output_row) {
+ const size_t source_row = selection.source_row(output_row);
+ present_fields.clear();
+ for (const auto* field : fields_to_residual) {
+ if (field->presence->get_data()[source_row] != 0) {
+ present_fields.push_back(field);
+ }
+ }
+ DORIS_CHECK(!present_fields.empty());
+ auto output = builder.begin_row();
+ append_merged_shredded_node(output,
view.residual_value_at(source_row), present_fields,
+ 0, present_fields.size(), 0,
source_row);
+ output.finish();
+ }
+ VariantBatchBuilder encoded = builder.finish_batch();
+ residual.insert_encoded_batch(encoded);
+#ifdef BE_TEST
+ slow_rows += run_end - run_begin;
+#endif
+ run_begin = run_end;
+ }
+#ifdef BE_TEST
+ _test_shredded_conflict_slow_rows += slow_rows;
+#endif
+}
+
+template <typename Selection>
+void ColumnVariantV2::_append_same_shredded_layout_rows(const ColumnVariantV2&
source,
+ const Selection&
selection) {
+ DCHECK(is_shredded());
+ DCHECK(source.is_shredded());
+ DCHECK_EQ(_shredded_fields.size(), source._shredded_fields.size());
+ if constexpr (std::is_same_v<Selection, ShreddedRangeSelection>) {
+ _append_encoded_range(source, selection.start, selection.size());
+ } else {
+ static_assert(std::is_same_v<Selection, ShreddedIndicesSelection>);
+ _append_encoded_indices(source, selection.begin, selection.end);
+ }
+ for (size_t index = 0; index < _shredded_fields.size(); ++index) {
+ auto& destination_field = _shredded_fields[index];
+ const auto& source_field = source._shredded_fields[index];
+ const auto& destination_values =
+ assert_cast<const ColumnVariantV2&>(*destination_field.values);
+ if (destination_values.is_typed()) {
+ DCHECK(shredded_children_append_compatible(destination_field,
source_field));
+ } else {
+ const auto& source_values = assert_cast<const
ColumnVariantV2&>(*source_field.values);
+ if (source_values.is_typed() &&
+ !has_present_selected_row(*source_field.presence, selection)) {
+ _append_missing_shredded_field(destination_field,
selection.size());
+ continue;
+ }
+ }
+ mutate_subcolumn(destination_field.values);
+ mutate_subcolumn<ColumnUInt8>(destination_field.presence);
+ selection.insert_from(*destination_field.values, *source_field.values);
Review Comment:
[P1] Do not encode inactive typed padding during S-to-S append. An encoded
destination child can reach this call with a typed source child, and
`selection.insert_from` converts every selected T row even when the source
field's presence bit is 0. Valid S columns may keep arbitrary padding there
(the new invalid-DATE element-at test demonstrates this contract), so DATE
conversion can throw on an unobservable row. Because the residual was already
appended, this path also leaves the destination with mismatched row counts
after the exception. Please mask absent rows to canonical padding and make the
direct append atomic; cover range and indices paths with inactive invalid typed
padding.
##########
be/src/storage/segment/variant/variant_column_writer_impl.cpp:
##########
@@ -1993,15 +1994,66 @@ Status VariantSubcolumnWriter::_append_v2(const
VariantColumnData& column, size_
return Status::OK();
};
- const auto view = source->read_view();
- if (!view.is_typed()) {
- return append_encoded(view, column.row_pos);
+ if (source->is_encoded()) {
+ const Status status = append_encoded(source->read_view(),
column.row_pos);
+ if (status.ok() && append_stats != nullptr) {
+ *append_stats = {};
+ }
+ return status;
}
- auto encoded_batch = ColumnVariantV2::create();
+ if (source->is_shredded()) {
+ VariantShredderAppendStats result_stats;
+ const ColumnVariantV2::ReadView view = source->read_view();
+ for (size_t offset = 0; offset < num_rows; ++offset) {
+ if (!outer_nulls.empty() && outer_nulls[offset] != 0) {
+ ++result_stats.native_shredded_rows;
+ continue;
+ }
+ const size_t input_row = column.row_pos + offset;
+ bool has_active_field = false;
+ for (size_t field = 0; field < view.shredded_field_count();
++field) {
+ if (view.shredded_field_presence(field).get_data()[input_row]
!= 0) {
+ has_active_field = true;
+ break;
+ }
+ }
+ if (!has_active_field) {
+ const VariantRef residual = view.residual_value_at(input_row);
+ if (!residual.is_null()) {
+ RETURN_IF_ERROR(_v2_builder->append(residual, _num_rows +
offset));
+ }
+ ++result_stats.native_shredded_rows;
+ continue;
+ }
+
+ // An extracted subcolumn is a whole-value consumer. Reconstruct
only rows that carry
+ // active shredded fields; missing-field/scalar/array rows use the
residual directly.
+ ColumnVariantV2::MutablePtr encoded_row;
+ RETURN_IF_CATCH_EXCEPTION(
+ { encoded_row =
source->materialize_encoded_range(input_row, 1); });
Review Comment:
[P2] Batch this reconstruction for extracted-column writes. For every active
S row, this loop has already scanned the field-presence layout and then
`materialize_encoded_range(row, 1)` scans it again while allocating a new
builder and temporary column. Dense shredded batches therefore do roughly 2*R*F
probes plus R allocations, and each extracted writer repeats it. Please reuse
one batch encoder/snapshot (or at least materialize active runs) and add a
many-row/many-field test that bounds encoded-range materializations.
##########
be/src/storage/segment/variant/v2/variant_shredder.cpp:
##########
@@ -659,6 +791,173 @@ Status VariantShredder::append(const
ColumnVariantV2::ReadView& view, size_t beg
}
}
+Status VariantShredder::append_shredded(const ColumnVariantV2& source, size_t
begin, size_t length,
+ std::span<const uint8_t> outer_nulls,
+ VariantShredderAppendStats*
append_stats) {
+ RETURN_IF_ERROR(_impl->require_collecting());
+ const ColumnVariantV2::ReadView view = source.read_view();
+ if (!view.is_shredded()) {
+ return _impl->fail(
+ Status::InvalidArgument("Variant shredder shredded append
requires S-state input"));
+ }
+ if (begin > view.size() || length > view.size() - begin) {
+ return _impl->fail(
+ Status::InvalidArgument("Variant shredder range [{}, {})
exceeds input size {}",
+ begin, begin + length, view.size()));
+ }
+ if (!outer_nulls.empty() && outer_nulls.size() != length) {
+ return _impl->fail(
+ Status::InvalidArgument("Variant shredder outer-null span has
{} rows, expected {}",
+ outer_nulls.size(), length));
+ }
+ if (length > std::numeric_limits<size_t>::max() - _impl->rows) {
+ return _impl->fail(Status::InvalidArgument("Variant shredder row count
overflows size_t"));
+ }
+
+ VariantShredderAppendStats result_stats;
+ try {
+ DorisVector<Impl::MetadataPathCache> residual_metadata_caches;
+ residual_metadata_caches.reserve(view.residual_metadata_count());
+ for (size_t metadata_index = 0; metadata_index <
view.residual_metadata_count();
+ ++metadata_index) {
+ residual_metadata_caches.emplace_back(
+
view.residual_metadata_at(static_cast<uint32_t>(metadata_index)));
+ }
+
+ const size_t field_count = view.shredded_field_count();
+ DorisVector<ColumnVariantV2::ReadView> field_views;
+ field_views.reserve(field_count);
+ DorisVector<std::optional<Impl::PathIndex>>
field_path_indices(field_count);
+ DorisVector<size_t> field_canonical_groups(field_count, 0);
+ std::unordered_map<std::string_view, size_t> canonical_field_groups;
+ canonical_field_groups.reserve(field_count);
+ for (size_t field_index = 0; field_index < field_count; ++field_index)
{
+
field_views.emplace_back(view.shredded_field_values(field_index).read_view());
+ const PathInData& field_path =
view.shredded_field_path(field_index);
+ const auto [group, inserted] = canonical_field_groups.emplace(
+ field_path.get_path(), canonical_field_groups.size());
+ static_cast<void>(inserted);
+ field_canonical_groups[field_index] = group->second;
+ }
+ DorisVector<std::string_view>
canonical_group_paths(canonical_field_groups.size());
+ for (const auto& [path, group] : canonical_field_groups) {
+ canonical_group_paths[group] = path;
+ }
+ DorisVector<size_t>
active_group_markers(canonical_field_groups.size(), 0);
+
+ DorisVector<size_t> active_fields;
+ active_fields.reserve(field_count);
+ DorisVector<size_t> active_groups;
+ active_groups.reserve(canonical_field_groups.size());
+ DorisVector<char> encoded_slow_scratch;
+ for (size_t offset = 0; offset < length; ++offset) {
+ if (!outer_nulls.empty() && outer_nulls[offset] != 0) {
+ _impl->append_default_root();
+ ++_impl->rows;
+ ++result_stats.native_shredded_rows;
+ continue;
+ }
+
+ const size_t input_row = begin + offset;
+ active_fields.clear();
+ active_groups.clear();
+ bool duplicate_active_path = false;
+ const size_t row_marker = offset + 1;
+ for (size_t field_index = 0; field_index < field_count;
++field_index) {
+ if
(view.shredded_field_presence(field_index).get_data()[input_row] != 0) {
+ active_fields.push_back(field_index);
+ if
(!_impl->shredded_field_participates(field_views[field_index], input_row)) {
+ continue;
+ }
+ const size_t group = field_canonical_groups[field_index];
+ duplicate_active_path |= active_group_markers[group] ==
row_marker;
+ if (active_group_markers[group] != row_marker) {
+ active_group_markers[group] = row_marker;
+ active_groups.push_back(group);
+ }
+ }
+ }
+
+ const uint32_t metadata_index =
view.residual_metadata_id_at(input_row);
+ if (metadata_index >= residual_metadata_caches.size()) {
+ return _impl->fail(Status::Corruption(
+ "Variant residual row {} metadata index {} exceeds {}
entries", input_row,
+ metadata_index, residual_metadata_caches.size()));
+ }
+ const VariantRef residual = view.residual_value_at(input_row);
+
+ // A literal dotted key and a nested path intentionally collide in
the legacy storage
+ // namespace. Probe only residual branches that can spell a
currently active canonical
+ // path; unrelated dotted keys do not trigger a full residual
pre-scan.
+ Impl::MetadataPathCache& residual_metadata_cache =
+ residual_metadata_caches[metadata_index];
+ bool residual_collision = false;
+ if (!duplicate_active_path) {
+ for (size_t group : active_groups) {
+ const std::string_view canonical_path =
canonical_group_paths[group];
+ if (canonical_path.find('.') != std::string_view::npos &&
+ _impl->residual_contains_canonical_leaf(residual,
canonical_path)) {
Review Comment:
[P2] Avoid rescanning the residual once per active nested field. Reader
assembly deliberately leaves each moved nested leaf's empty ancestor in the
residual, so F paths such as `p_i.leaf` produce F top-level empty objects. This
loop then calls `residual_contains_canonical_leaf` F times, and every call
linearly searches those root children, making the native write path O(F^2) per
row (about four million key comparisons at the default 2048-path limit) even
when no fallback is needed. Please traverse/index the residual once and merge
or probe the active canonical groups against that result, with a wide-S
inspection-count test.
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -1103,7 +2354,12 @@ void ColumnVariantV2::deserialize(StringRef* keys,
size_t num_rows) {
void ColumnVariantV2::update_hash_with_value(size_t row, SipHash& hash) const {
DCHECK_LT(row, size());
- if (_typed) {
+ if (is_shredded()) {
+ auto encoded = materialize_encoded_range(row, 1);
Review Comment:
[P2] Avoid rebuilding an encoded column per hashed S row.
`Block::update_hash` calls this virtual method in a row loop (and
Array/Map/Struct hashing can call it per nested value), so this branch performs
N `VariantBatchBuilder`/column allocations plus N full layout scans. The batch
hash overrides already materialize a range once. Please provide direct
canonical S hashing or a batch-scoped snapshot/cache, and add a multi-row
interface test that bounds materializations.
##########
be/src/core/data_type_serde/data_type_variant_v2_serde.cpp:
##########
@@ -208,71 +687,260 @@ char* DataTypeVariantV2SerDe::serialize(const IColumn&
column, char* buf, int be
buf = serialize_const_flag_and_row_num(&physical, buf, &saved_rows);
const auto& variant = assert_cast<const ColumnVariantV2&>(*physical);
DCHECK_EQ(variant.size(), saved_rows);
- unaligned_store<bool>(buf, variant.is_typed());
- buf += sizeof(bool);
- if (variant.is_typed()) {
- write_variant_v2_type(variant._typed_type, buf);
- return make_nullable(variant._typed_type)->serialize(*variant._typed,
buf, be_exec_version);
+
+ if (!variant.is_shredded()) {
+ return serialize_non_shredded_payload(variant, buf, be_exec_version);
}
- const DataTypeString string_type;
- buf = string_type.serialize(*variant._metadatas, buf, be_exec_version);
- buf = serialize_meta_ids(*variant._meta_ids, buf);
- return string_type.serialize(*variant._values, buf, be_exec_version);
+
+ unaligned_store<uint8_t>(buf,
static_cast<uint8_t>(VariantV2WireRepresentation::SHREDDED));
+ buf += sizeof(uint8_t);
+ unaligned_store<uint8_t>(buf, SHREDDED_WIRE_VERSION);
+ buf += sizeof(uint8_t);
+ char* payload_size_position = buf;
+ buf += sizeof(uint64_t);
+ char* const payload_begin = buf;
+
+ unaligned_store<uint8_t>(buf, variant._shredded_path_set_sealed ? 1 : 0);
+ buf += sizeof(uint8_t);
+ char* residual_size_position = buf;
+ buf += sizeof(uint64_t);
+ char* const residual_begin = buf;
+ buf = serialize_encoded_payload(variant, buf, be_exec_version);
+ unaligned_store<uint64_t>(
+ residual_size_position,
+ checked_wire_length(buf - residual_begin, "shredded residual
payload"));
+ unaligned_store<uint32_t>(
+ buf, checked_wire_count(variant._shredded_fields.size(), "shredded
field"));
+ buf += sizeof(uint32_t);
+ for (const auto& field : variant._shredded_fields) {
+ const auto& parts = field.path.get_parts();
+ unaligned_store<uint32_t>(buf, checked_wire_count(parts.size(), "path
part"));
+ buf += sizeof(uint32_t);
+ for (const auto& part : parts) {
+ unaligned_store<uint32_t>(buf, checked_wire_count(part.key.size(),
"path key byte"));
+ buf += sizeof(uint32_t);
+ if (!part.key.empty()) {
+ std::memcpy(buf, part.key.data(), part.key.size());
+ buf += part.key.size();
+ }
+ unaligned_store<uint8_t>(buf, part.is_nested ? 1 : 0);
+ buf += sizeof(uint8_t);
+ unaligned_store<uint8_t>(buf, part.anonymous_array_level);
+ buf += sizeof(uint8_t);
+ }
+ const auto& presence = static_cast<const
ColumnUInt8::Ptr&>(field.presence)->get_data();
+ unaligned_store<uint64_t>(buf, presence.size());
+ buf += sizeof(uint64_t);
+ if (!presence.empty()) {
+ std::memcpy(buf, presence.data(), presence.size());
+ buf += presence.size();
+ }
+ char* child_size_position = buf;
+ buf += sizeof(uint64_t);
+ char* const child_begin = buf;
+ buf = serialize_non_shredded_payload(assert_cast<const
ColumnVariantV2&>(*field.values),
+ buf, be_exec_version);
+ unaligned_store<uint64_t>(child_size_position,
+ checked_wire_length(buf - child_begin,
"shredded child payload"));
+ }
+ unaligned_store<uint64_t>(payload_size_position,
+ checked_wire_length(buf - payload_begin,
"shredded payload"));
+ return buf;
}
const char* DataTypeVariantV2SerDe::deserialize(const char* buf,
MutableColumnPtr* column,
int be_exec_version) {
+ constexpr size_t REPRESENTATION_OFFSET = VARIANT_V2_COLUMN_HEADER_BYTES;
+ const char* const representation_position = buf + REPRESENTATION_OFFSET;
+ const auto representation = static_cast<VariantV2WireRepresentation>(
+ unaligned_load<uint8_t>(representation_position));
+ if (representation == VariantV2WireRepresentation::SHREDDED) {
+ const char* const payload_size_position = representation_position +
sizeof(uint8_t) * 2;
+ const uint64_t payload_size =
unaligned_load<uint64_t>(payload_size_position);
+ const uintptr_t payload_address =
+ reinterpret_cast<uintptr_t>(payload_size_position) +
sizeof(uint64_t);
+ if (payload_size > std::numeric_limits<uintptr_t>::max() -
payload_address) {
+ throw Exception(Status::Corruption(
+ "Shredded ColumnVariantV2 payload length {} overflows
address space",
+ payload_size));
+ }
+ return deserialize(buf, reinterpret_cast<const char*>(payload_address
+ payload_size),
+ column, be_exec_version);
+ }
+
auto* destination = assert_cast<ColumnVariantV2*>(column->get());
- size_t saved_rows = 0;
- buf = deserialize_const_flag_and_row_num(buf, column, &saved_rows);
- const bool typed = unaligned_load<bool>(buf);
+ const auto is_const_flag = unaligned_load<uint8_t>(buf);
+ if (is_const_flag > 1) {
+ throw Exception(
+ Status::Corruption("ColumnVariantV2 has invalid const flag
{}", is_const_flag));
+ }
+ const bool is_const = is_const_flag != 0;
buf += sizeof(bool);
+ const auto logical_rows = unaligned_load<size_t>(buf);
+ buf += sizeof(size_t);
+ const auto saved_rows = unaligned_load<size_t>(buf);
+ buf += sizeof(size_t);
+ if ((is_const && saved_rows != 1) || (!is_const && saved_rows !=
logical_rows)) {
+ throw Exception(Status::Corruption(
+ "ColumnVariantV2 invalid row header: const={}, logical
rows={}, saved rows={}",
+ is_const, logical_rows, saved_rows));
+ }
- ColumnVariantV2::MutablePtr decoded;
- if (typed) {
- DataTypePtr type = read_variant_v2_type(buf);
- const DataTypePtr nullable_type = make_nullable(type);
- MutableColumnPtr typed_column = nullable_type->create_column();
- buf = nullable_type->deserialize(buf, &typed_column, be_exec_version);
- decoded = ColumnVariantV2::create_typed(std::move(typed_column),
std::move(type));
+ MutableColumnPtr decoded;
+ if (representation != VariantV2WireRepresentation::ENCODED &&
+ representation != VariantV2WireRepresentation::TYPED_SCALAR) {
+ throw Exception(Status::Corruption("Unknown ColumnVariantV2
representation tag {}",
+
static_cast<uint8_t>(representation)));
+ }
+ buf = deserialize_non_shredded_payload(buf, &decoded, be_exec_version);
+
+ const auto& decoded_variant = assert_cast<const
ColumnVariantV2&>(*decoded);
+ if (decoded_variant.size() != saved_rows) {
+ throw Exception(Status::Corruption(
+ "ColumnVariantV2 saved row count {} does not match decoded row
count {}",
+ saved_rows, decoded_variant.size()));
+ }
+ if (is_const) {
+ ColumnPtr decoded_data = std::move(decoded);
+ *column = ColumnConst::create(std::move(decoded_data), logical_rows);
} else {
- const DataTypeString string_type;
- MutableColumnPtr metadatas = string_type.create_column();
- MutableColumnPtr meta_ids = MetaIdsColumn::create();
- MutableColumnPtr values = string_type.create_column();
- buf = string_type.deserialize(buf, &metadatas, be_exec_version);
- buf = deserialize_meta_ids(buf, &meta_ids);
- buf = string_type.deserialize(buf, &values, be_exec_version);
-
- const auto& ids = assert_cast<const
MetaIdsColumn&>(*meta_ids).get_data();
- if (ids.size() != values->size()) {
+
destination->_adopt_state_from(assert_cast<ColumnVariantV2&>(*decoded));
+ }
+ return buf;
+}
+
+const char* DataTypeVariantV2SerDe::deserialize(const char* buf, const char*
end,
+ MutableColumnPtr* column, int
be_exec_version) {
+ if (buf == nullptr || end == nullptr || end < buf) {
+ throw Exception(Status::Corruption("ColumnVariantV2 has an invalid
wire buffer"));
+ }
+
+ ShreddedWireCursor cursor(buf, static_cast<size_t>(end - buf));
+ const auto header = read_serialized_column_header(cursor, "column header");
+ const char* const representation_position = cursor.position();
+ const auto representation =
+
static_cast<VariantV2WireRepresentation>(cursor.read<uint8_t>("representation
header"));
+
+ MutableColumnPtr decoded;
+ const char* result = nullptr;
+ if (representation == VariantV2WireRepresentation::ENCODED ||
+ representation == VariantV2WireRepresentation::TYPED_SCALAR) {
+ result = deserialize_non_shredded_payload(representation_position,
&decoded,
+ be_exec_version);
+ if (result > end) {
+ throw Exception(
+ Status::Corruption("ColumnVariantV2 payload exceeds the
provided wire buffer"));
+ }
+ } else if (representation == VariantV2WireRepresentation::SHREDDED) {
+ const auto wire_version = cursor.read<uint8_t>("shredded wire
version");
+ if (wire_version != SHREDDED_WIRE_VERSION_WITHOUT_LAYOUT_STATE &&
+ wire_version != SHREDDED_WIRE_VERSION) {
+ throw Exception(Status::Corruption(
+ "Unsupported shredded ColumnVariantV2 wire version {}",
wire_version));
+ }
+ const auto payload_size = cursor.read<uint64_t>("shredded payload
length");
+ const auto payload = cursor.read_bytes(payload_size, "shredded
payload");
+ result = cursor.position();
+
+ ShreddedWireCursor payload_cursor(payload.data(), payload.size());
+ bool path_set_sealed = true;
+ if (wire_version == SHREDDED_WIRE_VERSION) {
+ const uint8_t sealed_flag =
+ payload_cursor.read<uint8_t>("shredded path-set sealed
flag");
+ if (sealed_flag > 1) {
+ throw Exception(Status::Corruption(
+ "Shredded ColumnVariantV2 has invalid path-set sealed
flag {}",
+ sealed_flag));
+ }
+ path_set_sealed = sealed_flag != 0;
+ }
+ const auto residual_size = payload_cursor.read<uint64_t>("residual
payload length");
+ const auto residual_payload = payload_cursor.read_bytes(residual_size,
"residual payload");
+
+ auto residual = ColumnVariantV2::create();
+ const char* const residual_end = deserialize_encoded_payload(
Review Comment:
[P1] Bound the nested decoder before calling it. `residual_payload` is a
framed span, but `deserialize_encoded_payload` receives only its start pointer;
its String/meta-id decoders read embedded headers, resize, and memcpy before
this post-hoc end comparison. A valid outer S payload with an undersized
residual (or forged inner length) can therefore read or allocate beyond the
declared frame instead of returning `CORRUPTION`; the child path below has the
same problem. Please pass `(begin,end)`/a cursor through every nested E/T
decoder and add valid-outer/invalid-inner residual and child cases.
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -668,15 +1801,18 @@ void ColumnVariantV2::get(size_t row, Field& result)
const {
}
VariantField value;
- if (_typed) {
+ if (is_typed()) {
const auto& nullable = assert_cast<const ColumnNullable&>(*_typed);
visit_typed_scalar_column(nullable, _typed_type->get_primitive_type(),
_typed_type->get_scale(), row, row + 1,
[&](size_t, const VariantScalarRef& scalar) {
value =
VariantField::from_scalar(scalar);
});
- } else {
+ } else if (is_encoded()) {
value = VariantField::from_ref(get_value_ref(row));
+ } else {
+ auto encoded = materialize_encoded_range(row, 1);
Review Comment:
[P2] Avoid rebuilding an encoded column in generic `Field` extraction.
`map_agg_v2` calls `get` for every input row and permits Variant values, while
aggregate-key `VARIANT REPLACE` reaches `CopyStore::set_value -> get` per
selected group. Whole-root scans now produce S, so these paths allocate a
builder/column and scan every layout field per row or high-cardinality group.
Please build the owned `VariantField` directly or reuse a batch snapshot (and
preserve S through reader REPLACE where possible), with many-row tests that
bound materializations.
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -969,7 +2199,11 @@ void ColumnVariantV2::insert_data(const char* pos, size_t
length) {
StringRef ColumnVariantV2::serialize_value_into_arena(size_t row, Arena& arena,
const char*& begin)
const {
DCHECK_LT(row, size());
- if (_typed) {
+ if (is_shredded()) {
+ auto encoded = materialize_encoded_range(row, 1);
Review Comment:
[P2] Avoid publishing a temporary encoded column for each arena-serialized S
value. Serialized join/DISTINCT keys call this interface per row, and
Array/Map/Struct serialization calls the S child's size and write interfaces
separately, so a nested value is reconstructed at least twice (builder
allocation plus full layout scan each time). Please implement direct canonical
S sizing/writing or reuse a batch-scoped encoded snapshot across both passes,
and add direct plus nested arena-serialization tests that bound
materializations.
##########
be/src/storage/segment/variant/v2/variant_assembler.cpp:
##########
@@ -381,6 +463,51 @@ Status append_merge_value(const MergeValue& value, size_t
row, ObjectEmitter* em
value.cell);
}
+Status append_shredded_merge_value(const MergeValue& value, size_t row,
+ VariantShreddedColumnBuilder::Batch*
shredded_batch,
+ ObjectEmitter* emitter, bool* routed) {
+ DORIS_CHECK(shredded_batch != nullptr);
+ DORIS_CHECK(routed != nullptr);
+ *routed = false;
+ const std::optional<size_t> path_index =
+ value.materialized_slot != nullptr
+ ? value.materialized_slot->shredded_path_index
+ : shredded_batch->find_raw_path(value.raw_path,
value.depth);
+ if (!path_index.has_value()) {
+ return Status::OK();
+ }
+
+ if (value.materialized != nullptr &&
+ is_supported_variant_typed_identity(value.materialized->primitive)) {
+ shredded_batch->append_materialized(*path_index, row);
+ emitter->erase_leaf(value.raw_path, value.is_direct_subtree_value,
value.depth);
+ *routed = true;
+ return Status::OK();
+ }
+
+ // Structural cells must remain in the residual. Decode an exact candidate
once into row-local
+ // scratch so both the scalar decision and residual append consume the
same validated value.
+ VariantBatchBuilder scratch({.rows = 1});
Review Comment:
[P2] Reuse decoding scratch across the scan batch. This branch builds and
publishes a fresh one-row `VariantBatchBuilder` for every sparse candidate that
maps to the fixed layout; the new type/shape-conflict tests make that an
ordinary reachable scan shape. The root-sidecar branch repeats the same work
per row even though production binds every layout path, so `append_root` skips
extraction and only copies the decoded value to the residual. Please decode
through batch-owned scratch/direct classification and route bound-layout roots
directly to the residual, with a many-row test that bounds temporary builder
publications.
--
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]
