This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new bb02660ca2c branch-4.1:[opt](variant) Avoid rebuilding unshredded
Parquet Variant values (#66714)
bb02660ca2c is described below
commit bb02660ca2c447fdd6a6075cadb8e5fd75b538ef
Author: daidai <[email protected]>
AuthorDate: Fri Aug 14 10:09:13 2026 +0800
branch-4.1:[opt](variant) Avoid rebuilding unshredded Parquet Variant
values (#66714)
### What problem does this PR solve?
Problem Summary:
Import validated unshredded metadata/value bytes directly into
ColumnVariantV2 instead of recursively decoding and re-encoding them.
Add Parquet profile counters for direct-import time, rows, and bytes,
and cover lazy materialization and shredded-path isolation.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../core/column/variant_v2/column_variant_v2.cpp | 335 +++++++++++++++++++--
be/src/core/column/variant_v2/column_variant_v2.h | 34 +++
be/src/format_v2/parquet/parquet_profile.cpp | 9 +
be/src/format_v2/parquet/parquet_profile.h | 9 +
.../parquet/reader/variant_column_reader.cpp | 127 +++++++-
be/test/core/column/column_variant_v2_test.cpp | 174 +++++++++++
.../parquet/variant_column_reader_test.cpp | 202 +++++++++++++
7 files changed, 849 insertions(+), 41 deletions(-)
diff --git a/be/src/core/column/variant_v2/column_variant_v2.cpp
b/be/src/core/column/variant_v2/column_variant_v2.cpp
index 8996372a9d0..4a467582fe9 100644
--- a/be/src/core/column/variant_v2/column_variant_v2.cpp
+++ b/be/src/core/column/variant_v2/column_variant_v2.cpp
@@ -25,6 +25,7 @@
#include <mutex>
#include <string_view>
#include <typeinfo>
+#include <unordered_map>
#include <utility>
#include <vector>
@@ -124,6 +125,20 @@ void reserve_rows(ColumnString& values, MetaIdsColumn&
metadata_ids, size_t valu
metadata_ids.get_data().reserve(metadata_ids.size() + rows);
}
+bool byte_ranges_overlap(StringRef left, StringRef right) {
+ if (left.size == 0 || right.size == 0) {
+ return false;
+ }
+ DCHECK(left.data != nullptr);
+ DCHECK(right.data != nullptr);
+ const auto left_begin = reinterpret_cast<uintptr_t>(left.data);
+ const auto right_begin = reinterpret_cast<uintptr_t>(right.data);
+ // Compare start-address distances instead of forming end pointers, which
also avoids integer
+ // overflow for a malformed borrowed size.
+ return left_begin <= right_begin ? right_begin - left_begin < left.size
+ : left_begin - right_begin < right.size;
+}
+
size_t validate_selected_indices(const uint32_t* indices_begin, const
uint32_t* indices_end,
size_t source_rows) {
if (indices_begin == indices_end) {
@@ -937,40 +952,242 @@ void ColumnVariantV2::insert_encoded_rows( //
NOLINT(readability-function-size)
}
}
- if (_typed) {
- ensure_encoded();
+ _append_prevalidated_encoded_data(data);
+}
+
+struct ColumnVariantV2::EncodedRowsAppender::Impl {
+ using MetadataHashHeads =
+ std::unordered_map<size_t, uint32_t, std::hash<size_t>,
std::equal_to<size_t>,
+ CustomStdAllocator<std::pair<const size_t,
uint32_t>>>;
+
+ explicit Impl(ColumnVariantV2& destination) : column(&destination) {}
+
+ ColumnVariantV2* column;
+ MetadataHashHeads metadata_hash_heads;
+ DorisVector<uint32_t> metadata_hash_next;
+ size_t expected_rows = 0;
+ bool initialized = false;
+#ifdef BE_TEST
+ size_t metadata_comparisons = 0;
+#endif
+};
+
+ColumnVariantV2::EncodedRowsAppender::EncodedRowsAppender(ColumnVariantV2&
column)
+ : _impl(std::make_unique<Impl>(column)) {}
+
+ColumnVariantV2::EncodedRowsAppender::~EncodedRowsAppender() = default;
+
+ColumnVariantV2::EncodedRowsAppender::EncodedRowsAppender(EncodedRowsAppender&&)
noexcept = default;
+
+#ifdef BE_TEST
+size_t ColumnVariantV2::EncodedRowsAppender::metadata_comparisons_for_test()
const noexcept {
+ return _impl == nullptr ? 0 : _impl->metadata_comparisons;
+}
+#endif
+
+void ColumnVariantV2::EncodedRowsAppender::append( //
NOLINT(readability-function-size)
+ std::span<const VariantRef> rows) {
+ if (rows.empty()) {
+ return;
}
- 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, data.value_bytes.size, rows);
+ DORIS_CHECK(_impl != nullptr && _impl->column != nullptr)
+ << "Cannot use a moved-from Variant encoded-row appender";
+ ColumnVariantV2& column = *_impl->column;
+
+ 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);
+ }
- if (data.meta_ids.empty()) {
- const VariantMetadataRef metadata = metadata_at(0);
- const uint32_t id = _find_or_insert_metadata({metadata.data,
metadata.size});
- values.insert_many_continuous_binary_data(data.value_bytes.data,
data.value_offsets.data(),
- rows);
- metadata_ids.insert_many_vals(id, rows);
+ 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 (_impl->initialized) {
+ DORIS_CHECK(column._typed == nullptr && column._shredded == nullptr)
+ << "ColumnVariantV2 changed outside its encoded-row appender";
+ DORIS_CHECK_EQ(_impl->expected_rows, column.size())
+ << "ColumnVariantV2 changed outside its encoded-row appender";
+ const auto& current_metadatas = assert_cast<const ColumnString&>(
+ *static_cast<const IColumn::Ptr&>(column._metadatas));
+ DORIS_CHECK_EQ(_impl->metadata_hash_next.size(),
current_metadatas.size())
+ << "ColumnVariantV2 metadata changed outside its encoded-row
appender";
+ }
+
+ auto aliases_encoded_storage = [&](const ColumnVariantV2& owner) {
+ DORIS_CHECK(owner._typed == nullptr && owner._shredded == nullptr)
+ << "encoded-row owner must have encoded storage";
+ const auto& metadatas = assert_cast<const ColumnString&>(
+ *static_cast<const IColumn::Ptr&>(owner._metadatas));
+ const auto& values = assert_cast<const ColumnString&>(*owner._values);
+ const StringRef metadata_storage(metadatas.get_chars().data(),
+ metadatas.get_chars().size());
+ const StringRef value_storage(values.get_chars().data(),
values.get_chars().size());
+ for (const VariantRef row : rows) {
+ if (byte_ranges_overlap({row.metadata.data, row.metadata.size},
metadata_storage) ||
+ byte_ranges_overlap(row.value, value_storage)) {
+ return true;
+ }
+ }
+ return false;
+ };
+ const ColumnVariantV2* encoded_owner = nullptr;
+ if (column._shredded) {
+ // get_value_ref() on a shredded destination borrows from this cached
materialization.
+ // Check it before ensure_encoded() drops the state that owns those
bytes.
+ encoded_owner = &column._shredded->materialized_column();
+ } else if (!column._typed) {
+ encoded_owner = &column;
+ }
+ if (encoded_owner != nullptr && aliases_encoded_storage(*encoded_owner)) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant encoded-row input must not alias its
destination; use "
+ "insert_range_from() for same-column rows");
+ }
+
+ if (column._typed || column._shredded) {
+ column.ensure_encoded();
+ }
+ DORIS_CHECK(column._typed_type == nullptr) << "encoded state cannot retain
a typed data type";
+ require_exclusive(column._meta_ids, "metadata ids");
+ require_exclusive(column._values, "values");
+ auto& values = assert_cast<ColumnString&>(*column._values);
+ auto& metadata_ids = assert_cast<MetaIdsColumn&>(*column._meta_ids);
+
+ auto current_metadatas = [&]() -> const ColumnString& {
+ return assert_cast<const ColumnString&>(
+ *static_cast<const IColumn::Ptr&>(column._metadatas));
+ };
+ if (!_impl->initialized) {
+ const auto& metadatas = current_metadatas();
+ _impl->metadata_hash_heads.reserve(metadatas.size());
+ _impl->metadata_hash_next.reserve(metadatas.size());
+ for (size_t index = 0; index < metadatas.size(); ++index) {
+ const auto id = static_cast<uint32_t>(index);
+ const StringRef metadata = metadatas.get_data_at(id);
+ const size_t hash = StringRefHash {}(metadata);
+ const auto existing = _impl->metadata_hash_heads.find(hash);
+ const uint32_t next = existing == _impl->metadata_hash_heads.end()
+ ? UNMAPPED_METADATA_ID
+ : existing->second;
+ _impl->metadata_hash_next.push_back(next);
+ _impl->metadata_hash_heads[hash] = id;
+ }
+ _impl->expected_rows = column.size();
+ _impl->initialized = true;
+ }
+ reserve_rows(values, metadata_ids, total_value_bytes, rows.size());
+ auto find_or_insert_metadata = [&](VariantMetadataRef metadata) {
+ const StringRef metadata_bytes(metadata.data, metadata.size);
+ const size_t hash = StringRefHash {}(metadata_bytes);
+ auto existing = _impl->metadata_hash_heads.find(hash);
+ uint32_t id = existing == _impl->metadata_hash_heads.end() ?
UNMAPPED_METADATA_ID
+ :
existing->second;
+ while (id != UNMAPPED_METADATA_ID) {
+#ifdef BE_TEST
+ ++_impl->metadata_comparisons;
+#endif
+ if (current_metadatas().get_data_at(id) == metadata_bytes) {
+ return id;
+ }
+ id = _impl->metadata_hash_next[id];
+ }
+
+ const uint32_t new_id = column._append_metadata(metadata_bytes);
+ const uint32_t next = existing == _impl->metadata_hash_heads.end() ?
UNMAPPED_METADATA_ID
+ :
existing->second;
+ _impl->metadata_hash_next.push_back(next);
+ if (existing == _impl->metadata_hash_heads.end()) {
+ _impl->metadata_hash_heads.emplace(hash, new_id);
+ } else {
+ existing->second = new_id;
+ }
+ return new_id;
+ };
+
+ if (unique_metadatas.size() == 1) {
+ const uint32_t id = find_or_insert_metadata(unique_metadatas.front());
+ values.insert_many_strings_without_reserve(source_values.data(),
source_values.size());
+ metadata_ids.insert_many_vals(id, rows.size());
} else {
- DorisVector<uint32_t> remap(metadata_count, UNMAPPED_METADATA_ID);
+ DorisVector<uint32_t>
destination_metadata_ids(unique_metadatas.size());
+ for (size_t id = 0; id < unique_metadatas.size(); ++id) {
+ destination_metadata_ids[id] =
find_or_insert_metadata(unique_metadatas[id]);
+ }
+ values.insert_many_strings_without_reserve(source_values.data(),
source_values.size());
auto& destination_ids = metadata_ids.get_data();
- for (size_t row = 0; row < rows; ++row) {
- const uint32_t source_id = data.meta_ids[row];
- DCHECK_LT(source_id, metadata_count);
- const VariantMetadataRef metadata = metadata_at(source_id);
- if (remap[source_id] == UNMAPPED_METADATA_ID) {
- remap[source_id] = _find_or_insert_metadata({metadata.data,
metadata.size});
- }
- destination_ids.push_back(remap[source_id]);
+ for (size_t row = 0; row < rows.size(); ++row) {
+
destination_ids.push_back(destination_metadata_ids[source_metadata_ids[row]]);
}
- values.insert_many_continuous_binary_data(data.value_bytes.data,
data.value_offsets.data(),
- rows);
}
- DCHECK_EQ(_meta_ids->size(), _values->size());
- _check_invariants();
+ _impl->expected_rows = column.size();
+ DCHECK_EQ(column._meta_ids->size(), column._values->size());
+ column._check_invariants();
+}
+
+ColumnVariantV2::EncodedRowsAppender
ColumnVariantV2::create_encoded_rows_appender() {
+ return EncodedRowsAppender(*this);
}
void ColumnVariantV2::insert_encoded_batch(const VariantBatchBuilder& block) {
@@ -984,10 +1201,38 @@ void ColumnVariantV2::insert_encoded_batch(const
VariantBatchBuilder& block) {
const VariantMetadataRef metadata = block.metadata_ref();
const StringRef value_bytes = block.value_bytes();
+ DORIS_CHECK_LE(metadata.size, std::numeric_limits<uint32_t>::max())
+ << "VariantBatchBuilder metadata exceeds the ColumnString uint32
byte limit";
DORIS_CHECK_EQ(offsets.front(), 0);
DORIS_CHECK_EQ(static_cast<size_t>(offsets.back()), value_bytes.size);
- if (_typed) {
+ const std::array<uint32_t, 2> metadata_offsets {0,
static_cast<uint32_t>(metadata.size)};
+ _append_prevalidated_encoded_data({.metadata_bytes = {metadata.data,
metadata.size},
+ .metadata_offsets = metadata_offsets,
+ .meta_ids = {},
+ .value_bytes = value_bytes,
+ .value_offsets = offsets});
+}
+
+void ColumnVariantV2::_append_prevalidated_encoded_data(const EncodedDataView&
data) {
+ DCHECK(!data.metadata_offsets.empty());
+ DCHECK(!data.value_offsets.empty());
+ const size_t metadata_count = data.metadata_offsets.size() - 1;
+ const size_t rows = data.value_offsets.size() - 1;
+ if (rows == 0) {
+ return;
+ }
+
+ DCHECK_NE(metadata_count, 0);
+ DCHECK(data.meta_ids.empty() || data.meta_ids.size() == rows);
+ auto metadata_at = [&](uint32_t id) {
+ DCHECK_LT(id, metadata_count);
+ const uint32_t begin = data.metadata_offsets[id];
+ const uint32_t end = data.metadata_offsets[id + 1];
+ return StringRef(data.metadata_bytes.data + begin, end - begin);
+ };
+
+ if (_typed || _shredded) {
ensure_encoded();
}
DORIS_CHECK(_typed_type == nullptr) << "encoded state cannot retain a
typed data type";
@@ -995,11 +1240,28 @@ void ColumnVariantV2::insert_encoded_batch(const
VariantBatchBuilder& block) {
require_exclusive(_values, "values");
auto& values = assert_cast<ColumnString&>(*_values);
auto& metadata_ids = assert_cast<MetaIdsColumn&>(*_meta_ids);
- reserve_rows(values, metadata_ids, value_bytes.size, rows);
+ reserve_rows(values, metadata_ids, data.value_bytes.size, rows);
- const uint32_t id = _find_or_insert_metadata({metadata.data,
metadata.size});
- values.insert_many_continuous_binary_data(value_bytes.data,
offsets.data(), rows);
- metadata_ids.insert_many_vals(id, rows);
+ if (data.meta_ids.empty()) {
+ DCHECK_EQ(metadata_count, 1);
+ const uint32_t id = _find_or_insert_metadata(metadata_at(0));
+ values.insert_many_continuous_binary_data(data.value_bytes.data,
data.value_offsets.data(),
+ rows);
+ metadata_ids.insert_many_vals(id, rows);
+ } else {
+ DorisVector<uint32_t> remap(metadata_count, UNMAPPED_METADATA_ID);
+ auto& destination_ids = metadata_ids.get_data();
+ for (size_t row = 0; row < rows; ++row) {
+ const uint32_t source_id = data.meta_ids[row];
+ DCHECK_LT(source_id, metadata_count);
+ if (remap[source_id] == UNMAPPED_METADATA_ID) {
+ remap[source_id] =
_find_or_insert_metadata(metadata_at(source_id));
+ }
+ destination_ids.push_back(remap[source_id]);
+ }
+ values.insert_many_continuous_binary_data(data.value_bytes.data,
data.value_offsets.data(),
+ rows);
+ }
DCHECK_EQ(_meta_ids->size(), _values->size());
_check_invariants();
@@ -1893,6 +2155,15 @@ uint32_t
ColumnVariantV2::_find_or_insert_metadata(StringRef metadata) {
return id;
}
}
+
+ return _append_metadata(metadata);
+}
+
+uint32_t ColumnVariantV2::_append_metadata(StringRef metadata) {
+ DORIS_CHECK(metadata.data != nullptr || metadata.size == 0)
+ << "metadata bytes have a null pointer";
+ const auto& current_metadatas =
+ assert_cast<const ColumnString&>(*static_cast<const
IColumn::Ptr&>(_metadatas));
if (current_metadatas.size() == std::numeric_limits<uint32_t>::max()) {
throw Exception(ErrorCode::INVALID_ARGUMENT,
"ColumnVariantV2 metadata dictionary exceeds the
uint32 id limit");
diff --git a/be/src/core/column/variant_v2/column_variant_v2.h
b/be/src/core/column/variant_v2/column_variant_v2.h
index 7cd167663be..affea6a7e34 100644
--- a/be/src/core/column/variant_v2/column_variant_v2.h
+++ b/be/src/core/column/variant_v2/column_variant_v2.h
@@ -114,6 +114,34 @@ public:
std::span<const uint32_t> value_offsets;
};
+ // Scoped adapter for repeated encoded-row appends. It builds one hash
index over destination
+ // metadata ids and reuses it across calls, while the destination column
continues to own all
+ // copied bytes. Input rows must not borrow metadata or value bytes from
the destination,
+ // including its shredded materialization; use insert_range_from() for
that case. Do not mutate
+ // or destroy the destination while this adapter is alive.
+ class EncodedRowsAppender {
+ public:
+ ~EncodedRowsAppender();
+
+ EncodedRowsAppender(const EncodedRowsAppender&) = delete;
+ EncodedRowsAppender& operator=(const EncodedRowsAppender&) = delete;
+ EncodedRowsAppender(EncodedRowsAppender&&) noexcept;
+ EncodedRowsAppender& operator=(EncodedRowsAppender&&) = delete;
+
+ void append(std::span<const VariantRef> rows);
+
+#ifdef BE_TEST
+ size_t metadata_comparisons_for_test() const noexcept;
+#endif
+
+ private:
+ friend class ColumnVariantV2;
+ explicit EncodedRowsAppender(ColumnVariantV2& column);
+
+ struct Impl;
+ std::unique_ptr<Impl> _impl;
+ };
+
// Borrowed immutable adapter for whole-column E/T readers. The source
column owns every
// referenced column, type, and byte; any structural mutation invalidates
this view. Encoded
// bytes have already been validated at their insertion or deserialization
boundary.
@@ -183,6 +211,10 @@ public:
// metadata blob. Input buffers must not alias this column; use
insert_range_from for that case.
void insert_encoded_rows(const EncodedDataView& data);
+ // Reuse one appender when encoded rows arrive in bounded batches so
destination metadata is
+ // indexed once for the full import rather than scanned again for every
batch.
+ EncodedRowsAppender create_encoded_rows_appender();
+
// Direct trusted codec adapter. VariantBatchBuilder already produces
canonical metadata,
// validated values, and ColumnString-compatible uint32 offsets, so this
path copies its buffers
// without validating the encoded tree a second time.
@@ -249,6 +281,8 @@ private:
ColumnVariantV2();
ColumnVariantV2(const ColumnVariantV2& other);
+ void _append_prevalidated_encoded_data(const EncodedDataView& data);
+ uint32_t _append_metadata(StringRef metadata);
uint32_t _find_or_insert_metadata(StringRef metadata);
void _replace_shredded_state_with(const ColumnVariantV2& replacement);
void _ensure_serialized();
diff --git a/be/src/format_v2/parquet/parquet_profile.cpp
b/be/src/format_v2/parquet/parquet_profile.cpp
index eec657e741f..ce3a973bb02 100644
--- a/be/src/format_v2/parquet/parquet_profile.cpp
+++ b/be/src/format_v2/parquet/parquet_profile.cpp
@@ -111,6 +111,12 @@ void ParquetProfile::init(RuntimeProfile* profile) {
TUnit::TIME_NS,
parquet_profile);
variant_reconstructed_rows = add_persistent_counter(profile,
"VariantReconstructedRows",
TUnit::UNIT,
parquet_profile);
+ variant_unshredded_direct_import_time = add_persistent_counter(
+ profile, "VariantUnshreddedDirectImportTime", TUnit::TIME_NS,
parquet_profile);
+ variant_unshredded_direct_import_rows = add_persistent_counter(
+ profile, "VariantUnshreddedDirectImportRows", TUnit::UNIT,
parquet_profile);
+ variant_unshredded_direct_import_bytes = add_persistent_counter(
+ profile, "VariantUnshreddedDirectImportBytes", TUnit::BYTES,
parquet_profile);
variant_direct_leaf_rows =
add_persistent_counter(profile, "VariantDirectLeafRows",
TUnit::UNIT, parquet_profile);
variant_direct_leaf_path_misses = add_persistent_counter(profile,
"VariantDirectLeafPathMisses",
@@ -349,6 +355,9 @@ ParquetColumnReaderProfile
ParquetProfile::column_reader_profile() const {
.materialization_time = materialization_time,
.variant_reconstruction_time = variant_reconstruction_time,
.variant_reconstructed_rows = variant_reconstructed_rows,
+ .variant_unshredded_direct_import_time =
variant_unshredded_direct_import_time,
+ .variant_unshredded_direct_import_rows =
variant_unshredded_direct_import_rows,
+ .variant_unshredded_direct_import_bytes =
variant_unshredded_direct_import_bytes,
.variant_direct_leaf_rows = variant_direct_leaf_rows,
.variant_direct_leaf_path_misses = variant_direct_leaf_path_misses,
.variant_direct_leaf_residual_fallbacks =
variant_direct_leaf_residual_fallbacks,
diff --git a/be/src/format_v2/parquet/parquet_profile.h
b/be/src/format_v2/parquet/parquet_profile.h
index 18d285cf358..ef74f312cd1 100644
--- a/be/src/format_v2/parquet/parquet_profile.h
+++ b/be/src/format_v2/parquet/parquet_profile.h
@@ -42,6 +42,11 @@ struct ParquetColumnReaderProfile {
RuntimeProfile::Counter* materialization_time = nullptr; // value
materialization time (ns)
std::shared_ptr<RuntimeProfile::Counter> variant_reconstruction_time;
std::shared_ptr<RuntimeProfile::Counter> variant_reconstructed_rows;
+ // Complete unshredded roots imported without rebuilding their Variant
value tree. Bytes count
+ // the per-row metadata and value references presented to the direct
importer.
+ std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_import_time;
+ std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_import_rows;
+ std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_import_bytes;
std::shared_ptr<RuntimeProfile::Counter> variant_direct_leaf_rows;
std::shared_ptr<RuntimeProfile::Counter> variant_direct_leaf_path_misses;
std::shared_ptr<RuntimeProfile::Counter>
variant_direct_leaf_residual_fallbacks;
@@ -181,6 +186,10 @@ struct ParquetProfile {
RuntimeProfile::Counter* materialization_time = nullptr;
std::shared_ptr<RuntimeProfile::Counter> variant_reconstruction_time;
std::shared_ptr<RuntimeProfile::Counter> variant_reconstructed_rows;
+ // These counters are persistent because direct import can happen after
scanner teardown.
+ std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_import_time;
+ std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_import_rows;
+ std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_import_bytes;
std::shared_ptr<RuntimeProfile::Counter> variant_direct_leaf_rows;
std::shared_ptr<RuntimeProfile::Counter> variant_direct_leaf_path_misses;
std::shared_ptr<RuntimeProfile::Counter>
variant_direct_leaf_residual_fallbacks;
diff --git a/be/src/format_v2/parquet/reader/variant_column_reader.cpp
b/be/src/format_v2/parquet/reader/variant_column_reader.cpp
index d2c6fef0ed5..829ebc7b369 100644
--- a/be/src/format_v2/parquet/reader/variant_column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/variant_column_reader.cpp
@@ -26,6 +26,7 @@
#include <mutex>
#include <optional>
#include <string_view>
+#include <utility>
#include <vector>
#include "common/exception.h"
@@ -38,6 +39,7 @@
#include "core/column/column_vector.h"
#include "core/column/variant_v2/column_variant_v2.h"
#include "core/column/variant_v2/column_variant_v2_typed_column.h"
+#include "core/custom_allocator.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_variant_v2.h"
#include "core/value/variant/variant_batch_builder.h"
@@ -52,6 +54,9 @@ struct Cell {
bool is_null = false;
};
+constexpr std::array<char, 1> VARIANT_NULL_VALUE {static_cast<char>(
+ static_cast<uint8_t>(VariantPrimitiveId::NULL_VALUE) <<
VARIANT_VALUE_HEADER_SHIFT)};
+
Cell cell_at(const IColumn& column, size_t row) {
if (row >= column.size()) {
throw Exception(ErrorCode::CORRUPTION, "Parquet Variant row {} exceeds
column size {}", row,
@@ -444,9 +449,74 @@ void encode_variant_range(const ParquetColumnSchema&
schema, const IColumn& wrap
}
}
-ColumnVariantV2::MutablePtr encode_variant_column(const ParquetColumnSchema&
schema,
- const IColumn& physical,
- bool require_metadata =
true) {
+std::optional<std::pair<size_t, size_t>> unshredded_child_indices(
+ const ParquetColumnSchema& schema) {
+ if (schema.children.size() != 2 || find_child(schema, "typed_value",
nullptr) != nullptr) {
+ return std::nullopt;
+ }
+ size_t metadata_index = 0;
+ size_t value_index = 0;
+ if (find_child(schema, "metadata", &metadata_index) == nullptr ||
+ find_child(schema, "value", &value_index) == nullptr) {
+ return std::nullopt;
+ }
+ return std::pair {metadata_index, value_index};
+}
+
+void import_unshredded_variant_range(const ParquetColumnSchema& schema, const
ColumnStruct& wrapper,
+ const ColumnNullable* outer_nullable,
size_t metadata_index,
+ size_t value_index, size_t begin, size_t
end,
+ DorisVector<VariantRef>& encoded_rows,
+ ColumnVariantV2::EncodedRowsAppender&
appender,
+ int64_t* imported_bytes) {
+ encoded_rows.clear();
+ if (encoded_rows.capacity() < end - begin) {
+ encoded_rows.reserve(end - begin);
+ }
+ if (imported_bytes != nullptr) {
+ *imported_bytes = 0;
+ }
+ auto count_imported_bytes = [&](size_t bytes) {
+ if (imported_bytes == nullptr) {
+ return;
+ }
+ DORIS_CHECK_LE(bytes,
+ static_cast<size_t>(std::numeric_limits<int64_t>::max()
- *imported_bytes));
+ *imported_bytes += static_cast<int64_t>(bytes);
+ };
+ for (size_t row = begin; row < end; ++row) {
+ if (outer_nullable != nullptr &&
outer_nullable->get_null_map_data()[row] != 0) {
+ encoded_rows.push_back(
+ {.metadata = {.data = VARIANT_EMPTY_METADATA.data(),
+ .size = VARIANT_EMPTY_METADATA.size()},
+ .value = {VARIANT_NULL_VALUE.data(),
VARIANT_NULL_VALUE.size()}});
+ count_imported_bytes(VARIANT_EMPTY_METADATA.size());
+ count_imported_bytes(VARIANT_NULL_VALUE.size());
+ continue;
+ }
+
+ const Cell metadata = cell_at(wrapper.get_column(metadata_index), row);
+ if (metadata.is_null) {
+ throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} has
null metadata at row {}",
+ schema.name, row);
+ }
+ const StringRef metadata_bytes = metadata.column->get_data_at(row);
+ const Cell value = cell_at(wrapper.get_column(value_index), row);
+ const StringRef value_bytes =
+ value.is_null ? StringRef(VARIANT_NULL_VALUE.data(),
VARIANT_NULL_VALUE.size())
+ : value.column->get_data_at(row);
+ encoded_rows.push_back(
+ {.metadata = {.data = metadata_bytes.data, .size =
metadata_bytes.size},
+ .value = value_bytes});
+ count_imported_bytes(metadata_bytes.size);
+ count_imported_bytes(value_bytes.size);
+ }
+ appender.append(std::span<const VariantRef>(encoded_rows));
+}
+
+ColumnVariantV2::MutablePtr encode_variant_column(
+ const ParquetColumnSchema& schema, const IColumn& physical, bool
require_metadata = true,
+ const ParquetColumnReaderProfile* profile = nullptr) {
if (schema.kind != ParquetColumnSchemaKind::VARIANT) {
throw Exception(ErrorCode::INVALID_ARGUMENT, "Parquet column {} is not
Variant",
schema.name);
@@ -461,11 +531,50 @@ ColumnVariantV2::MutablePtr encode_variant_column(const
ParquetColumnSchema& sch
}
auto variants = ColumnVariantV2::create();
- constexpr size_t MAX_RECONSTRUCTION_BATCH_ROWS = 4096;
- for (size_t begin = 0; begin < physical.size(); begin +=
MAX_RECONSTRUCTION_BATCH_ROWS) {
- encode_variant_range(schema, wrapper, outer_nullable, begin,
- std::min(physical.size(), begin +
MAX_RECONSTRUCTION_BATCH_ROWS),
- require_metadata, *variants);
+ // A complete unshredded root already contains exact Variant Encoding V1
bytes. Keep the
+ // existing lazy materialization boundary, but validate and copy those
bytes without rebuilding
+ // the value tree through VariantBatchBuilder.
+ const auto unshredded_indices =
+ require_metadata ? unshredded_child_indices(schema) : std::nullopt;
+ constexpr size_t MAX_MATERIALIZATION_BATCH_ROWS = 4096;
+ DorisVector<VariantRef> encoded_rows;
+ std::optional<ColumnVariantV2::EncodedRowsAppender> encoded_rows_appender;
+ if (unshredded_indices.has_value()) {
+ encoded_rows.reserve(std::min(physical.size(),
MAX_MATERIALIZATION_BATCH_ROWS));
+
encoded_rows_appender.emplace(variants->create_encoded_rows_appender());
+ }
+ for (size_t begin = 0; begin < physical.size(); begin +=
MAX_MATERIALIZATION_BATCH_ROWS) {
+ const size_t end = std::min(physical.size(), begin +
MAX_MATERIALIZATION_BATCH_ROWS);
+ if (unshredded_indices.has_value()) {
+ DORIS_CHECK(encoded_rows_appender.has_value());
+ int64_t imported_bytes = 0;
+ RuntimeProfile::Counter* direct_import_time =
+ profile == nullptr ? nullptr
+ :
profile->variant_unshredded_direct_import_time.get();
+ {
+ // Update the shared timer on every exit, including validation
failures. Rows and
+ // bytes are published only after the complete chunk has been
appended.
+ SCOPED_TIMER(direct_import_time);
+ import_unshredded_variant_range(
+ schema, structure, outer_nullable,
unshredded_indices->first,
+ unshredded_indices->second, begin, end, encoded_rows,
+ *encoded_rows_appender, profile == nullptr ? nullptr :
&imported_bytes);
+ }
+ if (profile != nullptr) {
+ const auto imported_rows = static_cast<int64_t>(end - begin);
+ if (profile->variant_unshredded_direct_import_rows != nullptr)
{
+
COUNTER_UPDATE(profile->variant_unshredded_direct_import_rows.get(),
+ imported_rows);
+ }
+ if (profile->variant_unshredded_direct_import_bytes !=
nullptr) {
+
COUNTER_UPDATE(profile->variant_unshredded_direct_import_bytes.get(),
+ imported_bytes);
+ }
+ }
+ } else {
+ encode_variant_range(schema, wrapper, outer_nullable, begin, end,
require_metadata,
+ *variants);
+ }
}
return variants;
}
@@ -887,7 +996,7 @@ public:
}
if (!_materialized) {
SCOPED_TIMER(_profile.variant_reconstruction_time.get());
- _materialized = encode_variant_column(*_schema, *_physical);
+ _materialized = encode_variant_column(*_schema, *_physical, true,
&_profile);
update_counter(_profile.variant_reconstructed_rows,
static_cast<int64_t>(_physical->size()));
}
diff --git a/be/test/core/column/column_variant_v2_test.cpp
b/be/test/core/column/column_variant_v2_test.cpp
index 6e2683c39f7..1abb13f67c1 100644
--- a/be/test/core/column/column_variant_v2_test.cpp
+++ b/be/test/core/column/column_variant_v2_test.cpp
@@ -99,6 +99,18 @@ std::string empty_metadata_bytes() {
return metadata;
}
+std::string single_key_metadata_bytes(std::string_view key) {
+ EXPECT_LE(key.size(), std::numeric_limits<uint8_t>::max());
+ std::string metadata;
+ metadata.push_back(
+ static_cast<char>(VARIANT_ENCODING_VERSION |
VARIANT_METADATA_SORTED_STRINGS_MASK));
+ metadata.push_back(1);
+ metadata.push_back(0);
+ metadata.push_back(static_cast<char>(key.size()));
+ metadata.append(key);
+ return metadata;
+}
+
VariantField encoded_field(std::string metadata, std::string value) {
std::string field;
append_unsigned(field, metadata.size(), sizeof(uint32_t));
@@ -272,6 +284,51 @@ private:
ColumnVariantV2::MutablePtr _serialized;
};
+class MaterializableShreddedState final : public VariantShreddedState {
+public:
+ explicit MaterializableShreddedState(ColumnPtr materialized)
+ : _materialized(std::move(materialized)) {}
+
+ size_t size() const override { return _materialized->size(); }
+ size_t byte_size() const override { return _materialized->byte_size(); }
+ size_t allocated_bytes() const override { return
_materialized->allocated_bytes(); }
+ void sanity_check() const override { _materialized->sanity_check(); }
+ void for_each_subcolumn(const IColumn::ImutableColumnCallback& callback)
const override {
+ callback(*_materialized);
+ }
+ std::shared_ptr<VariantShreddedState> filter(const IColumn::Filter& filter,
+ ssize_t result_size_hint)
const override {
+ return std::make_shared<MaterializableShreddedState>(
+ _materialized->filter(filter, result_size_hint));
+ }
+ std::shared_ptr<VariantShreddedState> select_range(size_t start, size_t
length) const override {
+ return
std::make_shared<MaterializableShreddedState>(_materialized->cut(start,
length));
+ }
+ std::shared_ptr<VariantShreddedState> select_indices(
+ const uint32_t* indices_begin, const uint32_t* indices_end) const
override {
+ MutableColumnPtr selected = _materialized->clone_empty();
+ selected->insert_indices_from(*_materialized, indices_begin,
indices_end);
+ return
std::make_shared<MaterializableShreddedState>(std::move(selected));
+ }
+ bool can_materialize() const override { return true; }
+ bool try_append(const VariantShreddedState&) override { return false; }
+ std::optional<VariantShreddedTypedValue> find_typed_value(
+ std::span<const VariantShreddedPathSegment>) const override {
+ return std::nullopt;
+ }
+ std::optional<ColumnPtr> find_normalized_value(
+ std::span<const VariantShreddedPathSegment>) const override {
+ return std::nullopt;
+ }
+ const ColumnVariantV2& materialized_column() const override {
+ return assert_cast<const ColumnVariantV2&>(*_materialized);
+ }
+ const ColumnVariantV2& serialized_column() const override { return
materialized_column(); }
+
+private:
+ ColumnPtr _materialized;
+};
+
OwnedEncodedData late_invalid_encoded_rows() {
const VariantField first = encode_json("1");
const VariantField second = encode_json("2");
@@ -1010,6 +1067,108 @@ TEST(ColumnVariantV2Test,
InvalidEncodedRowsDoNotChangeTypedState) {
}
}
+TEST(ColumnVariantV2Test, InvalidEncodedRowsAppenderDoesNotChangeTypedState) {
+ constexpr std::array<int32_t, 1> TYPED_VALUES {7};
+ constexpr std::array<uint8_t, 1> TYPED_NULLS {0};
+ const VariantField first = encode_json("1");
+ const VariantField second = encode_json("2");
+ std::string invalid_value(second.ref().value.data,
second.ref().value.size);
+ invalid_value.push_back('\0');
+ const std::array<VariantRef, 2> invalid_rows {
+ first.ref(), VariantRef {.metadata = second.ref().metadata,
+ .value = {invalid_value.data(),
invalid_value.size()}}};
+
+ auto typed = typed_int32(TYPED_VALUES, TYPED_NULLS);
+ const IColumn* typed_storage = subcolumns(*typed).front().get();
+ auto appender = typed->create_encoded_rows_appender();
+ EXPECT_THROW(appender.append(invalid_rows), Exception);
+ ASSERT_TRUE(typed->is_typed());
+ EXPECT_EQ(subcolumns(*typed).front().get(), typed_storage);
+ expect_int32_rows(*typed, TYPED_VALUES, TYPED_NULLS);
+ typed->sanity_check();
+}
+
+TEST(ColumnVariantV2Test, EncodedRowsAppenderRejectsSameEncodedColumnBorrow) {
+ auto column = ColumnVariantV2::create();
+ insert_encoded_field(*column,
encode_json(R"({"old":"value-that-must-remain-owned"})"));
+ const std::string expected = json_at(*column, 0);
+ const VariantRef borrowed = column->get_value_ref(0);
+ // Repeating the borrowed value makes the requested append exceed the
existing value capacity,
+ // covering the reserve-induced invalidation that this API must reject
before mutation.
+ const std::vector<VariantRef> borrowed_rows(4096, borrowed);
+
+ auto appender = column->create_encoded_rows_appender();
+ try {
+ appender.append(borrowed_rows);
+ FAIL() << "expected same-column encoded rows to be rejected";
+ } catch (const Exception& exception) {
+ EXPECT_EQ(exception.code(), ErrorCode::INVALID_ARGUMENT);
+ EXPECT_NE(exception.message().find("must not alias"),
std::string::npos)
+ << exception.message();
+ }
+
+ ASSERT_EQ(column->size(), 1);
+ EXPECT_EQ(json_at(*column, 0), expected);
+ column->sanity_check();
+}
+
+TEST(ColumnVariantV2Test, EncodedRowsAppenderRejectsSameShreddedColumnBorrow) {
+ auto materialized = ColumnVariantV2::create();
+ insert_encoded_field(*materialized, encode_json(R"({"old":7})"));
+ auto column = ColumnVariantV2::create_shredded(
+
std::make_shared<MaterializableShreddedState>(std::move(materialized)));
+ const VariantRef borrowed = column->get_value_ref(0);
+
+ auto appender = column->create_encoded_rows_appender();
+ try {
+ appender.append(std::span<const VariantRef>(&borrowed, 1));
+ FAIL() << "expected same-column shredded rows to be rejected";
+ } catch (const Exception& exception) {
+ EXPECT_EQ(exception.code(), ErrorCode::INVALID_ARGUMENT);
+ EXPECT_NE(exception.message().find("must not alias"),
std::string::npos)
+ << exception.message();
+ }
+
+ EXPECT_TRUE(column->is_shredded());
+ ASSERT_EQ(column->size(), 1);
+ EXPECT_EQ(json_at(*column, 0), R"({"old":7})");
+ column->sanity_check();
+}
+
+TEST(ColumnVariantV2Test,
EncodedRowsAppenderInternsHighCardinalityMetadataAcrossChunks) {
+ constexpr size_t ROWS = 4097;
+ std::vector<std::string> metadata_storage;
+ metadata_storage.reserve(ROWS);
+ for (size_t row = 0; row < ROWS; ++row) {
+ metadata_storage.push_back(single_key_metadata_bytes("unused-" +
std::to_string(row)));
+ }
+ const std::string value = integer_value_bytes(7, 1);
+ std::vector<VariantRef> encoded_rows;
+ encoded_rows.reserve(ROWS);
+ for (const std::string& metadata : metadata_storage) {
+ encoded_rows.push_back({.metadata = {.data = metadata.data(), .size =
metadata.size()},
+ .value = {value.data(), value.size()}});
+ }
+
+ auto column = ColumnVariantV2::create();
+ auto appender = column->create_encoded_rows_appender();
+ appender.append(std::span<const VariantRef>(encoded_rows.data(), 4096));
+ appender.append(std::span<const VariantRef>(encoded_rows.data() + 4096,
1));
+
+ ASSERT_EQ(column->size(), ROWS);
+ ASSERT_EQ(metadata_count(*column), ROWS);
+ // A hash collision may require a byte comparison, but lookups must not
scan the growing
+ // destination dictionary as the old O(U^2) implementation did.
+ EXPECT_LT(appender.metadata_comparisons_for_test(), ROWS * 4);
+ for (const size_t row : {size_t {0}, size_t {4095}, size_t {4096}}) {
+ const VariantRef imported = column->get_value_ref(row);
+ EXPECT_EQ(as_view({imported.metadata.data, imported.metadata.size}),
metadata_storage[row]);
+ EXPECT_EQ(imported.value, StringRef(value));
+ EXPECT_EQ(imported.get_int(), 7);
+ }
+ column->sanity_check();
+}
+
TEST(ColumnVariantV2Test, InvalidEncodedRowsDoNotPartiallyAppendEncodedState) {
const OwnedEncodedData invalid = late_invalid_encoded_rows();
for (const bool omit_single_metadata_ids : {false, true}) {
@@ -1061,6 +1220,21 @@ TEST(ColumnVariantV2Test,
EmptyEncodedAppendsPreserveTypedState) {
ASSERT_TRUE(batch_destination->is_typed());
EXPECT_EQ(subcolumns(*batch_destination).front().get(), batch_storage);
expect_int32_rows(*batch_destination, VALUES, NULLS);
+
+ auto borrowed_destination = typed_int32(VALUES, NULLS);
+ const IColumn* borrowed_storage =
subcolumns(*borrowed_destination).front().get();
+ auto borrowed_appender =
borrowed_destination->create_encoded_rows_appender();
+ borrowed_appender.append(std::span<const VariantRef> {});
+ ASSERT_TRUE(borrowed_destination->is_typed());
+ EXPECT_EQ(subcolumns(*borrowed_destination).front().get(),
borrowed_storage);
+ expect_int32_rows(*borrowed_destination, VALUES, NULLS);
+
+ auto size_calls = std::make_shared<size_t>(0);
+ auto shredded_destination = ColumnVariantV2::create_shredded(
+ std::make_shared<CountingShreddedState>(1, size_calls));
+ auto shredded_appender =
shredded_destination->create_encoded_rows_appender();
+ shredded_appender.append(std::span<const VariantRef> {});
+ EXPECT_TRUE(shredded_destination->is_shredded());
}
TEST(ColumnVariantV2Test, ReadViewBorrowsValidatedEncodedState) {
diff --git a/be/test/format_v2/parquet/variant_column_reader_test.cpp
b/be/test/format_v2/parquet/variant_column_reader_test.cpp
index 74c79d515c8..3581b237436 100644
--- a/be/test/format_v2/parquet/variant_column_reader_test.cpp
+++ b/be/test/format_v2/parquet/variant_column_reader_test.cpp
@@ -24,6 +24,7 @@
#include <functional>
#include <initializer_list>
#include <limits>
+#include <string>
#include <string_view>
#include <vector>
@@ -68,6 +69,18 @@ MutableColumnPtr nullable_strings(const
std::vector<StringRef>& values,
return ColumnNullable::create(std::move(data), std::move(null_map));
}
+std::string single_key_metadata_bytes(std::string_view key) {
+ EXPECT_LE(key.size(), std::numeric_limits<uint8_t>::max());
+ std::string metadata;
+ metadata.push_back(
+ static_cast<char>(VARIANT_ENCODING_VERSION |
VARIANT_METADATA_SORTED_STRINGS_MASK));
+ metadata.push_back(1);
+ metadata.push_back(0);
+ metadata.push_back(static_cast<char>(key.size()));
+ metadata.append(key);
+ return metadata;
+}
+
ParquetColumnSchema unshredded_schema() {
ParquetColumnSchema schema;
schema.name = "payload";
@@ -572,6 +585,192 @@ TEST(VariantColumnReaderTest,
UnshreddedRowsPreserveSqlNullAndVariantNull) {
EXPECT_TRUE(variants.get_value_ref(2).is_null());
}
+TEST(VariantColumnReaderTest, UnshreddedDirectImportPreservesEncodedBytes) {
+ VariantBatchBuilder metadata_builder;
+ auto metadata_row = metadata_builder.begin_row();
+ auto metadata_object = metadata_row.start_object();
+ metadata_object.add_key(StringRef("unused"));
+ metadata_row.add_int(1);
+ metadata_object.finish();
+ metadata_row.finish();
+ VariantBatchBuilder metadata_batch = metadata_builder.finish_batch();
+ const VariantRef metadata_source = metadata_batch.value_at(0);
+
+ const std::array<char, 2> int_seven {
+ static_cast<char>(static_cast<uint8_t>(VariantPrimitiveId::INT8)
+ << VARIANT_VALUE_HEADER_SHIFT),
+ 7};
+ const StringRef encoded_value(int_seven.data(), int_seven.size());
+ MutableColumns fields;
+ fields.push_back(nullable_strings(
+ {{metadata_source.metadata.data, metadata_source.metadata.size}},
{0}));
+ fields.push_back(nullable_strings({encoded_value}, {0}));
+ auto physical = root_wrapper(std::move(fields));
+
+ RuntimeProfile runtime_profile("unshredded-direct-import");
+ ParquetProfile parquet_profile;
+ parquet_profile.init(&runtime_profile);
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ const Status status = materialize_variant_rows(unshredded_schema(),
*physical, output,
+
parquet_profile.column_reader_profile());
+ ASSERT_TRUE(status.ok()) << status;
+ const auto& variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(*output).get_nested_column());
+ ASSERT_TRUE(variants.is_shredded());
+ auto* reconstructed_rows =
runtime_profile.get_counter("VariantReconstructedRows");
+ auto* reconstruction_time =
runtime_profile.get_counter("VariantReconstructionTime");
+ auto* direct_import_time =
runtime_profile.get_counter("VariantUnshreddedDirectImportTime");
+ auto* direct_import_rows =
runtime_profile.get_counter("VariantUnshreddedDirectImportRows");
+ auto* direct_import_bytes =
runtime_profile.get_counter("VariantUnshreddedDirectImportBytes");
+ ASSERT_NE(reconstructed_rows, nullptr);
+ ASSERT_NE(reconstruction_time, nullptr);
+ ASSERT_NE(direct_import_time, nullptr);
+ ASSERT_NE(direct_import_rows, nullptr);
+ ASSERT_NE(direct_import_bytes, nullptr);
+ EXPECT_EQ(direct_import_rows->value(), 0);
+ EXPECT_EQ(direct_import_bytes->value(), 0);
+
+ const VariantRef imported = variants.get_value_ref(0);
+ EXPECT_EQ(StringRef(imported.metadata.data, imported.metadata.size),
+ StringRef(metadata_source.metadata.data,
metadata_source.metadata.size));
+ EXPECT_EQ(imported.value, encoded_value);
+ EXPECT_EQ(imported.get_int(), 7);
+ EXPECT_EQ(reconstructed_rows->value(), 1);
+ EXPECT_EQ(direct_import_rows->value(), 1);
+ EXPECT_EQ(direct_import_bytes->value(),
+ static_cast<int64_t>(metadata_source.metadata.size +
encoded_value.size));
+ EXPECT_GT(direct_import_time->value(), 0);
+ EXPECT_GE(reconstruction_time->value(), direct_import_time->value());
+}
+
+TEST(VariantColumnReaderTest,
UnshreddedDirectImportCrossesMaterializationBatchBoundary) {
+ constexpr size_t ROWS = 4097;
+ const std::array<char, 2> int_seven {
+ static_cast<char>(static_cast<uint8_t>(VariantPrimitiveId::INT8)
+ << VARIANT_VALUE_HEADER_SHIFT),
+ 7};
+ const StringRef encoded_value(int_seven.data(), int_seven.size());
+ std::vector<std::string> metadata_storage;
+ std::vector<StringRef> metadata_rows;
+ metadata_storage.reserve(ROWS);
+ metadata_rows.reserve(ROWS);
+ size_t metadata_bytes = 0;
+ for (size_t row = 0; row < ROWS; ++row) {
+ metadata_storage.push_back(single_key_metadata_bytes("unused-" +
std::to_string(row)));
+ metadata_bytes += metadata_storage.back().size();
+ metadata_rows.emplace_back(metadata_storage.back());
+ }
+ MutableColumns fields;
+ fields.push_back(nullable_strings(metadata_rows,
std::vector<uint8_t>(ROWS, 0)));
+ fields.push_back(nullable_strings(std::vector<StringRef>(ROWS,
encoded_value),
+ std::vector<uint8_t>(ROWS, 0)));
+ auto physical = root_wrapper(std::move(fields), NullMap(ROWS, 0));
+
+ RuntimeProfile runtime_profile("unshredded-direct-import-batches");
+ ParquetProfile parquet_profile;
+ parquet_profile.init(&runtime_profile);
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ const Status status = materialize_variant_rows(unshredded_schema(),
*physical, output,
+
parquet_profile.column_reader_profile());
+ ASSERT_TRUE(status.ok()) << status;
+ const auto& variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(*output).get_nested_column());
+ ASSERT_EQ(variants.size(), ROWS);
+ EXPECT_EQ(variants.get_value_ref(0).get_int(), 7);
+ EXPECT_EQ(variants.get_value_ref(4095).get_int(), 7);
+ EXPECT_EQ(variants.get_value_ref(4096).get_int(), 7);
+ EXPECT_EQ(StringRef(variants.get_value_ref(0).metadata.data,
+ variants.get_value_ref(0).metadata.size),
+ metadata_rows[0]);
+ EXPECT_EQ(StringRef(variants.get_value_ref(4096).metadata.data,
+ variants.get_value_ref(4096).metadata.size),
+ metadata_rows[4096]);
+ auto* direct_import_rows =
runtime_profile.get_counter("VariantUnshreddedDirectImportRows");
+ auto* direct_import_bytes =
runtime_profile.get_counter("VariantUnshreddedDirectImportBytes");
+ ASSERT_NE(direct_import_rows, nullptr);
+ ASSERT_NE(direct_import_bytes, nullptr);
+ EXPECT_EQ(direct_import_rows->value(), ROWS);
+ EXPECT_EQ(direct_import_bytes->value(),
+ static_cast<int64_t>(metadata_bytes + ROWS *
encoded_value.size));
+}
+
+TEST(VariantColumnReaderTest,
UnshreddedDirectImportProfilesImmediateFailureTime) {
+ const std::array<char, 1> invalid_value {static_cast<char>(0xff)};
+ const StringRef metadata(VARIANT_EMPTY_METADATA.data(),
VARIANT_EMPTY_METADATA.size());
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({{invalid_value.data(),
invalid_value.size()}}, {0}));
+ auto physical = root_wrapper(std::move(fields));
+
+ RuntimeProfile
runtime_profile("unshredded-direct-import-immediate-failure");
+ ParquetProfile parquet_profile;
+ parquet_profile.init(&runtime_profile);
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ const Status status = materialize_variant_rows(unshredded_schema(),
*physical, output,
+
parquet_profile.column_reader_profile());
+ ASSERT_TRUE(status.ok()) << status;
+ const auto& variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(*output).get_nested_column());
+ EXPECT_THROW((void)variants.get_value_ref(0), Exception);
+
+ auto* reconstructed_rows =
runtime_profile.get_counter("VariantReconstructedRows");
+ auto* direct_import_time =
runtime_profile.get_counter("VariantUnshreddedDirectImportTime");
+ auto* direct_import_rows =
runtime_profile.get_counter("VariantUnshreddedDirectImportRows");
+ auto* direct_import_bytes =
runtime_profile.get_counter("VariantUnshreddedDirectImportBytes");
+ ASSERT_NE(reconstructed_rows, nullptr);
+ ASSERT_NE(direct_import_time, nullptr);
+ ASSERT_NE(direct_import_rows, nullptr);
+ ASSERT_NE(direct_import_bytes, nullptr);
+ EXPECT_EQ(reconstructed_rows->value(), 0);
+ EXPECT_GT(direct_import_time->value(), 0);
+ EXPECT_EQ(direct_import_rows->value(), 0);
+ EXPECT_EQ(direct_import_bytes->value(), 0);
+}
+
+TEST(VariantColumnReaderTest,
UnshreddedDirectImportProfilesCompletedChunksBeforeFailure) {
+ constexpr size_t VALID_ROWS = 4096;
+ constexpr size_t ROWS = VALID_ROWS + 1;
+ const std::array<char, 2> int_seven {
+ static_cast<char>(static_cast<uint8_t>(VariantPrimitiveId::INT8)
+ << VARIANT_VALUE_HEADER_SHIFT),
+ 7};
+ const std::array<char, 1> invalid_value {static_cast<char>(0xff)};
+ const StringRef metadata(VARIANT_EMPTY_METADATA.data(),
VARIANT_EMPTY_METADATA.size());
+ const StringRef encoded_value(int_seven.data(), int_seven.size());
+ std::vector<StringRef> values(VALID_ROWS, encoded_value);
+ values.emplace_back(invalid_value.data(), invalid_value.size());
+ MutableColumns fields;
+ fields.push_back(nullable_strings(std::vector<StringRef>(ROWS, metadata),
+ std::vector<uint8_t>(ROWS, 0)));
+ fields.push_back(nullable_strings(values, std::vector<uint8_t>(ROWS, 0)));
+ auto physical = root_wrapper(std::move(fields), NullMap(ROWS, 0));
+
+ RuntimeProfile runtime_profile("unshredded-direct-import-late-failure");
+ ParquetProfile parquet_profile;
+ parquet_profile.init(&runtime_profile);
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ const Status status = materialize_variant_rows(unshredded_schema(),
*physical, output,
+
parquet_profile.column_reader_profile());
+ ASSERT_TRUE(status.ok()) << status;
+ const auto& variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(*output).get_nested_column());
+ EXPECT_THROW((void)variants.get_value_ref(0), Exception);
+
+ auto* reconstructed_rows =
runtime_profile.get_counter("VariantReconstructedRows");
+ auto* direct_import_time =
runtime_profile.get_counter("VariantUnshreddedDirectImportTime");
+ auto* direct_import_rows =
runtime_profile.get_counter("VariantUnshreddedDirectImportRows");
+ auto* direct_import_bytes =
runtime_profile.get_counter("VariantUnshreddedDirectImportBytes");
+ ASSERT_NE(reconstructed_rows, nullptr);
+ ASSERT_NE(direct_import_time, nullptr);
+ ASSERT_NE(direct_import_rows, nullptr);
+ ASSERT_NE(direct_import_bytes, nullptr);
+ EXPECT_EQ(reconstructed_rows->value(), 0);
+ EXPECT_GT(direct_import_time->value(), 0);
+ EXPECT_EQ(direct_import_rows->value(), VALID_ROWS);
+ EXPECT_EQ(direct_import_bytes->value(),
+ static_cast<int64_t>(VALID_ROWS * (metadata.size +
encoded_value.size)));
+}
+
TEST(VariantColumnReaderTest,
RequiredPhysicalGroupAppendsToNullableExternalSlot) {
const std::array<char, 2> int_seven {
static_cast<char>(static_cast<uint8_t>(VariantPrimitiveId::INT8)
@@ -1690,6 +1889,8 @@ TEST(VariantColumnReaderTest,
ShreddedStateOutlivesScannerProfile) {
reused_profile.init(runtime_profile.get());
EXPECT_EQ(parquet_profile.variant_reconstructed_rows,
reused_profile.variant_reconstructed_rows);
+ EXPECT_EQ(parquet_profile.variant_unshredded_direct_import_rows,
+ reused_profile.variant_unshredded_direct_import_rows);
auto visible_output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
ASSERT_TRUE(materialize_variant_rows(shredded_int64_schema(),
shredded_int64_physical({7}),
@@ -1699,6 +1900,7 @@ TEST(VariantColumnReaderTest,
ShreddedStateOutlivesScannerProfile) {
assert_cast<const
ColumnNullable&>(*visible_output).get_nested_column());
EXPECT_EQ(visible_variants.get_value_ref(0).get_int(), 7);
EXPECT_EQ(runtime_profile->get_counter("VariantReconstructedRows")->value(), 1);
+
EXPECT_EQ(runtime_profile->get_counter("VariantUnshreddedDirectImportRows")->value(),
0);
auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
ASSERT_TRUE(materialize_variant_rows(shredded_int64_schema(),
shredded_int64_physical({42}),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
