github-actions[bot] commented on code in PR #66714:
URL: https://github.com/apache/doris/pull/66714#discussion_r3772664578
##########
be/src/core/column/variant_v2/column_variant_v2.cpp:
##########
@@ -973,6 +974,112 @@ void ColumnVariantV2::insert_encoded_rows( //
NOLINT(readability-function-size)
_check_invariants();
}
+void ColumnVariantV2::insert_encoded_rows(std::span<const VariantRef> rows) {
+ if (rows.empty()) {
+ return;
+ }
+
+ 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 metadata_ids_by_value;
+ DorisVector<VariantMetadataRef> unique_metadatas;
+ DorisVector<uint32_t> source_metadata_ids;
+ DorisVector<StringRef> source_values(rows.size());
+ size_t total_value_bytes = 0;
+ for (size_t row = 0; row < rows.size(); ++row) {
+ const VariantRef value = rows[row];
+ if (value.metadata.data == nullptr && value.metadata.size != 0) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant encoded metadata has a null data pointer
for {} bytes",
+ value.metadata.size);
+ }
+ if (value.value.data == nullptr && value.value.size != 0) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant encoded value has a null data pointer for
{} bytes",
+ value.value.size);
+ }
+
+ const std::string_view metadata_key(
+ value.metadata.data == nullptr ? "" : value.metadata.data,
value.metadata.size);
+ uint32_t source_metadata_id = 0;
+ if (unique_metadatas.empty()) {
+ validate_variant_metadata(value.metadata);
+ unique_metadatas.push_back(value.metadata);
+ } else if (unique_metadatas.size() == 1 &&
metadata_ids_by_value.empty() &&
+ StringRef(unique_metadatas.front().data,
unique_metadatas.front().size) ==
+ StringRef(value.metadata.data,
value.metadata.size)) {
+ // Iceberg files normally share one metadata dictionary across a
batch. Avoid a hash
+ // table and per-row ids until a second distinct dictionary is
actually observed.
+ } else {
+ if (metadata_ids_by_value.empty()) {
+ const VariantMetadataRef first = unique_metadatas.front();
+ metadata_ids_by_value.emplace(
+ std::string_view(first.data == nullptr ? "" :
first.data, first.size), 0);
+ source_metadata_ids.resize(rows.size());
+ }
+ auto metadata_id = metadata_ids_by_value.find(metadata_key);
+ if (metadata_id != metadata_ids_by_value.end()) {
+ source_metadata_id = metadata_id->second;
+ } else {
+ if (unique_metadatas.size() ==
std::numeric_limits<uint32_t>::max()) {
+ throw Exception(
+ ErrorCode::INVALID_ARGUMENT,
+ "Variant encoded metadata dictionary exceeds the
uint32 id limit");
+ }
+ validate_variant_metadata(value.metadata);
+ source_metadata_id =
static_cast<uint32_t>(unique_metadatas.size());
+ unique_metadatas.push_back(value.metadata);
+ metadata_ids_by_value.emplace(metadata_key,
source_metadata_id);
+ }
+ }
+ if (!source_metadata_ids.empty()) {
+ source_metadata_ids[row] = source_metadata_id;
+ }
+ validate_variant_payload(value);
+ source_values[row] = value.value;
+ if (value.value.size > std::numeric_limits<size_t>::max() -
total_value_bytes) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant encoded value bytes exceed the size_t
limit");
+ }
+ total_value_bytes += value.value.size;
+ }
+
+ // Validate the complete input before changing a typed/shredded
destination. Failed and empty
+ // appends must preserve its representation just like the EncodedDataView
overload does.
+ if (_typed || _shredded) {
+ ensure_encoded();
+ }
+ DORIS_CHECK(_typed_type == nullptr) << "encoded state cannot retain a
typed data type";
+ require_exclusive(_meta_ids, "metadata ids");
+ require_exclusive(_values, "values");
+ auto& values = assert_cast<ColumnString&>(*_values);
+ auto& metadata_ids = assert_cast<MetaIdsColumn&>(*_meta_ids);
+ reserve_rows(values, metadata_ids, total_value_bytes, rows.size());
+
+ if (unique_metadatas.size() == 1) {
+ const VariantMetadataRef metadata = unique_metadatas.front();
+ const uint32_t id = _find_or_insert_metadata({metadata.data,
metadata.size});
+ values.insert_many_strings_without_reserve(source_values.data(),
source_values.size());
+ metadata_ids.insert_many_vals(id, rows.size());
+ } else {
+ DorisVector<uint32_t>
destination_metadata_ids(unique_metadatas.size());
+ for (size_t id = 0; id < unique_metadatas.size(); ++id) {
+ const VariantMetadataRef metadata = unique_metadatas[id];
+ destination_metadata_ids[id] =
_find_or_insert_metadata({metadata.data, metadata.size});
Review Comment:
[P2] Avoid quadratic metadata interning
Valid unshredded rows may carry different metadata dictionaries, and the new
exact-byte test shows that even a primitive may have unused metadata keys. If
each primitive row has a distinct unused key, this loop calls
`_find_or_insert_metadata()` once per unique dictionary; that helper linearly
scans the growing destination dictionary, making a 4,096-row batch do about 8.4
million metadata comparisons and larger/appended states O(U^2). The old builder
dropped unused primitive metadata and emitted one empty dictionary, so this
regresses the exact case the new path newly preserves. Please keep exact bytes
but use a hash-backed/amortized-linear destination ID merge across chunks, with
a high-cardinality regression test.
##########
be/src/format_v2/parquet/reader/variant_column_reader.cpp:
##########
@@ -887,9 +996,24 @@ class ParquetVariantShreddedState final : public
VariantShreddedState {
}
if (!_materialized) {
SCOPED_TIMER(_profile.variant_reconstruction_time.get());
- _materialized = encode_variant_column(*_schema, *_physical);
+ VariantMaterializationStats stats;
+ VariantMaterializationStats* stats_output =
+ _profile.variant_unshredded_direct_import_time != nullptr
||
+
_profile.variant_unshredded_direct_import_rows != nullptr ||
+
_profile.variant_unshredded_direct_import_bytes != nullptr
+ ? &stats
+ : nullptr;
+ _materialized = encode_variant_column(*_schema, *_physical, true,
stats_output);
Review Comment:
[P2] Preserve direct-import accounting on failures
The direct-import counters are only published after
`encode_variant_column()` returns, so a validation error drops all of the
stack-local stats. This is observable even for an immediately corrupt row
(`VariantReconstructionTime` records the failed attempt while
`VariantUnshreddedDirectImportTime` stays zero), and a 4,097-row input with a
corrupt last row also loses the time/rows/bytes from the first successfully
imported 4,096-row chunk. That makes failed-scan profiles hide work on exactly
the error exits where attribution is needed. Please publish the timer
exception-safely and flush completed chunk rows/bytes as the chunks finish
(with a late-corruption profile test).
--
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]
