This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c19802899e3 [improvement](be) Import unshredded Parquet Variant values
without rebuilding them (#68959)
c19802899e3 is described below
commit c19802899e344513d38b83427222e1d29e51939e
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Tue Oct 13 16:21:35 2026 +0800
[improvement](be) Import unshredded Parquet Variant values without
rebuilding them (#68959)
### What problem does this PR solve?
Issue Number: None
Related PR: #66714, #66910, #67016
Problem Summary:
**In short.** This ports branch-4.1 #66714 to master. When the native
Parquet reader reads an unshredded Variant column, master decodes every
value and encodes it again through `VariantBatchBuilder`, although the
bytes already are Variant Encoding V1. With this PR the reader validates
those bytes and copies them into `ColumnVariantV2`. Because the copy
validates every metadata dictionary, a corrupt metadata key is reported
instead of being dropped.
**Background.** A Parquet Variant column without shredding stores two
binary fields per row: `metadata` (the key dictionary) and `value`.
`ColumnVariantV2` keeps rows the same way, as deduplicated metadata
blobs plus one value per row. master #67016, the port of branch-4.1
#66910, added a direct path seek for these columns. In branch-4.1,
#66910 builds on #66714, which never reached master, so #67016
re-encoded the values it found with `append_variant_rows` and left out
the two #66910 tests that need #66714.
**The problem, and what it cost.**
| Reading an unshredded Parquet Variant | master | with this PR |
|---|---|---|
| The whole value (`SELECT v`) | every row decoded and re-encoded
through `VariantBatchBuilder` (`encode_variant_range`) | metadata and
value validated, then copied; the profile reports
`VariantUnshreddedDirectImportTime`, `Rows` and `Bytes` |
| One path (`SELECT v['a']`), direct seek | each found value re-encoded
through `append_variant_rows`, which keeps one metadata dictionary per
batch and splits the batch in halves when rows use different
dictionaries | found values copied through the same importer |
| A row whose metadata has a corrupt key that the value does not use
(for example invalid UTF-8) | re-encoding drops the key, and the row
reads as if it were valid | the read fails with a corruption error, as
in branch-4.x |
**How this PR fixes it.**
- `ColumnVariantV2::EncodedRowsAppender`, as in branch-4.x: it validates
each distinct metadata dictionary once and every value, rejects input
that borrows bytes from the destination column (including its shredded
materialization), and then copies the rows, interning metadata through a
hash index that it keeps across batches. A failed append leaves a typed
or shredded destination unchanged. `insert_encoded_rows` and
`insert_encoded_batch` now share `_append_prevalidated_encoded_data`.
- `variant_column_reader.cpp`: when the column is materialized, complete
unshredded roots go through the appender in 4096-row batches, and
`normalize_unshredded_path` uses it instead of `append_variant_rows`,
which is removed. Three persistent profile counters report the time,
rows and bytes.
- The code follows master's `ColumnVariantV2`, which holds typed
`ColumnString` members where branch-4.x cast `IColumn`s. Hunks of the
same files that belong to other branch-4.x PRs (#66575, #66941) are not
included.
**Results.**
- BE UT: the 4 new `ColumnVariantV2Test` cases, the 4 new
`VariantColumnReaderTest` cases, the two #66910 corrupt-metadata cases
and two new assertions pass, together with the other Variant, JNI and
Parquet schema suites (755 tests).
- With the production change reverted, exactly the six new reader cases
fail; the two corrupt-metadata cases fail because the read throws
nothing.
Co-authored-by: daidai <[email protected]>
Co-authored-by: lihangyu <[email protected]>
---
.../core/column/variant_v2/column_variant_v2.cpp | 331 +++++++++++++++++++--
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 | 132 ++++++--
be/test/core/column/column_variant_v2_test.cpp | 174 +++++++++++
.../parquet/variant_column_reader_test.cpp | 302 +++++++++++++++++++
7 files changed, 927 insertions(+), 64 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 48ee53a8f9f..edab4bc6ecb 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>
@@ -130,6 +131,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) {
@@ -979,42 +994,239 @@ void ColumnVariantV2::insert_encoded_rows( //
NOLINT(readability-function-size)
}
}
- if (_typed || _shredded) {
- // Validate the incoming batch before materializing the destination so
malformed input
- // cannot mutate a typed or segmented Variant column as a side effect.
- 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 = *_values;
- auto& metadata_ids = *_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 = *static_cast<const
ColumnString::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 = *owner._metadatas;
+ const auto& values = *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 = *column._values;
+ auto& metadata_ids = *column._meta_ids;
+
+ auto current_metadatas = [&]() -> const ColumnString& {
+ return *static_cast<const ColumnString::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) {
@@ -1028,9 +1240,37 @@ 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);
+ 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) {
// Preserve the no-op behavior above: an empty encoded append must not
materialize the
// destination's lazy Variant representation.
@@ -1041,11 +1281,28 @@ void ColumnVariantV2::insert_encoded_batch(const
VariantBatchBuilder& block) {
require_exclusive(_values, "values");
auto& values = *_values;
auto& metadata_ids = *_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();
@@ -2180,6 +2437,14 @@ 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 = *static_cast<const
ColumnString::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 351b6b28a50..976a4a66e31 100644
--- a/be/src/core/column/variant_v2/column_variant_v2.h
+++ b/be/src/core/column/variant_v2/column_variant_v2.h
@@ -119,6 +119,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.
@@ -204,6 +232,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.
@@ -282,6 +314,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 8da28ac7d19..d4fff78e504 100644
--- a/be/src/format_v2/parquet/parquet_profile.cpp
+++ b/be/src/format_v2/parquet/parquet_profile.cpp
@@ -107,6 +107,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_unshredded_direct_seek_time = add_persistent_counter(
profile, "VariantUnshreddedDirectSeekTime", TUnit::TIME_NS,
parquet_profile);
variant_unshredded_direct_seek_rows = add_persistent_counter(
@@ -355,6 +361,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_unshredded_direct_seek_time =
variant_unshredded_direct_seek_time,
.variant_unshredded_direct_seek_rows =
variant_unshredded_direct_seek_rows,
.variant_unshredded_direct_seek_bytes =
variant_unshredded_direct_seek_bytes,
diff --git a/be/src/format_v2/parquet/parquet_profile.h
b/be/src/format_v2/parquet/parquet_profile.h
index 7ec112e0a55..4a0ef53e9bc 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;
// Pure metadata+value roots can seek a requested path before constructing
a root column.
std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_seek_time;
std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_seek_rows;
@@ -184,6 +189,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_unshredded_direct_seek_time;
std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_seek_rows;
std::shared_ptr<RuntimeProfile::Counter>
variant_unshredded_direct_seek_bytes;
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 cf8912735d5..7a7d926b1a5 100644
--- a/be/src/format_v2/parquet/reader/variant_column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/variant_column_reader.cpp
@@ -479,9 +479,60 @@ std::optional<std::pair<size_t, size_t>>
unshredded_child_indices(
return std::pair {metadata_index, value_index};
}
-ColumnVariantV2::MutablePtr encode_variant_column(const ParquetColumnSchema&
schema,
- const IColumn& physical,
- bool require_metadata =
true) {
+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);
@@ -496,11 +547,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;
}
@@ -1041,27 +1131,6 @@ UnshreddedPathScan scan_unshredded_path(const
ParquetColumnSchema& schema, const
return std::move(typed_values).finish(std::move(path_cache));
}
-void append_variant_rows(ColumnVariantV2& output, std::span<const VariantRef>
rows) {
- try {
- VariantBatchBuilder builder(VariantBatchBuilder::ReserveHint {.rows =
rows.size()});
- for (const VariantRef value : rows) {
- auto row = builder.begin_row();
- row.add_value(value);
- row.finish();
- }
- output.insert_encoded_batch(builder.finish_batch());
- } catch (...) {
- if (rows.size() <= 1) {
- throw;
- }
- // A builder has one metadata dictionary. Split heterogeneous batches
while preserving the
- // original validation error for a malformed single row.
- const size_t middle = rows.size() / 2;
- append_variant_rows(output, rows.first(middle));
- append_variant_rows(output, rows.subspan(middle));
- }
-}
-
ColumnPtr normalize_unshredded_path(const ParquetColumnSchema& schema, const
IColumn& physical,
std::pair<size_t, size_t> child_indices,
std::span<const
VariantShreddedPathSegment> path,
@@ -1074,6 +1143,7 @@ ColumnPtr normalize_unshredded_path(const
ParquetColumnSchema& schema, const ICo
DORIS_CHECK_EQ(structure.tuple_size(), schema.children.size());
auto values = ColumnVariantV2::create();
+ auto appender = values->create_encoded_rows_appender();
auto nulls = ColumnUInt8::create();
nulls->reserve(physical.size());
constexpr size_t MAX_DIRECT_SEEK_BATCH_ROWS = 4096;
@@ -1126,7 +1196,7 @@ ColumnPtr normalize_unshredded_path(const
ParquetColumnSchema& schema, const ICo
count_copied_bytes(null_value);
}
}
- append_variant_rows(*values, encoded_rows);
+ appender.append(std::span<const VariantRef>(encoded_rows));
}
*copied_bytes = total_copied_bytes;
return ColumnNullable::create(std::move(values), std::move(nulls));
@@ -1503,7 +1573,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 90eae92e41c..aa645e471ad 100644
--- a/be/test/core/column/column_variant_v2_test.cpp
+++ b/be/test/core/column/column_variant_v2_test.cpp
@@ -104,6 +104,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));
@@ -227,6 +239,51 @@ struct OwnedEncodedData {
}
};
+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(IColumn::ColumnCallback 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");
@@ -1067,6 +1124,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}) {
@@ -1118,6 +1277,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 a27c9132be6..bec546affb8 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>
@@ -70,6 +71,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;
+}
+
Status extract_object_key(const ColumnVariantV2& source, const ColumnNullable&
nullable,
StringRef key, ColumnPtr* result) {
const std::array segments {VariantElementV2PathSegment::object_key(key)};
@@ -963,6 +976,106 @@ TEST(VariantColumnReaderTest,
UnshreddedCachedMissingStillValidatesEveryObject)
EXPECT_FALSE(result);
}
+TEST(VariantColumnReaderTest,
UnshreddedIntegerLeafKeepsCorruptMetadataOnFallback) {
+ VariantBatchBuilder builder;
+ auto row = builder.begin_row();
+ auto root = row.start_object();
+ root.add_key(StringRef("time_us"));
+ row.add_int(7);
+ root.add_key(StringRef("zz_unused"));
+ row.add_int(1);
+ root.finish();
+ row.finish();
+ VariantBatchBuilder batch = builder.finish_batch();
+ const VariantRef encoded = batch.value_at(0);
+
+ std::string invalid_metadata(encoded.metadata.data, encoded.metadata.size);
+ const size_t invalid_key = invalid_metadata.find("zz_unused");
+ ASSERT_NE(invalid_key, std::string::npos);
+ invalid_metadata[invalid_key] = static_cast<char>(0xff);
+
+ MutableColumns fields;
+ fields.push_back(nullable_strings({StringRef(invalid_metadata)}, {0}));
+ fields.push_back(nullable_strings({encoded.value}, {0}));
+ auto physical = root_wrapper(std::move(fields));
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ ASSERT_TRUE(materialize_variant_rows(unshredded_schema(), *physical,
output).ok());
+ const auto& root_nullable = assert_cast<const ColumnNullable&>(*output);
+ const auto& root_variants =
+ assert_cast<const
ColumnVariantV2&>(root_nullable.get_nested_column());
+ const std::array segments
{VariantElementV2PathSegment::object_key(StringRef("time_us"))};
+ std::unique_ptr<ResolvedVariantElementV2Path> path;
+ ASSERT_TRUE(resolve_variant_element_v2_path(segments, &path).ok());
+ ColumnPtr result;
+ ASSERT_TRUE(extract_variant_element_v2(
+ root_variants, *path,
+ std::span<const
uint8_t>(root_nullable.get_null_map_data().data(),
+
root_nullable.get_null_map_data().size()),
+ &result)
+ .ok());
+
+ const auto& result_variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(*result).get_nested_column());
+ ASSERT_TRUE(result_variants.is_shredded());
+ EXPECT_THROW((void)result_variants.get_value_ref(0), Exception);
+}
+
+TEST(VariantColumnReaderTest, UnshreddedIntegerLeafValidatesLeadingJsonNull) {
+ VariantBatchBuilder invalid_builder;
+ auto invalid_row = invalid_builder.begin_row();
+ auto invalid_root = invalid_row.start_object();
+ invalid_root.add_key(StringRef("time_us"));
+ invalid_row.add_null();
+ invalid_root.add_key(StringRef("zz_unused"));
+ invalid_row.add_int(1);
+ invalid_root.finish();
+ invalid_row.finish();
+ VariantBatchBuilder invalid_batch = invalid_builder.finish_batch();
+ const VariantRef invalid = invalid_batch.value_at(0);
+
+ VariantBatchBuilder valid_builder;
+ auto valid_row = valid_builder.begin_row();
+ auto valid_root = valid_row.start_object();
+ valid_root.add_key(StringRef("time_us"));
+ valid_row.add_int(7);
+ valid_root.finish();
+ valid_row.finish();
+ VariantBatchBuilder valid_batch = valid_builder.finish_batch();
+ const VariantRef valid = valid_batch.value_at(0);
+
+ std::string invalid_metadata(invalid.metadata.data, invalid.metadata.size);
+ const size_t invalid_key = invalid_metadata.find("zz_unused");
+ ASSERT_NE(invalid_key, std::string::npos);
+ invalid_metadata[invalid_key] = static_cast<char>(0xff);
+
+ MutableColumns fields;
+ fields.push_back(nullable_strings(
+ {StringRef(invalid_metadata), StringRef(valid.metadata.data,
valid.metadata.size)},
+ {0, 0}));
+ fields.push_back(nullable_strings({invalid.value, valid.value}, {0, 0}));
+ auto physical = root_wrapper(std::move(fields), {0, 0});
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ ASSERT_TRUE(materialize_variant_rows(unshredded_schema(), *physical,
output).ok());
+ const auto& root_nullable = assert_cast<const ColumnNullable&>(*output);
+ const auto& root_variants =
+ assert_cast<const
ColumnVariantV2&>(root_nullable.get_nested_column());
+ const std::array segments
{VariantElementV2PathSegment::object_key(StringRef("time_us"))};
+ std::unique_ptr<ResolvedVariantElementV2Path> path;
+ ASSERT_TRUE(resolve_variant_element_v2_path(segments, &path).ok());
+ ColumnPtr result;
+ ASSERT_TRUE(extract_variant_element_v2(
+ root_variants, *path,
+ std::span<const
uint8_t>(root_nullable.get_null_map_data().data(),
+
root_nullable.get_null_map_data().size()),
+ &result)
+ .ok());
+
+ const auto& result_variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(*result).get_nested_column());
+ ASSERT_TRUE(result_variants.is_shredded());
+ EXPECT_THROW((void)result_variants.get_value_ref(0), Exception);
+}
+
TEST(VariantColumnReaderTest,
UnshreddedIntegerLeafReportsCorruptPayloadOnFallback) {
VariantBatchBuilder builder;
auto row = builder.begin_row();
@@ -1061,6 +1174,192 @@ TEST(VariantColumnReaderTest,
UnshreddedMixedScalarLeafFallsBackToSubtree) {
EXPECT_EQ(result_variants.get_value_ref(1).get_string(),
StringRef("seven"));
}
+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)
@@ -2259,6 +2558,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}),
@@ -2268,6 +2569,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]
