This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 b4b2f795576 [fix](variant) Preserve nested Variant append atomicity
(#66421)
b4b2f795576 is described below
commit b4b2f7955768fd53db62709745cbbdf059c0653e
Author: Gabriel <[email protected]>
AuthorDate: Tue Aug 4 19:44:54 2026 +0800
[fix](variant) Preserve nested Variant append atomicity (#66421)
### What problem does this PR solve?
Iceberg Variant reads need broader coverage across physical shredding
states, nested containers, delete files, projection pruning, predicate
pruning, and delayed materialization. Review also exposed a reader bug:
a corrupt lazily decoded Variant nested after an already appended
sibling could leave the destination column partially mutated.
### What is changed?
- Make compatible nested-column appends transactional for nullable,
struct, array, and map outputs so a failed lazy Variant fallback
restores every child and offset.
- Add a focused Parquet reader test for lazy corruption after partial
nested appends.
- Use explicit unshredded Spark fixtures and the stable shredded fixture
for physical-mode coverage.
- Strengthen Iceberg checks for multiple active scanners, PUFFIN
deletion-vector metadata and observable deleted rows, completion-aware
profile polling, merged-profile counter accounting, and Variant-specific
delayed materialization.
### Tests
- BE ASAN Variant tests: 559 passed, 5 existing conditional skips, 0
failed.
- Regression framework unit tests: 3 passed, 0 failed.
- Iceberg Variant regression suite: 1 suite passed, 0 failed.
- BE C++ ASAN build: passed.
- clang-format and diff validation: passed.
---
.../parquet/reader/variant_column_reader.cpp | 101 ++-
.../parquet/variant_column_reader_test.cpp | 894 +++++++++++++++++++++
.../iceberg/test_iceberg_variant_read.out | 59 +-
.../iceberg/test_iceberg_variant_read.groovy | 552 ++++++++++++-
4 files changed, 1555 insertions(+), 51 deletions(-)
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 26e98e5128b..43ff808cf5b 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 <vector>
#include "common/exception.h"
#include "core/assert_cast.h"
@@ -852,16 +853,30 @@ ColumnPtr transform_node(const
VariantMaterializationNode& plan, ColumnPtr physi
void append_compatible_column(IColumn& output, const IColumn& converted) {
if (auto* output_nullable = check_and_get_column<ColumnNullable>(output)) {
- if (const auto* converted_nullable =
check_and_get_column<ColumnNullable>(converted)) {
- append_compatible_column(output_nullable->get_nested_column(),
- converted_nullable->get_nested_column());
- output_nullable->get_null_map_column().insert_range_from(
- converted_nullable->get_null_map_column(), 0,
converted.size());
- } else {
- append_compatible_column(output_nullable->get_nested_column(),
converted);
- // External slots and nested Iceberg fields may remain nullable
even when one file's
- // physical node is required. Preserve that destination invariant
with non-null bits.
- output_nullable->push_false_to_nullmap(converted.size());
+ auto& nested = output_nullable->get_nested_column();
+ auto& null_map = output_nullable->get_null_map_column();
+ const size_t nested_size = nested.size();
+ const size_t null_map_size = null_map.size();
+ try {
+ if (const auto* converted_nullable =
check_and_get_column<ColumnNullable>(converted)) {
+ append_compatible_column(nested,
converted_nullable->get_nested_column());
+
null_map.insert_range_from(converted_nullable->get_null_map_column(), 0,
+ converted.size());
+ } else {
+ append_compatible_column(nested, converted);
+ // External slots and nested Iceberg fields may remain
nullable even when one
+ // file's physical node is required. Preserve that destination
invariant with
+ // non-null bits.
+ output_nullable->push_false_to_nullmap(converted.size());
+ }
+ } catch (...) {
+ if (nested.size() > nested_size) {
+ nested.pop_back(nested.size() - nested_size);
+ }
+ if (null_map.size() > null_map_size) {
+ null_map.pop_back(null_map.size() - null_map_size);
+ }
+ throw;
}
return;
}
@@ -885,8 +900,25 @@ void append_compatible_column(IColumn& output, const
IColumn& converted) {
throw Exception(ErrorCode::CORRUPTION,
"Parquet Variant materialization produced an
incompatible STRUCT");
}
+ std::vector<size_t> original_sizes(output_struct->tuple_size());
for (size_t i = 0; i < output_struct->tuple_size(); ++i) {
- append_compatible_column(output_struct->get_column(i),
converted_struct->get_column(i));
+ original_sizes[i] = output_struct->get_column(i).size();
+ }
+ try {
+ for (size_t i = 0; i < output_struct->tuple_size(); ++i) {
+ append_compatible_column(output_struct->get_column(i),
+ converted_struct->get_column(i));
+ }
+ } catch (...) {
+ // Variant corruption can surface only during lazy fallback after
earlier siblings
+ // were appended. Roll every child back to preserve the
failed-append invariant.
+ for (size_t i = 0; i < output_struct->tuple_size(); ++i) {
+ auto& child = output_struct->get_column(i);
+ if (child.size() > original_sizes[i]) {
+ child.pop_back(child.size() - original_sizes[i]);
+ }
+ }
+ throw;
}
return;
}
@@ -897,12 +929,22 @@ void append_compatible_column(IColumn& output, const
IColumn& converted) {
throw Exception(ErrorCode::CORRUPTION,
"Parquet Variant materialization produced an
incompatible ARRAY");
}
- const size_t element_base = output_array->get_data().size();
- append_compatible_column(output_array->get_data(),
converted_array->get_data());
+ auto& output_data = output_array->get_data();
auto& output_offsets = output_array->get_offsets();
- output_offsets.reserve(output_offsets.size() +
converted_array->size());
- for (const auto offset : converted_array->get_offsets()) {
- output_offsets.push_back(element_base + offset);
+ const size_t element_base = output_data.size();
+ const size_t offsets_size = output_offsets.size();
+ try {
+ append_compatible_column(output_data, converted_array->get_data());
+ output_offsets.reserve(output_offsets.size() +
converted_array->size());
+ for (const auto offset : converted_array->get_offsets()) {
+ output_offsets.push_back(element_base + offset);
+ }
+ } catch (...) {
+ if (output_data.size() > element_base) {
+ output_data.pop_back(output_data.size() - element_base);
+ }
+ output_offsets.resize(offsets_size);
+ throw;
}
return;
}
@@ -913,13 +955,28 @@ void append_compatible_column(IColumn& output, const
IColumn& converted) {
throw Exception(ErrorCode::CORRUPTION,
"Parquet Variant materialization produced an
incompatible MAP");
}
- const size_t element_base = output_map->get_keys().size();
- append_compatible_column(output_map->get_keys(),
converted_map->get_keys());
- append_compatible_column(output_map->get_values(),
converted_map->get_values());
+ auto& output_keys = output_map->get_keys();
+ auto& output_values = output_map->get_values();
auto& output_offsets = output_map->get_offsets();
- output_offsets.reserve(output_offsets.size() + converted_map->size());
- for (const auto offset : converted_map->get_offsets()) {
- output_offsets.push_back(element_base + offset);
+ const size_t element_base = output_keys.size();
+ const size_t values_size = output_values.size();
+ const size_t offsets_size = output_offsets.size();
+ try {
+ append_compatible_column(output_keys, converted_map->get_keys());
+ append_compatible_column(output_values,
converted_map->get_values());
+ output_offsets.reserve(output_offsets.size() +
converted_map->size());
+ for (const auto offset : converted_map->get_offsets()) {
+ output_offsets.push_back(element_base + offset);
+ }
+ } catch (...) {
+ if (output_keys.size() > element_base) {
+ output_keys.pop_back(output_keys.size() - element_base);
+ }
+ if (output_values.size() > values_size) {
+ output_values.pop_back(output_values.size() - values_size);
+ }
+ output_offsets.resize(offsets_size);
+ throw;
}
return;
}
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 40c18635aad..24ace262fab 100644
--- a/be/test/format_v2/parquet/variant_column_reader_test.cpp
+++ b/be/test/format_v2/parquet/variant_column_reader_test.cpp
@@ -20,20 +20,33 @@
#include <gtest/gtest.h>
#include <array>
+#include <cmath>
+#include <functional>
+#include <initializer_list>
+#include <limits>
#include <string_view>
#include <vector>
+#include "common/exception.h"
#include "core/assert_cast.h"
#include "core/column/column_array.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_map.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_struct.h"
#include "core/column/variant_v2/column_variant_v2.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_date_or_datetime_v2.h"
+#include "core/data_type/data_type_decimal.h"
+#include "core/data_type/data_type_map.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_struct.h"
+#include "core/data_type/data_type_timestamptz.h"
#include "core/data_type/data_type_variant_v2.h"
+#include "core/value/timestamptz_value.h"
#include "core/value/variant/variant_batch_builder.h"
#include "core/value/variant/variant_parquet_encoding.h"
#include "exprs/function/function_variant_element_v2.h"
@@ -90,6 +103,16 @@ ParquetColumnSchema shredded_int64_schema() {
return schema;
}
+ParquetColumnSchema shredded_primitive_schema(DataTypePtr type) {
+ auto schema = unshredded_schema();
+ auto typed = std::make_unique<ParquetColumnSchema>();
+ typed->name = "typed_value";
+ typed->kind = ParquetColumnSchemaKind::PRIMITIVE;
+ typed->type = make_nullable(std::move(type));
+ schema.children.push_back(std::move(typed));
+ return schema;
+}
+
ParquetColumnSchema shredded_object_schema() {
auto schema = unshredded_schema();
auto typed = std::make_unique<ParquetColumnSchema>();
@@ -110,6 +133,12 @@ ParquetColumnSchema shredded_object_schema() {
return schema;
}
+ParquetColumnSchema shredded_named_object_schema(std::string field_name) {
+ auto schema = shredded_object_schema();
+ schema.children.back()->children[0]->name = std::move(field_name);
+ return schema;
+}
+
ParquetColumnSchema shredded_binary_object_schema() {
auto schema = shredded_object_schema();
auto* leaf = schema.children.back()->children[0]->children[0].get();
@@ -136,6 +165,17 @@ ParquetColumnSchema shredded_array_schema() {
return schema;
}
+ParquetColumnSchema shredded_mixed_array_schema() {
+ auto schema = shredded_array_schema();
+ auto* element = schema.children.back()->children[0].get();
+ auto value = std::make_unique<ParquetColumnSchema>();
+ value->name = "value";
+ value->kind = ParquetColumnSchemaKind::PRIMITIVE;
+ value->type = make_nullable(std::make_shared<DataTypeString>());
+ element->children.insert(element->children.begin(), std::move(value));
+ return schema;
+}
+
MutableColumnPtr shredded_int64_physical(const std::vector<int64_t>& values) {
const std::array<char, 1> ignored {0};
const StringRef metadata(VARIANT_EMPTY_METADATA.data(),
VARIANT_EMPTY_METADATA.size());
@@ -157,6 +197,20 @@ MutableColumnPtr shredded_int64_physical(const
std::vector<int64_t>& values) {
return ColumnNullable::create(std::move(structure), std::move(root_nulls));
}
+MutableColumnPtr shredded_primitive_physical(MutableColumnPtr typed) {
+ const size_t rows = typed->size();
+ const std::array<char, 1> ignored {0};
+ const StringRef metadata(VARIANT_EMPTY_METADATA.data(),
VARIANT_EMPTY_METADATA.size());
+ MutableColumns fields;
+ fields.push_back(nullable_strings(std::vector<StringRef>(rows, metadata),
+ std::vector<uint8_t>(rows, 0)));
+ fields.push_back(nullable_strings(std::vector<StringRef>(rows,
{ignored.data(), 0}),
+ std::vector<uint8_t>(rows, 1)));
+ fields.push_back(std::move(typed));
+ return ColumnNullable::create(ColumnStruct::create(std::move(fields)),
+ ColumnUInt8::create(rows, 0));
+}
+
MutableColumnPtr projected_shredded_object_physical(const
std::vector<int64_t>& values,
const IColumn**
decoded_leaf = nullptr) {
auto integers = ColumnInt64::create();
@@ -182,6 +236,37 @@ MutableColumnPtr projected_shredded_object_physical(const
std::vector<int64_t>&
return ColumnNullable::create(std::move(root),
ColumnUInt8::create(values.size(), 0));
}
+MutableColumnPtr root_wrapper(MutableColumns fields, NullMap root_nulls = {0});
+MutableColumnPtr nullable_int64(const std::vector<int64_t>& values,
+ const std::vector<uint8_t>& nulls);
+
+MutableColumnPtr complete_shredded_object_physical(std::string_view
residual_key,
+ int64_t residual_value,
int64_t typed_value) {
+ VariantBatchBuilder builder;
+ auto row = builder.begin_row();
+ auto object = row.start_object();
+ object.add_key(StringRef(residual_key.data(), residual_key.size()));
+ row.add_int(residual_value);
+ object.finish();
+ row.finish();
+ VariantBatchBuilder batch = builder.finish_batch();
+ const VariantRef residual = batch.value_at(0);
+
+ MutableColumns wrapper_fields;
+ wrapper_fields.push_back(nullable_int64({typed_value}, {0}));
+ MutableColumns object_fields;
+
object_fields.push_back(ColumnNullable::create(ColumnStruct::create(std::move(wrapper_fields)),
+ ColumnUInt8::create(1, 0)));
+ MutableColumns root_fields;
+ root_fields.push_back(
+ nullable_strings({StringRef(residual.metadata.data,
residual.metadata.size)}, {0}));
+ root_fields.push_back(
+ nullable_strings({StringRef(residual.value.data,
residual.value.size)}, {0}));
+
root_fields.push_back(ColumnNullable::create(ColumnStruct::create(std::move(object_fields)),
+ ColumnUInt8::create(1, 0)));
+ return root_wrapper(std::move(root_fields));
+}
+
MutableColumnPtr projected_two_field_object_physical(const
std::vector<int64_t>& first,
const
std::vector<int64_t>& second) {
DORIS_CHECK(first.size() == second.size());
@@ -226,6 +311,58 @@ MutableColumnPtr projected_wide_object_physical(size_t
field_count, int64_t valu
ColumnUInt8::create(1, 0));
}
+std::string materialization_error(const ParquetColumnSchema& schema, ColumnPtr
physical) {
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ const Status status = materialize_variant_rows(schema,
std::move(physical), output);
+ if (!status.ok()) {
+ return status.to_string();
+ }
+ try {
+ const auto& variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const
ColumnNullable&>(*output).get_nested_column());
+ (void)variants.get_value_ref(0);
+ } catch (const Exception& exception) {
+ return exception.what();
+ }
+ return {};
+}
+
+MutableColumnPtr root_wrapper(MutableColumns fields, NullMap root_nulls) {
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().assign(root_nulls.begin(), root_nulls.end());
+ return ColumnNullable::create(ColumnStruct::create(std::move(fields)),
std::move(null_map));
+}
+
+MutableColumnPtr nullable_int64(const std::vector<int64_t>& values,
+ const std::vector<uint8_t>& nulls) {
+ auto data = ColumnInt64::create();
+ data->get_data().assign(values.begin(), values.end());
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().assign(nulls.begin(), nulls.end());
+ return ColumnNullable::create(std::move(data), std::move(null_map));
+}
+
+template <typename ColumnType, typename Value>
+MutableColumnPtr nullable_fixed(std::initializer_list<Value> values,
+ std::initializer_list<uint8_t> nulls) {
+ auto data = ColumnType::create();
+ for (const Value& value : values) {
+ data->insert_value(value);
+ }
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().assign(nulls.begin(), nulls.end());
+ return ColumnNullable::create(std::move(data), std::move(null_map));
+}
+
+template <typename ColumnType, typename Value>
+MutableColumnPtr nullable_decimal(uint32_t scale, std::initializer_list<Value>
values) {
+ auto data = ColumnType::create(0, scale);
+ for (const Value& value : values) {
+ data->insert_value(value);
+ }
+ return ColumnNullable::create(std::move(data),
ColumnUInt8::create(values.size(), 0));
+}
+
} // namespace
TEST(VariantColumnReaderTest, UnshreddedRowsPreserveSqlNullAndVariantNull) {
@@ -304,6 +441,177 @@ TEST(VariantColumnReaderTest,
ShreddedIntegerKeepsDeclaredPhysicalWidth) {
EXPECT_EQ(variants.get_value_ref(0).primitive_id(),
VariantPrimitiveId::INT64);
}
+TEST(VariantColumnReaderTest, ReconstructsShreddedPrimitiveTypeMatrix) {
+ auto decode = [](ParquetColumnSchema schema, MutableColumnPtr typed,
+ const std::function<void(const ColumnVariantV2&)>&
verify) {
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ const Status status = materialize_variant_rows(
+ schema, shredded_primitive_physical(std::move(typed)), output);
+ ASSERT_TRUE(status.ok()) << status;
+ verify(assert_cast<const ColumnVariantV2&>(
+ assert_cast<const
ColumnNullable&>(*output).get_nested_column()));
+ };
+
+ decode(shredded_primitive_schema(std::make_shared<DataTypeBool>()),
+ nullable_fixed<ColumnUInt8, UInt8>({0, 1}, {0, 0}), [](const auto&
values) {
+ EXPECT_EQ(values.get_value_ref(0).primitive_id(),
VariantPrimitiveId::FALSE_VALUE);
+ EXPECT_EQ(values.get_value_ref(1).primitive_id(),
VariantPrimitiveId::TRUE_VALUE);
+ });
+
+ auto verify_integer = [&](DataTypePtr type, MutableColumnPtr typed, int
width, int64_t first,
+ int64_t second) {
+ auto schema = shredded_primitive_schema(std::move(type));
+ schema.children.back()->type_descriptor.integer_bit_width = width;
+ decode(std::move(schema), std::move(typed), [&](const auto& values) {
+ EXPECT_EQ(values.get_value_ref(0).get_int(), first);
+ EXPECT_EQ(values.get_value_ref(1).get_int(), second);
+ });
+ };
+ verify_integer(
+ std::make_shared<DataTypeInt8>(),
+ nullable_fixed<ColumnInt8, Int8>(
+ {std::numeric_limits<Int8>::min(),
std::numeric_limits<Int8>::max()}, {0, 0}),
+ 8, std::numeric_limits<Int8>::min(),
std::numeric_limits<Int8>::max());
+ verify_integer(
+ std::make_shared<DataTypeInt16>(),
+ nullable_fixed<ColumnInt16, Int16>(
+ {std::numeric_limits<Int16>::min(),
std::numeric_limits<Int16>::max()}, {0, 0}),
+ 16, std::numeric_limits<Int16>::min(),
std::numeric_limits<Int16>::max());
+ verify_integer(
+ std::make_shared<DataTypeInt32>(),
+ nullable_fixed<ColumnInt32, Int32>(
+ {std::numeric_limits<Int32>::min(),
std::numeric_limits<Int32>::max()}, {0, 0}),
+ 32, std::numeric_limits<Int32>::min(),
std::numeric_limits<Int32>::max());
+ verify_integer(
+ std::make_shared<DataTypeInt64>(),
+ nullable_fixed<ColumnInt64, Int64>(
+ {std::numeric_limits<Int64>::min(),
std::numeric_limits<Int64>::max()}, {0, 0}),
+ 64, std::numeric_limits<Int64>::min(),
std::numeric_limits<Int64>::max());
+
+ decode(shredded_primitive_schema(std::make_shared<DataTypeFloat32>()),
+ nullable_fixed<ColumnFloat32,
Float32>({std::numeric_limits<Float32>::quiet_NaN(),
+
std::numeric_limits<Float32>::infinity()},
+ {0, 0}),
+ [](const auto& values) {
+ EXPECT_TRUE(std::isnan(values.get_value_ref(0).get_float()));
+ EXPECT_TRUE(std::isinf(values.get_value_ref(1).get_float()));
+ });
+ decode(shredded_primitive_schema(std::make_shared<DataTypeFloat64>()),
+ nullable_fixed<ColumnFloat64,
Float64>({-std::numeric_limits<Float64>::infinity(), 1.25},
+ {0, 0}),
+ [](const auto& values) {
+ EXPECT_EQ(values.get_value_ref(0).get_double(),
+ -std::numeric_limits<Float64>::infinity());
+ EXPECT_EQ(values.get_value_ref(1).get_double(), 1.25);
+ });
+
+ {
+ auto schema =
shredded_primitive_schema(std::make_shared<DataTypeDecimal32>(9, 2));
+ schema.children.back()->type_descriptor.decimal_precision = 9;
+ schema.children.back()->type_descriptor.decimal_scale = 2;
+ decode(std::move(schema),
+ nullable_decimal<ColumnDecimal32, Decimal32>(2, {Decimal32
{12345}, Decimal32 {-1}}),
+ [](const auto& values) {
+ EXPECT_EQ(values.get_value_ref(0).get_decimal(),
(VariantDecimal {12345, 2, 4}));
+ EXPECT_EQ(values.get_value_ref(1).get_decimal(),
(VariantDecimal {-1, 2, 4}));
+ });
+ }
+ {
+ auto schema =
shredded_primitive_schema(std::make_shared<DataTypeDecimal64>(18, 3));
+ schema.children.back()->type_descriptor.decimal_precision = 18;
+ schema.children.back()->type_descriptor.decimal_scale = 3;
+ decode(std::move(schema),
+ nullable_decimal<ColumnDecimal64, Decimal64>(
+ 3, {Decimal64 {123456789}, Decimal64 {-123456789}}),
+ [](const auto& values) {
+ EXPECT_EQ(values.get_value_ref(0).get_decimal(),
+ (VariantDecimal {123456789, 3, 8}));
+ EXPECT_EQ(values.get_value_ref(1).get_decimal(),
+ (VariantDecimal {-123456789, 3, 8}));
+ });
+ }
+ {
+ auto schema =
shredded_primitive_schema(std::make_shared<DataTypeDecimal128>(38, 4));
+ schema.children.back()->type_descriptor.decimal_precision = 38;
+ schema.children.back()->type_descriptor.decimal_scale = 4;
+ decode(std::move(schema),
+ nullable_decimal<ColumnDecimal128V3, Decimal128V3>(
+ 4, {Decimal128V3
{static_cast<Int128>(1234567890123456789LL)}}),
+ [](const auto& values) {
+ EXPECT_EQ(values.get_value_ref(0).get_decimal(),
+ (VariantDecimal {1234567890123456789LL, 4, 16}));
+ });
+ }
+
+ const auto date = DateV2Value<DateV2ValueType>::create_from_olap_date(
+ (static_cast<uint32_t>(1970) << 9) | (static_cast<uint32_t>(1) <<
5) | 2);
+ decode(shredded_primitive_schema(std::make_shared<DataTypeDateV2>()),
+ nullable_fixed<ColumnDateV2, DateV2Value<DateV2ValueType>>({date},
{0}),
+ [](const auto& values) {
EXPECT_EQ(values.get_value_ref(0).get_date(), 1); });
+
+ auto datetime =
DateV2Value<DateTimeV2ValueType>::create_from_olap_datetime(19700101000001ULL);
+ datetime.set_microsecond(234567);
+ {
+ auto schema =
shredded_primitive_schema(std::make_shared<DataTypeDateTimeV2>(6));
+ schema.children.back()->type_descriptor.time_unit =
ParquetTimeUnit::MICROS;
+ schema.children.back()->type_descriptor.timestamp_is_adjusted_to_utc =
false;
+ decode(std::move(schema),
+ nullable_fixed<ColumnDateTimeV2,
DateV2Value<DateTimeV2ValueType>>({datetime}, {0}),
+ [](const auto& values) {
+
EXPECT_EQ(values.get_value_ref(0).get_timestamp_ntz_micros(), 1234567);
+ });
+ }
+ TimestampTzValue timestamp;
+ timestamp.unchecked_set_time(1970, 1, 1, 0, 0, 2, 345678);
+ {
+ auto schema =
shredded_primitive_schema(std::make_shared<DataTypeTimeStampTz>(6));
+ schema.children.back()->type_descriptor.time_unit =
ParquetTimeUnit::MICROS;
+ schema.children.back()->type_descriptor.timestamp_is_adjusted_to_utc =
true;
+ decode(std::move(schema),
+ nullable_fixed<ColumnTimeStampTz,
TimestampTzValue>({timestamp}, {0}),
+ [](const auto& values) {
+ EXPECT_EQ(values.get_value_ref(0).get_timestamp_micros(),
2345678);
+ });
+ }
+
+ auto verify_bytes = [&](bool string_annotation, bool uuid) {
+ const std::array<uint8_t, 16> bytes {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10,
11, 12, 13, 14, 15};
+ auto schema =
shredded_primitive_schema(std::make_shared<DataTypeString>());
+ schema.children.back()->type_descriptor.is_string_annotation =
string_annotation;
+ schema.children.back()->type_descriptor.is_uuid = uuid;
+ auto strings = ColumnString::create();
+ if (uuid) {
+ strings->insert_data(reinterpret_cast<const char*>(bytes.data()),
bytes.size());
+ } else {
+ strings->insert_data("bytes", 5);
+ }
+ auto typed = ColumnNullable::create(std::move(strings),
ColumnUInt8::create(1, 0));
+ decode(std::move(schema), std::move(typed), [&](const auto& values) {
+ if (uuid) {
+ EXPECT_EQ(values.get_value_ref(0).get_uuid(), bytes);
+ } else if (string_annotation) {
+ EXPECT_EQ(values.get_value_ref(0).get_string(),
StringRef("bytes"));
+ } else {
+ EXPECT_EQ(values.get_value_ref(0).get_binary(),
StringRef("bytes"));
+ }
+ });
+ };
+ verify_bytes(false, false);
+ verify_bytes(true, false);
+ verify_bytes(false, true);
+}
+
+TEST(VariantColumnReaderTest, RejectsInvalidShreddedUuidWidth) {
+ auto schema =
shredded_primitive_schema(std::make_shared<DataTypeString>());
+ schema.children.back()->type_descriptor.is_uuid = true;
+ auto strings = ColumnString::create();
+ strings->insert_data("short", 5);
+ auto typed = ColumnNullable::create(std::move(strings),
ColumnUInt8::create(1, 0));
+ const std::string error =
+ materialization_error(schema,
shredded_primitive_physical(std::move(typed)));
+ EXPECT_NE(error.find("UUID has 5 bytes instead of 16"), std::string::npos)
<< error;
+}
+
TEST(VariantColumnReaderTest, DifferentMetadataDictionariesRemainIndependent) {
VariantBatchBuilder first_builder;
auto first_row = first_builder.begin_row();
@@ -349,6 +657,31 @@ TEST(VariantColumnReaderTest,
DifferentMetadataDictionariesRemainIndependent) {
EXPECT_EQ(field.get_int(), 2);
}
+TEST(VariantColumnReaderTest,
AppendsCompleteShreddedStatesWithDifferentSchemasAndMetadata) {
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ auto first_schema = shredded_named_object_schema("a");
+ ASSERT_TRUE(materialize_variant_rows(first_schema,
+
complete_shredded_object_physical("left", 1, 11), output)
+ .ok());
+ auto second_schema = shredded_named_object_schema("b");
+ ASSERT_TRUE(materialize_variant_rows(second_schema,
+
complete_shredded_object_physical("right", 2, 22), output)
+ .ok());
+
+ const auto& variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(*output).get_nested_column());
+ ASSERT_EQ(variants.size(), 2);
+ VariantRef field;
+ ASSERT_TRUE(variants.get_value_ref(0).object_find(StringRef("left"),
&field));
+ EXPECT_EQ(field.get_int(), 1);
+ ASSERT_TRUE(variants.get_value_ref(0).object_find(StringRef("a"), &field));
+ EXPECT_EQ(field.get_int(), 11);
+ ASSERT_TRUE(variants.get_value_ref(1).object_find(StringRef("right"),
&field));
+ EXPECT_EQ(field.get_int(), 2);
+ ASSERT_TRUE(variants.get_value_ref(1).object_find(StringRef("b"), &field));
+ EXPECT_EQ(field.get_int(), 22);
+}
+
TEST(VariantColumnReaderTest, ShreddedObjectFieldMayOmitResidualValueColumn) {
const std::array<char, 1> ignored {0};
const StringRef metadata(VARIANT_EMPTY_METADATA.data(),
VARIANT_EMPTY_METADATA.size());
@@ -776,6 +1109,449 @@ TEST(VariantColumnReaderTest,
MaterializesShreddedArrayElements) {
EXPECT_EQ(value.array_at(1).get_int(), 4);
}
+TEST(VariantColumnReaderTest, RejectsCorruptShreddedWrappersWithoutCrashing) {
+ 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 std::array<char, 1> ignored {0};
+ const StringRef metadata(VARIANT_EMPTY_METADATA.data(),
VARIANT_EMPTY_METADATA.size());
+ const StringRef residual_int(int_seven.data(), int_seven.size());
+ auto expect_error = [](const std::string& error, std::string_view
expected) {
+ EXPECT_NE(error.find(expected), std::string::npos) << error;
+ };
+ std::string_view current_case;
+
+ try {
+ {
+ current_case = "null metadata";
+ SCOPED_TRACE("null metadata");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {1}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ expect_error(
+ materialization_error(unshredded_schema(),
root_wrapper(std::move(fields))),
+ "null metadata");
+ }
+ {
+ current_case = "wrapper without carriers";
+ SCOPED_TRACE("wrapper without carriers");
+ auto schema = unshredded_schema();
+ schema.children.pop_back();
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ expect_error(materialization_error(schema,
root_wrapper(std::move(fields))),
+ "neither value nor typed_value");
+ }
+ {
+ current_case = "scalar with residual";
+ SCOPED_TRACE("scalar with residual");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ fields.push_back(nullable_int64({8}, {0}));
+ expect_error(
+ materialization_error(shredded_int64_schema(),
root_wrapper(std::move(fields))),
+ "scalar typed_value cannot have residual");
+ }
+ {
+ current_case = "object with scalar residual";
+ SCOPED_TRACE("object with scalar residual");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ MutableColumns wrapper_fields;
+ wrapper_fields.push_back(nullable_int64({9}, {0}));
+ MutableColumns object_fields;
+ object_fields.push_back(ColumnNullable::create(
+ ColumnStruct::create(std::move(wrapper_fields)),
ColumnUInt8::create(1, 0)));
+
fields.push_back(ColumnNullable::create(ColumnStruct::create(std::move(object_fields)),
+ ColumnUInt8::create(1,
0)));
+ expect_error(materialization_error(shredded_object_schema(),
+
root_wrapper(std::move(fields))),
+ "non-object residual");
+ }
+ {
+ current_case = "object field count mismatch";
+ SCOPED_TRACE("object field count mismatch");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({{ignored.data(), 0}}, {1}));
+ MutableColumns unexpected_object_fields;
+ unexpected_object_fields.push_back(nullable_int64({1}, {0}));
+ unexpected_object_fields.push_back(nullable_int64({2}, {0}));
+ fields.push_back(ColumnNullable::create(
+ ColumnStruct::create(std::move(unexpected_object_fields)),
+ ColumnUInt8::create(1, 0)));
+ expect_error(materialization_error(shredded_object_schema(),
+
root_wrapper(std::move(fields))),
+ "physical field count mismatch");
+ }
+ {
+ current_case = "array with residual";
+ SCOPED_TRACE("array with residual");
+ MutableColumns empty_wrapper_fields;
+ empty_wrapper_fields.push_back(nullable_int64({}, {}));
+ auto empty_elements = ColumnNullable::create(
+ ColumnStruct::create(std::move(empty_wrapper_fields)),
ColumnUInt8::create());
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->insert_value(0);
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ fields.push_back(ColumnNullable::create(
+ ColumnArray::create(std::move(empty_elements),
std::move(offsets)),
+ ColumnUInt8::create(1, 0)));
+ expect_error(
+ materialization_error(shredded_array_schema(),
root_wrapper(std::move(fields))),
+ "array typed_value cannot have residual");
+ }
+ {
+ current_case = "null array element wrapper";
+ SCOPED_TRACE("null array element wrapper");
+ MutableColumns wrapper_fields;
+ wrapper_fields.push_back(nullable_int64({0}, {1}));
+ auto wrappers = ColumnStruct::create(std::move(wrapper_fields));
+ auto elements = ColumnNullable::create(std::move(wrappers),
ColumnUInt8::create(1, 1));
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->insert_value(1);
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({{ignored.data(), 0}}, {1}));
+ fields.push_back(ColumnNullable::create(
+ ColumnArray::create(std::move(elements),
std::move(offsets)),
+ ColumnUInt8::create(1, 0)));
+ expect_error(
+ materialization_error(shredded_array_schema(),
root_wrapper(std::move(fields))),
+ "array element wrapper is null");
+ }
+ {
+ current_case = "missing array element";
+ SCOPED_TRACE("missing array element");
+ MutableColumns element_fields;
+ element_fields.push_back(nullable_strings({{ignored.data(), 0}},
{1}));
+ element_fields.push_back(nullable_int64({0}, {1}));
+ auto elements =
ColumnNullable::create(ColumnStruct::create(std::move(element_fields)),
+ ColumnUInt8::create(1, 0));
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->insert_value(1);
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({{ignored.data(), 0}}, {1}));
+ fields.push_back(ColumnNullable::create(
+ ColumnArray::create(std::move(elements),
std::move(offsets)),
+ ColumnUInt8::create(1, 0)));
+ expect_error(materialization_error(shredded_mixed_array_schema(),
+
root_wrapper(std::move(fields))),
+ "array element is missing");
+ }
+ {
+ current_case = "root field count mismatch";
+ SCOPED_TRACE("root field count mismatch");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ fields.push_back(nullable_int64({8}, {0}));
+ fields.push_back(nullable_int64({9}, {0}));
+ expect_error(
+ materialization_error(shredded_int64_schema(),
root_wrapper(std::move(fields))),
+ "physical field count mismatch");
+ }
+ {
+ current_case = "invalid metadata";
+ SCOPED_TRACE("invalid metadata");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({StringRef("bad")}, {0}));
+ fields.push_back(nullable_strings({residual_int}, {0}));
+ expect_error(
+ materialization_error(unshredded_schema(),
root_wrapper(std::move(fields))),
+ "metadata");
+ }
+ {
+ current_case = "invalid residual value";
+ SCOPED_TRACE("invalid residual value");
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({{invalid_value.data(),
invalid_value.size()}}, {0}));
+ expect_error(
+ materialization_error(unshredded_schema(),
root_wrapper(std::move(fields))),
+ "Variant");
+ }
+ } catch (const std::exception& error) {
+ FAIL() << "Unexpected exception in " << current_case << ": " <<
error.what();
+ }
+}
+
+TEST(VariantColumnReaderTest, ImmediateCorruptionLeavesDestinationUnchanged) {
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ ASSERT_TRUE(
+ materialize_variant_rows(shredded_int64_schema(),
shredded_int64_physical({7}), output)
+ .ok());
+ MutableColumns invalid_fields;
+ invalid_fields.push_back(nullable_strings(
+ {{VARIANT_EMPTY_METADATA.data(), VARIANT_EMPTY_METADATA.size()}},
{0}));
+ const Status status = materialize_variant_rows(shredded_int64_schema(),
+
root_wrapper(std::move(invalid_fields)), output);
+ EXPECT_FALSE(status.ok());
+ ASSERT_EQ(output->size(), 1);
+ const auto& variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(*output).get_nested_column());
+ EXPECT_EQ(variants.get_value_ref(0).get_int(), 7);
+}
+
+TEST(VariantColumnReaderTest, LazyNestedCorruptionLeavesDestinationUnchanged) {
+ const std::array<char, 1> invalid_value {static_cast<char>(0xff)};
+ const StringRef metadata(VARIANT_EMPTY_METADATA.data(),
VARIANT_EMPTY_METADATA.size());
+ auto corrupt_variant = [&]() {
+ MutableColumns fields;
+ fields.push_back(nullable_strings({metadata}, {0}));
+ fields.push_back(nullable_strings({{invalid_value.data(),
invalid_value.size()}}, {0}));
+ return root_wrapper(std::move(fields));
+ };
+ auto label_schema = []() {
+ auto schema = std::make_unique<ParquetColumnSchema>();
+ schema->name = "label";
+ schema->kind = ParquetColumnSchemaKind::PRIMITIVE;
+ schema->type = make_nullable(std::make_shared<DataTypeString>());
+ return schema;
+ };
+ auto make_plan = [](const ParquetColumnSchema& root) {
+ auto build = [&](auto&& self, const ParquetColumnSchema* schema)
+ -> std::unique_ptr<VariantMaterializationNode> {
+ auto node = std::make_unique<VariantMaterializationNode>();
+ node->schema = schema;
+ node->contains_variant = schema->kind ==
ParquetColumnSchemaKind::VARIANT;
+ for (const auto& child_schema : schema->children) {
+ auto child = self(self, child_schema.get());
+ node->contains_variant = node->contains_variant ||
child->contains_variant;
+ node->children.push_back(std::move(child));
+ }
+ return node;
+ };
+ return build(build, &root);
+ };
+ auto make_struct_schema = [&](ParquetColumnSchema variant_schema) {
+ ParquetColumnSchema root;
+ root.name = "row";
+ root.kind = ParquetColumnSchemaKind::STRUCT;
+ root.children.push_back(label_schema());
+
root.children.push_back(std::make_unique<ParquetColumnSchema>(std::move(variant_schema)));
+ return root;
+ };
+ auto make_struct_physical = [&](std::string_view label, MutableColumnPtr
variant) {
+ MutableColumns fields;
+ fields.push_back(nullable_strings({StringRef(label.data(),
label.size())}, {0}));
+ fields.push_back(std::move(variant));
+ return ColumnStruct::create(std::move(fields));
+ };
+ const auto element_type = std::make_shared<DataTypeStruct>(
+ DataTypes {make_nullable(std::make_shared<DataTypeString>()),
+ make_nullable(std::make_shared<DataTypeVariantV2>())},
+ Strings {"label", "payload"});
+
+ {
+ auto output = element_type->create_column();
+ auto valid_schema = make_struct_schema(shredded_int64_schema());
+ auto valid_plan = make_plan(valid_schema);
+ ASSERT_TRUE(materialize_variant_columns(
+ *valid_plan,
+ *make_struct_physical("before",
shredded_int64_physical({7})), output)
+ .ok());
+
+ auto corrupt_schema = make_struct_schema(unshredded_schema());
+ auto corrupt_plan = make_plan(corrupt_schema);
+ const Status status = materialize_variant_columns(
+ *corrupt_plan, *make_struct_physical("after",
corrupt_variant()), output);
+ EXPECT_FALSE(status.ok());
+
+ const auto& structure = assert_cast<const ColumnStruct&>(*output);
+ const auto& label = assert_cast<const
ColumnNullable&>(structure.get_column(0));
+ EXPECT_EQ(label.size(), 1);
+ EXPECT_EQ(label.get_null_map_data(), (NullMap {0}));
+ EXPECT_EQ(label.get_nested_column().get_data_at(0).to_string(),
"before");
+ const auto& payload = assert_cast<const
ColumnNullable&>(structure.get_column(1));
+ EXPECT_EQ(payload.size(), 1);
+ EXPECT_EQ(payload.get_null_map_data(), (NullMap {0}));
+ EXPECT_EQ(assert_cast<const
ColumnVariantV2&>(payload.get_nested_column())
+ .get_value_ref(0)
+ .get_int(),
+ 7);
+ }
+
+ {
+ auto output =
std::make_shared<DataTypeArray>(element_type)->create_column();
+ auto valid_element_schema =
make_struct_schema(shredded_int64_schema());
+ ParquetColumnSchema valid_schema;
+ valid_schema.name = "rows";
+ valid_schema.kind = ParquetColumnSchemaKind::LIST;
+ valid_schema.children.push_back(
+
std::make_unique<ParquetColumnSchema>(std::move(valid_element_schema)));
+ auto valid_plan = make_plan(valid_schema);
+ auto valid_offsets = ColumnArray::ColumnOffsets::create();
+ valid_offsets->insert_value(1);
+ auto valid_physical =
+ ColumnArray::create(make_struct_physical("before",
shredded_int64_physical({7})),
+ std::move(valid_offsets));
+ ASSERT_TRUE(materialize_variant_columns(*valid_plan, *valid_physical,
output).ok());
+
+ auto corrupt_element_schema = make_struct_schema(unshredded_schema());
+ ParquetColumnSchema corrupt_schema;
+ corrupt_schema.name = "rows";
+ corrupt_schema.kind = ParquetColumnSchemaKind::LIST;
+ corrupt_schema.children.push_back(
+
std::make_unique<ParquetColumnSchema>(std::move(corrupt_element_schema)));
+ auto corrupt_plan = make_plan(corrupt_schema);
+ auto corrupt_offsets = ColumnArray::ColumnOffsets::create();
+ corrupt_offsets->insert_value(1);
+ auto corrupt_physical = ColumnArray::create(
+ make_struct_physical("after", corrupt_variant()),
std::move(corrupt_offsets));
+ const Status status = materialize_variant_columns(*corrupt_plan,
*corrupt_physical, output);
+ EXPECT_FALSE(status.ok());
+
+ const auto& array = assert_cast<const ColumnArray&>(*output);
+ EXPECT_EQ(array.get_offsets(), (ColumnArray::Offsets64 {1}));
+ const auto& element = assert_cast<const
ColumnNullable&>(array.get_data());
+ EXPECT_EQ(element.get_null_map_data(), (NullMap {0}));
+ const auto& structure = assert_cast<const
ColumnStruct&>(element.get_nested_column());
+ const auto& label = assert_cast<const
ColumnNullable&>(structure.get_column(0));
+ EXPECT_EQ(label.size(), 1);
+ EXPECT_EQ(label.get_null_map_data(), (NullMap {0}));
+ EXPECT_EQ(label.get_nested_column().get_data_at(0).to_string(),
"before");
+ const auto& payload = assert_cast<const
ColumnNullable&>(structure.get_column(1));
+ EXPECT_EQ(payload.size(), 1);
+ EXPECT_EQ(payload.get_null_map_data(), (NullMap {0}));
+ EXPECT_EQ(assert_cast<const
ColumnVariantV2&>(payload.get_nested_column())
+ .get_value_ref(0)
+ .get_int(),
+ 7);
+ }
+
+ {
+ auto output =
+
std::make_shared<DataTypeMap>(make_nullable(std::make_shared<DataTypeString>()),
+
make_nullable(std::make_shared<DataTypeVariantV2>()))
+ ->create_column();
+ auto make_map_schema = [&](ParquetColumnSchema variant_schema) {
+ ParquetColumnSchema root;
+ root.name = "entries";
+ root.kind = ParquetColumnSchemaKind::MAP;
+ root.children.push_back(label_schema());
+ root.children.push_back(
+
std::make_unique<ParquetColumnSchema>(std::move(variant_schema)));
+ return root;
+ };
+ auto make_map_physical = [&](std::string_view key, MutableColumnPtr
variant) {
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->insert_value(1);
+ return ColumnMap::create(nullable_strings({StringRef(key.data(),
key.size())}, {0}),
+ std::move(variant), std::move(offsets));
+ };
+
+ auto valid_schema = make_map_schema(shredded_int64_schema());
+ auto valid_plan = make_plan(valid_schema);
+ ASSERT_TRUE(materialize_variant_columns(
+ *valid_plan, *make_map_physical("before",
shredded_int64_physical({7})),
+ output)
+ .ok());
+ auto corrupt_schema = make_map_schema(unshredded_schema());
+ auto corrupt_plan = make_plan(corrupt_schema);
+ const Status status = materialize_variant_columns(
+ *corrupt_plan, *make_map_physical("after", corrupt_variant()),
output);
+ EXPECT_FALSE(status.ok());
+
+ const auto& map = assert_cast<const ColumnMap&>(*output);
+ EXPECT_EQ(map.get_offsets(), (ColumnArray::Offsets64 {1}));
+ const auto& keys = assert_cast<const ColumnNullable&>(map.get_keys());
+ EXPECT_EQ(keys.size(), 1);
+ EXPECT_EQ(keys.get_null_map_data(), (NullMap {0}));
+ EXPECT_EQ(keys.get_nested_column().get_data_at(0).to_string(),
"before");
+ const auto& values = assert_cast<const
ColumnNullable&>(map.get_values());
+ EXPECT_EQ(values.size(), 1);
+ EXPECT_EQ(values.get_null_map_data(), (NullMap {0}));
+ EXPECT_EQ(assert_cast<const
ColumnVariantV2&>(values.get_nested_column())
+ .get_value_ref(0)
+ .get_int(),
+ 7);
+ }
+}
+
+TEST(VariantColumnReaderTest, MaterializesMixedRootArraysAndNullKinds) {
+ VariantBatchBuilder residual_builder;
+ {
+ auto row = residual_builder.begin_row();
+ row.add_null();
+ row.finish();
+ }
+ {
+ auto row = residual_builder.begin_row();
+ auto object = row.start_object();
+ object.add_key(StringRef("x"));
+ row.add_int(2);
+ object.finish();
+ row.finish();
+ }
+ {
+ auto row = residual_builder.begin_row();
+ auto array = row.start_array();
+ row.add_int(3);
+ row.add_int(4);
+ array.finish();
+ row.finish();
+ }
+ {
+ auto row = residual_builder.begin_row();
+ row.add_string(StringRef("tail"));
+ row.finish();
+ }
+ VariantBatchBuilder residuals = residual_builder.finish_batch();
+ const VariantRef first = residuals.value_at(0);
+ std::vector<StringRef> residual_values;
+ for (size_t row = 0; row < residuals.num_rows(); ++row) {
+ residual_values.push_back(residuals.value_at(row).value);
+ }
+ residual_values.insert(residual_values.begin() + 1, StringRef {});
+
+ MutableColumns element_fields;
+ element_fields.push_back(nullable_strings(residual_values, {0, 1, 0, 0,
0}));
+ element_fields.push_back(nullable_int64({0, 1, 0, 0, 0}, {1, 0, 1, 1, 1}));
+ auto elements =
ColumnNullable::create(ColumnStruct::create(std::move(element_fields)),
+ ColumnUInt8::create(5, 0));
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->get_data().assign({0, 5, 5, 5});
+ auto arrays = ColumnArray::create(std::move(elements), std::move(offsets));
+
+ const StringRef metadata(first.metadata.data, first.metadata.size);
+ const std::array<char, 1> ignored {0};
+ MutableColumns root_fields;
+ root_fields.push_back(nullable_strings({metadata, metadata, metadata,
metadata}, {0, 0, 0, 0}));
+ root_fields.push_back(nullable_strings(
+ {{ignored.data(), 0}, {ignored.data(), 0}, {ignored.data(), 0},
{ignored.data(), 0}},
+ {1, 1, 1, 1}));
+ auto typed_nulls = ColumnUInt8::create(4, 0);
+ typed_nulls->get_data()[2] = 1;
+ typed_nulls->get_data()[3] = 1;
+ root_fields.push_back(ColumnNullable::create(std::move(arrays),
std::move(typed_nulls)));
+ auto physical = root_wrapper(std::move(root_fields), {0, 0, 0, 1});
+
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ ASSERT_TRUE(materialize_variant_rows(shredded_mixed_array_schema(),
*physical, output).ok());
+ const auto& nullable = assert_cast<const ColumnNullable&>(*output);
+ EXPECT_EQ(nullable.get_null_map_data(), (NullMap {0, 0, 0, 1}));
+ const auto& variants = assert_cast<const
ColumnVariantV2&>(nullable.get_nested_column());
+ EXPECT_EQ(variants.get_value_ref(0).num_elements(), 0);
+ const VariantRef mixed = variants.get_value_ref(1);
+ ASSERT_EQ(mixed.num_elements(), 5);
+ EXPECT_TRUE(mixed.array_at(0).is_null());
+ EXPECT_EQ(mixed.array_at(1).get_int(), 1);
+ VariantRef object_field;
+ ASSERT_TRUE(mixed.array_at(2).object_find(StringRef("x"), &object_field));
+ EXPECT_EQ(object_field.get_int(), 2);
+ EXPECT_EQ(mixed.array_at(3).array_at(1).get_int(), 4);
+ EXPECT_EQ(mixed.array_at(4).get_string(), StringRef("tail"));
+ EXPECT_TRUE(variants.get_value_ref(2).is_null());
+}
+
TEST(VariantColumnReaderTest, MaterializesVariantNestedInStruct) {
const std::array<char, 2> int_seven {
static_cast<char>(static_cast<uint8_t>(VariantPrimitiveId::INT8)
@@ -817,6 +1593,124 @@ TEST(VariantColumnReaderTest,
MaterializesVariantNestedInStruct) {
EXPECT_EQ(variants.get_value_ref(0).get_int(), 7);
}
+TEST(VariantColumnReaderTest,
MaterializesPhysicallyShreddedVariantInStructArrayAndMap) {
+ auto make_plan_child = [](const ParquetColumnSchema* schema) {
+ auto child = std::make_unique<VariantMaterializationNode>();
+ child->schema = schema;
+ child->contains_variant = schema->kind ==
ParquetColumnSchemaKind::VARIANT;
+ return child;
+ };
+
+ {
+ ParquetColumnSchema root_schema;
+ root_schema.name = "root";
+ root_schema.kind = ParquetColumnSchemaKind::STRUCT;
+ root_schema.children.push_back(
+
std::make_unique<ParquetColumnSchema>(shredded_int64_schema()));
+ VariantMaterializationNode plan;
+ plan.schema = &root_schema;
+ plan.contains_variant = true;
+
plan.children.push_back(make_plan_child(root_schema.children[0].get()));
+ MutableColumns physical_fields;
+ physical_fields.push_back(shredded_int64_physical({11}));
+ auto physical = ColumnStruct::create(std::move(physical_fields));
+ auto output = std::make_shared<DataTypeStruct>(
+ DataTypes
{make_nullable(std::make_shared<DataTypeVariantV2>())},
+ Strings {"v"})
+ ->create_column();
+ ASSERT_TRUE(materialize_variant_columns(plan, *physical, output).ok());
+ const auto& variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(
+ assert_cast<const
ColumnStruct&>(*output).get_column(0))
+ .get_nested_column());
+ EXPECT_EQ(variants.get_value_ref(0).get_int(), 11);
+ }
+
+ {
+ ParquetColumnSchema root_schema;
+ root_schema.name = "items";
+ root_schema.kind = ParquetColumnSchemaKind::LIST;
+ root_schema.children.push_back(
+
std::make_unique<ParquetColumnSchema>(shredded_int64_schema()));
+ VariantMaterializationNode plan;
+ plan.schema = &root_schema;
+ plan.contains_variant = true;
+
plan.children.push_back(make_plan_child(root_schema.children[0].get()));
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->insert_value(2);
+ auto physical = ColumnArray::create(shredded_int64_physical({12, 13}),
std::move(offsets));
+ auto output = std::make_shared<DataTypeArray>(
+
make_nullable(std::make_shared<DataTypeVariantV2>()))
+ ->create_column();
+ ASSERT_TRUE(materialize_variant_columns(plan, *physical, output).ok());
+ const auto& variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(
+ assert_cast<const ColumnArray&>(*output).get_data())
+ .get_nested_column());
+ EXPECT_EQ(variants.get_value_ref(0).get_int(), 12);
+ EXPECT_EQ(variants.get_value_ref(1).get_int(), 13);
+ }
+
+ {
+ ParquetColumnSchema root_schema;
+ root_schema.name = "entries";
+ root_schema.kind = ParquetColumnSchemaKind::MAP;
+ auto key_schema = std::make_unique<ParquetColumnSchema>();
+ key_schema->name = "key";
+ key_schema->kind = ParquetColumnSchemaKind::PRIMITIVE;
+ key_schema->type = std::make_shared<DataTypeString>();
+ root_schema.children.push_back(std::move(key_schema));
+ root_schema.children.push_back(
+
std::make_unique<ParquetColumnSchema>(shredded_int64_schema()));
+ VariantMaterializationNode plan;
+ plan.schema = &root_schema;
+ plan.contains_variant = true;
+
plan.children.push_back(make_plan_child(root_schema.children[0].get()));
+
plan.children.push_back(make_plan_child(root_schema.children[1].get()));
+ auto keys = ColumnString::create();
+ keys->insert_data("a", 1);
+ keys->insert_data("b", 1);
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->insert_value(2);
+ auto physical = ColumnMap::create(std::move(keys),
shredded_int64_physical({14, 15}),
+ std::move(offsets));
+ auto output =
+
std::make_shared<DataTypeMap>(std::make_shared<DataTypeString>(),
+
make_nullable(std::make_shared<DataTypeVariantV2>()))
+ ->create_column();
+ ASSERT_TRUE(materialize_variant_columns(plan, *physical, output).ok());
+ const auto& variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(
+ assert_cast<const ColumnMap&>(*output).get_values())
+ .get_nested_column());
+ EXPECT_EQ(variants.get_value_ref(0).get_int(), 14);
+ EXPECT_EQ(variants.get_value_ref(1).get_int(), 15);
+ }
+}
+
+TEST(VariantColumnReaderTest,
ProjectedShreddedStateRejectsRootMaterialization) {
+ auto schema = shredded_object_schema();
+ schema.local_id = 0;
+ schema.children[2]->local_id = 2;
+ schema.children[2]->children[0]->local_id = 0;
+ schema.children[2]->children[0]->children[0]->local_id = 0;
+ auto projection = format::LocalColumnIndex::partial_local(0);
+ projection.children.push_back(format::LocalColumnIndex::partial_local(2));
+
projection.children.back().children.push_back(format::LocalColumnIndex::partial_local(0));
+ projection.children.back().children.back().children.push_back(
+ format::LocalColumnIndex::local(0));
+ VariantMaterializationNode plan;
+ plan.schema = &schema;
+ plan.contains_variant = true;
+ plan.variant_projection = std::move(projection);
+ auto output =
make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
+ ASSERT_TRUE(materialize_variant_columns(plan,
projected_shredded_object_physical({17}), output)
+ .ok());
+ const auto& variants = assert_cast<const ColumnVariantV2&>(
+ assert_cast<const ColumnNullable&>(*output).get_nested_column());
+ EXPECT_THROW((void)variants.get_value_ref(0), Exception);
+}
+
TEST(VariantColumnReaderTest, AlignsNestedPrimitiveNullabilityAroundVariant) {
const std::array<char, 2> int_seven {
static_cast<char>(static_cast<uint8_t>(VariantPrimitiveId::INT8)
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_variant_read.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_variant_read.out
index a7c41169d8d..88408fc6dea 100644
---
a/regression-test/data/external_table_p0/iceberg/test_iceberg_variant_read.out
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_variant_read.out
@@ -21,6 +21,13 @@
8 false {"n":30,"name":"same","ok":true}
9 false
{"arr":[5,6],"n":40,"name":"carol","nested":{"city":"bj"},"new_key":"new","ok":true,"ratio":4.5}
+-- !variant_root_array_projection --
+1 false [] \N \N \N \N
+2 false [null,1,{"x":2},[3,4],"tail"] null 1 2 4
+3 false [{"nested":[null,{"y":5}]}] {"nested":[null,{"y":5}]}
\N \N \N
+4 false null \N \N \N \N
+5 true \N \N \N \N \N
+
-- !variant_path_expressions --
1 ALICE 11 1.5 true 3 hz
10 DAVE 51 5.5 false 15 sz
@@ -47,13 +54,52 @@
8 30
9 40
--- !variant_implicit_shredded_filter --
+-- !variant_multi_file_serial --
+2 20 \N 2 \N {"b":2,"shared":20,"z":200}
+3 30 3 \N \N {"a":3,"shared":30,"z":300}
+4 40 \N \N \N {"c":4,"shared":40}
+5 50 \N 5 500
{"b":5,"new_field":{"k":500},"shared":50}
+
+-- !variant_multi_file_parallel --
+2 20 \N 2 \N {"b":2,"shared":20,"z":200}
+3 30 3 \N \N {"a":3,"shared":30,"z":300}
+4 40 \N \N \N {"c":4,"shared":40}
+5 50 \N 5 500
{"b":5,"new_field":{"k":500},"shared":50}
+
+-- !variant_type_matrix --
+true -128 -32768 2147483647 -9223372036854775808 true true
-1234567890.1234 1970-01-02 1970-01-01T00:00:01.234567
"YmluYXJ5" false
+
+-- !variant_multi_row_group_result --
+192 8000 8191 1554336
+
+-- !variant_deletion_vector_current --
+2048 0 4094 4192256
+
+-- !variant_deletion_vector_before_delete --
+4096 0 4095 8386560
+
+-- !variant_equality_delete_current --
+1 10 keep-one {"label":"keep-one","n":10}
+3 30 keep-three {"label":"keep-three","n":30}
+
+-- !variant_equality_delete_before_delete --
+1 10 keep-one {"label":"keep-one","n":10}
+2 20 delete {"label":"delete","n":20}
+3 30 keep-three {"label":"keep-three","n":30}
+
+-- !variant_implicit_filter --
10
{"arr":[7,8],"n":50,"name":"dave","nested":{"city":"sz"},"ok":false,"ratio":5.5}
11
{"arr":[9,10],"n":60,"name":null,"nested":{"city":null},"ok":true,"ratio":6.5}
9
{"arr":[5,6],"n":40,"name":"carol","nested":{"city":"bj"},"new_key":"new","ok":true,"ratio":4.5}
+-- !variant_shredded_only_time_travel --
+1095 3001 4095 3885060
+
+-- !variant_mixed_before_delete --
+1096 3001 5000 3890060
+
-- !variant_page_pruning_result --
-1095 3001 4095
+1094 3001 4094
-- !variant_aggregate --
false 2 70 4
@@ -112,6 +158,15 @@ true 5 170 4.17
-- !variant_delete_only_merge --
0
+-- !variant_position_delete_alignment --
+10 dave 50
{"arr":[7,8],"n":50,"name":"dave","nested":{"city":"sz"},"ok":false,"ratio":5.5}
+9 carol 40
{"arr":[5,6],"n":40,"name":"carol","nested":{"city":"bj"},"new_key":"new","ok":true,"ratio":4.5}
+
+-- !variant_before_position_delete --
+10 dave 50
{"arr":[7,8],"n":50,"name":"dave","nested":{"city":"sz"},"ok":false,"ratio":5.5}
+11 null 60
{"arr":[9,10],"n":60,"name":null,"nested":{"city":null},"ok":true,"ratio":6.5}
+9 carol 40
{"arr":[5,6],"n":40,"name":"carol","nested":{"city":"bj"},"new_key":"new","ok":true,"ratio":4.5}
+
-- !variant_orc_missing_column --
1 \N
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy
index 7b3bc3e833d..7161574b962 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy
@@ -56,6 +56,39 @@ suite("test_iceberg_variant_read",
.withPathStyleAccessEnabled(true)
.withCredentials(new AWSStaticCredentialsProvider(credentials))
.build()
+ def executeCommand = { String command, int timeoutSeconds = 300 ->
+ StringBuilder stdout = new StringBuilder()
+ StringBuilder stderr = new StringBuilder()
+ def process = new ProcessBuilder("/bin/bash", "-c", command).start()
+ process.consumeProcessOutput(stdout, stderr)
+ process.waitForOrKill(timeoutSeconds * 1000)
+ assertEquals(0, process.exitValue(),
+ "Command failed\nstdout:\n${stdout}\nstderr:\n${stderr}")
+ return stdout.toString()
+ }
+ String dockerCommand =
context.config.otherConfigs.get("externalDockerCommand") ?: "docker"
+ String sparkContainer =
context.config.otherConfigs.get("icebergSparkContainer")
+ if (sparkContainer == null || sparkContainer.isEmpty()) {
+ String containers = executeCommand(
+ "${dockerCommand} ps --format '{{.ID}}\t{{.Names}}'", 30)
+ def matches = []
+ containers.readLines().each { String line ->
+ String containerId = line.split(/\t/, 2)[0]
+ String probe = "${dockerCommand} exec ${containerId} bash -lc " +
+ "'test -f /mnt/SUCCESS && command -v spark-sql >/dev/null'"
+ try {
+ executeCommand(probe, 30)
+ matches.add(containerId)
+ } catch (Throwable ignored) {
+ // Only the Spark service contains the Iceberg writer
dependencies.
+ }
+ }
+ assertEquals(1, matches.size(), "Expected exactly one usable Spark
Iceberg container")
+ sparkContainer = matches[0]
+ }
+ def runInSparkContainer = { String command ->
+ executeCommand("${dockerCommand} exec ${sparkContainer} bash -lc
'${command}'", 300)
+ }
def latestSnapshotId = { String tableName ->
List<List<Object>> rows = spark_iceberg """
@@ -90,16 +123,125 @@ suite("test_iceberg_variant_read",
(6, parse_json('42')),
(7, parse_json('"root-string"'));
ALTER TABLE demo.${dbName}.variant_values SET TBLPROPERTIES (
- 'write.parquet.shred-variants'='true',
+ 'write.parquet.shred-variants'='false',
'write.parquet.variant-inference-buffer-size'='100'
);
INSERT INTO demo.${dbName}.variant_values
- WITH (`shred-variants`=true, `variant-inference-buffer-size`=100)
VALUES
+ VALUES
(8, parse_json('{"ok":true,"n":30,"name":"same"}')),
(9,
parse_json('{"name":"carol","n":40,"ratio":4.5,"ok":true,"arr":[5,6],"nested":{"city":"bj"},"new_key":"new"}')),
(10,
parse_json('{"name":"dave","n":50,"ratio":5.5,"ok":false,"arr":[7,8],"nested":{"city":"sz"}}')),
(11,
parse_json('{"name":null,"n":60,"ratio":6.5,"ok":true,"arr":[9,10],"nested":{"city":null}}'));
+ DROP TABLE IF EXISTS demo.${dbName}.variant_root_arrays;
+ CREATE TABLE demo.${dbName}.variant_root_arrays (id INT, v VARIANT)
USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false',
+ 'write.parquet.variant-inference-buffer-size'='100'
+ );
+ INSERT INTO demo.${dbName}.variant_root_arrays VALUES
+ (1, parse_json('[]')),
+ (2, parse_json('[null,1,{"x":2},[3,4],"tail"]')),
+ (3, parse_json('[{"nested":[null,{"y":5}]}]')),
+ (4, parse_json('null')),
+ (5, NULL);
+
+ DROP TABLE IF EXISTS demo.${dbName}.variant_multi_file;
+ CREATE TABLE demo.${dbName}.variant_multi_file (id INT, v VARIANT)
USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false'
+ );
+ INSERT INTO demo.${dbName}.variant_multi_file
+ VALUES (1, parse_json('{"a":1,"shared":10}'));
+ ALTER TABLE demo.${dbName}.variant_multi_file SET TBLPROPERTIES (
+ 'write.parquet.shred-variants'='false',
+ 'write.parquet.variant-inference-buffer-size'='1'
+ );
+ INSERT INTO demo.${dbName}.variant_multi_file
+ VALUES (2, parse_json('{"b":2,"shared":20,"z":200}'));
+ INSERT INTO demo.${dbName}.variant_multi_file
+ VALUES (3, parse_json('{"z":300,"shared":30,"a":3}'));
+ ALTER TABLE demo.${dbName}.variant_multi_file SET TBLPROPERTIES
+ ('write.parquet.shred-variants'='false');
+ INSERT INTO demo.${dbName}.variant_multi_file
+ VALUES (4, parse_json('{"c":4,"shared":40}'));
+ ALTER TABLE demo.${dbName}.variant_multi_file SET TBLPROPERTIES
+ ('write.parquet.shred-variants'='false');
+ INSERT INTO demo.${dbName}.variant_multi_file
+ VALUES (5,
parse_json('{"shared":50,"b":5,"new_field":{"k":500}}'));
+
+ DROP TABLE IF EXISTS demo.${dbName}.variant_type_matrix;
+ CREATE TABLE demo.${dbName}.variant_type_matrix (id INT, v VARIANT)
USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false',
+ 'write.parquet.variant-inference-buffer-size'='100'
+ );
+ INSERT INTO demo.${dbName}.variant_type_matrix SELECT 1,
to_variant_object(named_struct(
+ 'bool_value', true,
+ 'tiny_value', CAST(-128 AS TINYINT),
+ 'small_value', CAST(-32768 AS SMALLINT),
+ 'int_value', CAST(2147483647 AS INT),
+ 'big_value', CAST('-9223372036854775808' AS BIGINT),
+ 'float_value', CAST('NaN' AS FLOAT),
+ 'double_value', CAST('Infinity' AS DOUBLE),
+ 'decimal_value', CAST('-1234567890.1234' AS DECIMAL(20, 4)),
+ 'date_value', CAST('1970-01-02' AS DATE),
+ 'timestamp_value', TIMESTAMP'1970-01-01 00:00:01.234567',
+ 'binary_value', CAST('binary' AS BINARY),
+ 'null_value', CAST(NULL AS INT)
+ ));
+
+ DROP TABLE IF EXISTS demo.${dbName}.variant_multi_row_group;
+ CREATE TABLE demo.${dbName}.variant_multi_row_group (id INT, v
VARIANT) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false',
+ 'write.parquet.variant-inference-buffer-size'='100',
+ 'write.parquet.row-group-size-bytes'='4096'
+ );
+ SET spark.sql.shuffle.partitions=1;
+ INSERT INTO demo.${dbName}.variant_multi_row_group
+ SELECT /*+ COALESCE(1) */ CAST(id AS INT), parse_json(concat(
+ '{"n":', id, ',"padding":"', repeat('x', 256), '"}'))
+ FROM range(0, 8192);
+
+ DROP TABLE IF EXISTS demo.${dbName}.variant_deletion_vector;
+ CREATE TABLE demo.${dbName}.variant_deletion_vector (id INT, v
VARIANT) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false',
+ 'write.parquet.variant-inference-buffer-size'='100',
+ 'write.delete.mode'='merge-on-read',
+ 'read.parquet.vectorization.enabled'='false',
+ 'write.parquet.row-group-size-bytes'='4096'
+ );
+ INSERT INTO demo.${dbName}.variant_deletion_vector
+ SELECT /*+ COALESCE(1) */ CAST(id AS INT), parse_json(concat('{"n":',
id, ',"keep":',
+ IF(id % 2 = 0, 'true', 'false'), '}'))
+ FROM range(0, 4096);
+ RESET spark.sql.shuffle.partitions;
+
+ DROP TABLE IF EXISTS demo.${dbName}.variant_equality_delete;
+ CREATE TABLE demo.${dbName}.variant_equality_delete (id INT, v
VARIANT) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false',
+ 'write.parquet.variant-inference-buffer-size'='100'
+ );
+ INSERT INTO demo.${dbName}.variant_equality_delete VALUES
+ (1, parse_json('{"n":10,"label":"keep-one"}')),
+ (2, parse_json('{"n":20,"label":"delete"}')),
+ (3, parse_json('{"n":30,"label":"keep-three"}'));
+
DROP TABLE IF EXISTS demo.${dbName}.variant_page_pruning;
DROP TABLE IF EXISTS demo.${dbName}.variant_nested;
@@ -112,7 +254,7 @@ suite("test_iceberg_variant_read",
TBLPROPERTIES (
'format-version'='3',
'write.format.default'='parquet',
- 'write.parquet.shred-variants'='true',
+ 'write.parquet.shred-variants'='false',
'write.parquet.variant-inference-buffer-size'='100'
);
INSERT INTO demo.${dbName}.variant_nested SELECT
@@ -134,7 +276,7 @@ suite("test_iceberg_variant_read",
TBLPROPERTIES (
'format-version'='3',
'write.format.default'='parquet',
- 'write.parquet.shred-variants'='true',
+ 'write.parquet.shred-variants'='false',
'write.parquet.variant-inference-buffer-size'='100'
);
INSERT INTO demo.${dbName}.variant_signed_selector
@@ -156,10 +298,105 @@ suite("test_iceberg_variant_read",
INSERT INTO demo.${dbName}.variant_write_guard VALUES (1);
"""
+ List<List<Object>> multiFileDataFiles = spark_iceberg """
+ SELECT COUNT(*) FROM demo.${dbName}.variant_multi_file.files WHERE
content = 0
+ """
+ assertEquals(1, multiFileDataFiles.size())
+ assertTrue(Long.parseLong(multiFileDataFiles[0][0].toString()) > 1,
+ "The parallel Variant fixture must contain multiple data files")
+
+ List<List<Object>> multiRowGroupFiles = spark_iceberg """
+ SELECT COUNT(*) FROM demo.${dbName}.variant_multi_row_group.files
WHERE content = 0
+ """
+ assertEquals(1, multiRowGroupFiles.size())
+ assertEquals("1", multiRowGroupFiles[0][0].toString(),
+ "The multi-row-group fixture must contain exactly one data file")
+
+ String equalityDeleteBaseSnapshot =
latestSnapshotId("variant_equality_delete")
+ String equalityDeleteJava = '''
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.Parquet;
+
+public class AppendVariantEqualityDelete {
+ public static void main(String[] args) throws Exception {
+ Map<String, String> props = new HashMap<>();
+ props.put("type", "rest");
+ props.put("uri", "http://rest:8181";);
+ props.put("warehouse", "s3://warehouse/wh/");
+ props.put("io-impl", "org.apache.iceberg.aws.s3.S3FileIO");
+ props.put("s3.endpoint", "http://minio:9000";);
+ props.put("s3.path-style-access", "true");
+ props.put("s3.region", "us-east-1");
+ Catalog catalog = CatalogUtil.buildIcebergCatalog("demo", props, null);
+ Table table = catalog.loadTable(TableIdentifier.of(args[0], args[1]));
+ Schema equalitySchema = table.schema().select("id");
+ int fieldId = table.schema().findField("id").fieldId();
+ OutputFile output = table.io().newOutputFile(
+ table.location() + "/data/variant-equality-delete-" +
+ System.currentTimeMillis() + ".parquet");
+ EqualityDeleteWriter<Record> writer = Parquet.writeDeletes(output)
+ .forTable(table)
+ .rowSchema(equalitySchema)
+ .withSpec(PartitionSpec.unpartitioned())
+ .createWriterFunc(GenericParquetWriter::create)
+ .equalityFieldIds(fieldId)
+ .overwrite()
+ .buildEqualityWriter();
+ GenericRecord record = GenericRecord.create(equalitySchema);
+ record.setField("id", Integer.valueOf(args[2]));
+ writer.write(record);
+ writer.close();
+ DeleteFile deleteFile = writer.toDeleteFile();
+ table.newRowDelta().addDeletes(deleteFile).commit();
+ }
+}
+'''
+ String encodedEqualityDeleteJava =
+ equalityDeleteJava.getBytes("UTF-8").encodeBase64().toString()
+ runInSparkContainer(
+ "echo ${encodedEqualityDeleteJava} | base64 -d " +
+ ">/tmp/AppendVariantEqualityDelete.java && " +
+ "javac -cp \"/opt/spark/jars/*\" " +
+ "/tmp/AppendVariantEqualityDelete.java && " +
+ "java -cp \"/tmp:/opt/spark/jars/*\"
AppendVariantEqualityDelete " +
+ "${dbName} variant_equality_delete 2")
+
String writeGuardSourceSnapshot = latestSnapshotId("variant_write_guard")
+ String deletionVectorBaseSnapshot =
latestSnapshotId("variant_deletion_vector")
spark_iceberg """
ALTER TABLE demo.${dbName}.variant_write_guard ADD COLUMN payload
VARIANT
"""
+ spark_iceberg """
+ DELETE FROM demo.${dbName}.variant_deletion_vector WHERE id % 2 = 1
+ """
+ List<List<Object>> deletionVectorFiles = spark_iceberg """
+ SELECT file_format, content_offset, content_size_in_bytes
+ FROM demo.${dbName}.variant_deletion_vector.files
+ WHERE content = 1
+ """
+ assertFalse(deletionVectorFiles.isEmpty(),
+ "The Variant deletion fixture must expose a live delete file")
+ deletionVectorFiles.each { List<Object> deleteFile ->
+ assertEquals("PUFFIN", deleteFile[0].toString().toUpperCase(),
+ "The format-v3 Variant fixture must use PUFFIN deletion
vectors")
+ assertTrue(Long.parseLong(deleteFile[1].toString()) >= 0,
+ "A PUFFIN deletion vector must expose its content offset")
+ assertTrue(Long.parseLong(deleteFile[2].toString()) > 0,
+ "A PUFFIN deletion vector must expose its content size")
+ }
// Register a stable Iceberg metadata fixture so the page-pruning case
always uses a
// standards-compliant shredded Variant file, independent of the Spark
writer version.
@@ -175,6 +412,21 @@ suite("test_iceberg_variant_read",
metadata_file =>
's3a://warehouse/wh/${dbName}/variant_page_pruning/metadata/${shreddedMetadataName}')
"""
+ String shreddedOnlySnapshot = latestSnapshotId("variant_page_pruning")
+ spark_iceberg_multi """
+ ALTER TABLE demo.${dbName}.variant_page_pruning SET TBLPROPERTIES (
+ 'read.parquet.vectorization.enabled'='false',
+ 'write.delete.mode'='merge-on-read'
+ );
+ INSERT INTO demo.${dbName}.variant_page_pruning VALUES
+ (5000, parse_json('{"n":5000,"padding":"mixed-unshredded"}'));
+ """
+ String mixedBeforeDeleteSnapshot = latestSnapshotId("variant_page_pruning")
+ // One deletion vector targets the shredded fixture and another targets
the appended
+ // unshredded file, forcing both physical states through the same scan and
delete alignment.
+ spark_iceberg """
+ DELETE FROM demo.${dbName}.variant_page_pruning WHERE id IN (4095,
5000)
+ """
sql """drop catalog if exists ${catalogName}"""
sql """
@@ -198,27 +450,61 @@ suite("test_iceberg_variant_read",
sql """set profile_level=2"""
def profileAction = new ProfileAction(context)
- def getProfileByToken = { String token ->
- for (int retry = 0; retry < 20; ++retry) {
- List profileData = profileAction.getProfileList()
- for (final def profileItem in profileData) {
- if (profileItem["Sql Statement"].toString().contains(token)) {
- return profileAction.getProfile(profileItem["Profile
ID"].toString())
- }
+ def mergedProfile = { String profile ->
+ if (!profile.contains("MergedProfile:")) {
+ return profile
+ }
+ String merged = profile.substring(profile.indexOf("MergedProfile:"))
+ int end = merged.length()
+ ["DetailProfile(", "Execution Profile:", "Appendix:"].each { String
sectionName ->
+ int sectionIndex = merged.indexOf(sectionName)
+ if (sectionIndex > 0) {
+ end = Math.min(end, sectionIndex)
}
- Thread.sleep(500)
}
- throw new IllegalStateException("Missing profile for token: " + token)
+ return merged.substring(0, end)
}
def counterSum = { String profile, String counterName ->
- Pattern pattern = Pattern.compile(Pattern.quote(counterName) +
":\\s*([0-9,]+)")
- Matcher matcher = pattern.matcher(profile)
+ Pattern pattern = Pattern.compile("(?m)^\\s*(?:-\\s*)?" +
+ Pattern.quote(counterName) + ":\\s+([^\\n]+)")
+ Matcher matcher = pattern.matcher(mergedProfile(profile))
long sum = 0
while (matcher.find()) {
- sum += Long.parseLong(matcher.group(1).replace(",", ""))
+ String valueText = matcher.group(1)
+ // Merged counters may be human-readable; the parenthesized value
is the exact sum.
+ Matcher exact =
Pattern.compile("\\(([0-9,]+)\\)").matcher(valueText)
+ Matcher number = Pattern.compile("([0-9,]+)").matcher(valueText)
+ if (exact.find()) {
+ sum += Long.parseLong(exact.group(1).replace(",", ""))
+ } else if (number.find()) {
+ sum += Long.parseLong(number.group(1).replace(",", ""))
+ }
}
return sum
}
+ def profileInfoValues = { String profile, String infoName ->
+ Pattern pattern = Pattern.compile(
+ Pattern.quote(infoName) + ":\\s*\\[([^\\]]*)\\]")
+ Matcher matcher = pattern.matcher(profile)
+ if (!matcher.find()) {
+ return []
+ }
+ return matcher.group(1).split(",").collect { String value ->
value.trim() }
+ .findAll { String value -> !value.isEmpty() }
+ .collect { String value -> Long.parseLong(value.replace(",",
"")) }
+ }
+ def getProfileByToken = { String token, List<String> positiveCounters = []
->
+ String lastProfile = profileAction.getProfileBySql(token,
positiveCounters)
+ if (positiveCounters.every { String counter -> counterSum(lastProfile,
counter) > 0 }) {
+ return lastProfile
+ }
+ return profileAction.waitProfile({
+ lastProfile = profileAction.getProfileBySql(token,
positiveCounters)
+ return positiveCounters.every {
+ String counter -> counterSum(lastProfile, counter) > 0
+ } ? lastProfile : ""
+ }, [], "Completed profile with positive counters ${positiveCounters}
for ${token}")
+ }
String evolutionInitial = latestSnapshotId("variant_evolution")
sql """ALTER TABLE variant_evolution CREATE TAG variant_initial"""
@@ -306,6 +592,18 @@ suite("test_iceberg_variant_read",
ORDER BY id
"""
+ order_qt_variant_root_array_projection """
+ SELECT id,
+ v IS NULL,
+ CAST(v AS STRING),
+ CAST(v[1] AS STRING),
+ CAST(v[2] AS INT),
+ CAST(v[3]['x'] AS INT),
+ CAST(v[4][2] AS INT)
+ FROM variant_root_arrays
+ ORDER BY id
+ """
+
order_qt_variant_path_expressions """
SELECT id,
UPPER(CAST(v['name'] AS STRING)),
@@ -327,8 +625,8 @@ suite("test_iceberg_variant_read",
ORDER BY id
"""
- // The first INSERT is unshredded while the second is shredded. Keep both
small files on one
- // scanner so their complete and leaf-only physical states must be
projected before batching.
+ // Keep the independent Spark writes on one scanner to exercise metadata
dictionaries and
+ // complete Variant state transitions across file boundaries before
batching.
sql "set parallel_pipeline_task_num=1"
sql "set max_file_scanners_concurrency=1"
order_qt_variant_cross_file_leaf_projection """
@@ -337,44 +635,218 @@ suite("test_iceberg_variant_read",
ORDER BY id
"""
- // Keep the root Variant as output while the implicit scalar comparison
drives the shredded
- // typed_value statistics/page-index path.
- order_qt_variant_implicit_shredded_filter """
+ order_qt_variant_multi_file_serial """
+ SELECT id,
+ CAST(v['shared'] AS INT),
+ CAST(v['a'] AS INT),
+ CAST(v['b'] AS INT),
+ CAST(v['new_field']['k'] AS INT),
+ CAST(v AS STRING)
+ FROM variant_multi_file
+ WHERE v['shared'] >= 20
+ ORDER BY id
+ """
+ sql "set parallel_pipeline_task_num=4"
+ sql "set max_file_scanners_concurrency=8"
+ sql "set min_file_scanners_concurrency=4"
+ order_qt_variant_multi_file_parallel """
+ SELECT id,
+ CAST(v['shared'] AS INT),
+ CAST(v['a'] AS INT),
+ CAST(v['b'] AS INT),
+ CAST(v['new_field']['k'] AS INT),
+ CAST(v AS STRING)
+ FROM variant_multi_file
+ WHERE v['shared'] >= 20
+ ORDER BY id
+ """
+ String parallelScanToken =
+ "iceberg_variant_parallel_scan_" + UUID.randomUUID().toString()
+ List<List<Object>> parallelScanRows = sql """
+ SELECT '${parallelScanToken}', id,
+ CAST(v['shared'] AS INT),
+ CAST(v['a'] AS INT),
+ CAST(v['b'] AS INT),
+ CAST(v['new_field']['k'] AS INT),
+ CAST(v AS STRING)
+ FROM variant_multi_file
+ WHERE v['shared'] >= 20
+ ORDER BY id
+ """
+ assertEquals(4, parallelScanRows.size(),
+ "The parallel Variant query must read rows from multiple data
files")
+ String parallelScanProfile = profileAction.getProfileBySql(
+ parallelScanToken, ["PerScannerRowsRead"])
+ if (profileInfoValues(parallelScanProfile, "PerScannerRowsRead")
+ .count { long rows -> rows > 0 } <= 1) {
+ parallelScanProfile = profileAction.waitProfile({
+ String profile = profileAction.getProfileBySql(
+ parallelScanToken, ["PerScannerRowsRead"])
+ return profileInfoValues(profile, "PerScannerRowsRead")
+ .count { long rows -> rows > 0 } > 1 ? profile : ""
+ }, [], "Completed parallel Variant profile with multiple non-empty
scanners")
+ }
+ assertTrue(profileInfoValues(parallelScanProfile, "PerScannerRowsRead")
+ .count { long rows -> rows > 0 } > 1,
+ "The parallel Variant query did not use multiple non-empty
scanners")
+ sql "set min_file_scanners_concurrency=1"
+
+ order_qt_variant_type_matrix """
+ SELECT CAST(v['bool_value'] AS BOOLEAN),
+ CAST(v['tiny_value'] AS TINYINT),
+ CAST(v['small_value'] AS SMALLINT),
+ CAST(v['int_value'] AS INT),
+ CAST(v['big_value'] AS BIGINT),
+ ISNAN(CAST(v['float_value'] AS FLOAT)),
+ ISINF(CAST(v['double_value'] AS DOUBLE)),
+ CAST(v['decimal_value'] AS DECIMAL(20, 4)),
+ CAST(v['date_value'] AS DATE),
+ CAST(v['timestamp_value'] AS DATETIMEV2(6)),
+ CAST(v['binary_value'] AS STRING),
+ v['null_value'] IS NULL
+ FROM variant_type_matrix
+ """
+
+ String multiRowGroupColdToken =
+ "iceberg_variant_multi_row_group_cold_" +
UUID.randomUUID().toString()
+ sql """
+ SELECT '${multiRowGroupColdToken}', COUNT(*), MIN(id), MAX(id)
+ FROM variant_multi_row_group
+ WHERE CAST(v['n'] AS INT) >= 8000
+ """
+ String multiRowGroupColdProfile = getProfileByToken(multiRowGroupColdToken,
+ ["RowGroupsTotalNum", "VariantDirectLeafPathMisses",
"VariantReconstructedRows",
+ "FilteredRowsByLazyRead"]).toString()
+ assertTrue(counterSum(multiRowGroupColdProfile, "RowGroupsTotalNum") > 1,
+ "The generated Variant file did not contain multiple Parquet
row groups")
+ assertTrue(counterSum(multiRowGroupColdProfile,
"VariantDirectLeafPathMisses") > 0,
+ "The unshredded scan did not record its direct-leaf fallback")
+ assertTrue(counterSum(multiRowGroupColdProfile,
"VariantReconstructedRows") > 0,
+ "The unshredded scan did not reconstruct Variant rows")
+ assertTrue(counterSum(multiRowGroupColdProfile, "FilteredRowsByLazyRead")
> 0,
+ "The unshredded Variant predicate did not defer non-predicate
columns")
+ String multiRowGroupWarmToken =
+ "iceberg_variant_multi_row_group_warm_" +
UUID.randomUUID().toString()
+ sql """
+ SELECT '${multiRowGroupWarmToken}', COUNT(*), MIN(id), MAX(id)
+ FROM variant_multi_row_group
+ WHERE CAST(v['n'] AS INT) >= 8000
+ """
+ String multiRowGroupWarmProfile = getProfileByToken(multiRowGroupWarmToken,
+ ["VariantDirectLeafPathMisses"]).toString()
+ assertTrue(counterSum(multiRowGroupWarmProfile,
"VariantDirectLeafPathMisses") > 0,
+ "The warm unshredded scan did not preserve its direct-leaf
fallback")
+ qt_variant_multi_row_group_result """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_multi_row_group
+ WHERE CAST(v['n'] AS INT) >= 8000
+ """
+
+ qt_variant_deletion_vector_current """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_deletion_vector
+ WHERE v['n'] >= 0
+ """
+ qt_variant_deletion_vector_before_delete """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_deletion_vector FOR VERSION AS OF
${deletionVectorBaseSnapshot}
+ WHERE v['n'] >= 0
+ """
+ order_qt_variant_equality_delete_current """
+ SELECT id, CAST(v['n'] AS INT), CAST(v['label'] AS STRING), CAST(v AS
STRING)
+ FROM variant_equality_delete
+ WHERE v['n'] >= 0
+ ORDER BY id
+ """
+ order_qt_variant_equality_delete_before_delete """
+ SELECT id, CAST(v['n'] AS INT), CAST(v['label'] AS STRING), CAST(v AS
STRING)
+ FROM variant_equality_delete FOR VERSION AS OF
${equalityDeleteBaseSnapshot}
+ WHERE v['n'] >= 0
+ ORDER BY id
+ """
+
+ // Keep the root Variant as output while the scalar comparison exercises
the fallback path for
+ // the unshredded Spark files.
+ order_qt_variant_implicit_filter """
SELECT id, CAST(v AS STRING)
FROM variant_values
WHERE v['n'] > 35
ORDER BY id
"""
- // The query projects the root Variant, while the predicate uses
typed_value page metadata.
+ qt_variant_shredded_only_time_travel """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_page_pruning FOR VERSION AS OF ${shreddedOnlySnapshot}
+ WHERE CAST(v['n'] AS INT) > 3000
+ """
+ qt_variant_mixed_before_delete """
+ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT))
+ FROM variant_page_pruning FOR VERSION AS OF
${mixedBeforeDeleteSnapshot}
+ WHERE CAST(v['n'] AS INT) > 3000
+ """
+
+ // The complete Variant is the only scanned output column outside the
predicate. A positive
+ // lazy-read count therefore proves Variant output deferral rather than
deferral of an id
+ // sibling, while the row relationship proves reconstruction happens after
filtering.
+ String lazyVariantToken =
+ "iceberg_variant_lazy_materialization_" +
UUID.randomUUID().toString()
+ List<List<Object>> lazyVariantRows = sql """
+ SELECT '${lazyVariantToken}', CAST(v AS STRING)
+ FROM variant_page_pruning FOR VERSION AS OF ${shreddedOnlySnapshot}
+ WHERE CAST(v['n'] AS INT) > 3000
+ """
+ String lazyVariantProfile = getProfileByToken(lazyVariantToken,
+ ["VariantDirectLeafRows", "VariantReconstructedRows",
+ "FilteredRowsByLazyRead"]).toString()
+ long reconstructedVariantRows =
+ counterSum(lazyVariantProfile, "VariantReconstructedRows")
+ assertEquals((long) lazyVariantRows.size(), reconstructedVariantRows,
+ "Complete Variant reconstruction must be limited to selected
output rows")
+ assertTrue(counterSum(lazyVariantProfile, "VariantDirectLeafRows") >
+ reconstructedVariantRows,
+ "Variant output was not deferred until after its shredded-leaf
predicate")
+ assertTrue(counterSum(lazyVariantProfile, "FilteredRowsByLazyRead") > 0,
+ "The shredded predicate did not defer complete Variant output")
+
+ // The query projects the complete Variant while its predicate reads the
shredded typed leaf.
+ // The appended unshredded file must fall back independently in the same
scan.
String pagePruningToken = "iceberg_variant_page_pruning_" +
UUID.randomUUID().toString()
sql """
SELECT '${pagePruningToken}', id, CAST(v AS STRING)
FROM variant_page_pruning
- WHERE v['n'] > 3000
+ WHERE CAST(v['n'] AS INT) > 3000
ORDER BY id
"""
- String pagePruningProfile = getProfileByToken(pagePruningToken).toString()
+ String pagePruningProfile = getProfileByToken(pagePruningToken,
+ ["FilteredRowsByPage", "VariantLeafProjections",
"VariantDirectLeafPathMisses",
+ "VariantDirectLeafRows", "VariantReconstructedRows"]).toString()
assertTrue(counterSum(pagePruningProfile, "FilteredRowsByPage") > 0,
"Shredded Variant typed_value did not filter any Parquet page")
// The predicate_access_paths contract keeps the typed leaf eager while
the complete Variant
// root is read through the independent deferred-output projection.
assertTrue(counterSum(pagePruningProfile, "VariantLeafProjections") > 0,
"A root Variant output query did not retain its typed predicate
leaf projection")
+ assertTrue(counterSum(pagePruningProfile, "VariantDirectLeafPathMisses") >
0,
+ "The mixed scan did not fall back for its unshredded Variant
file")
+ assertTrue(counterSum(pagePruningProfile, "VariantDirectLeafRows") > 0,
+ "The mixed scan did not evaluate rows from the shredded typed
leaf")
+ assertTrue(counterSum(pagePruningProfile, "VariantReconstructedRows") > 0,
+ "The mixed scan did not reconstruct complete Variant output")
String leafProjectionToken =
"iceberg_variant_leaf_projection_" + UUID.randomUUID().toString()
sql """
SELECT '${leafProjectionToken}', COUNT(*)
FROM variant_page_pruning
- WHERE v['n'] > 3000
+ WHERE CAST(v['n'] AS INT) > 3000
"""
- String leafProjectionProfile =
getProfileByToken(leafProjectionToken).toString()
+ String leafProjectionProfile = getProfileByToken(leafProjectionToken,
+ ["VariantLeafProjections"]).toString()
assertTrue(counterSum(leafProjectionProfile, "VariantLeafProjections") > 0,
"Variant typed predicate did not retain a physical leaf
projection")
qt_variant_page_pruning_result """
SELECT COUNT(*), MIN(id), MAX(id)
FROM variant_page_pruning
- WHERE v['n'] > 3000
+ WHERE CAST(v['n'] AS INT) > 3000
"""
order_qt_variant_aggregate """
@@ -521,6 +993,7 @@ suite("test_iceberg_variant_read",
// A delete-only MERGE emits only position deletes. It must remain
available even though
// update/insert actions would route the unchanged Variant through the
unsupported data writer.
+ String beforePositionDeleteSnapshot = latestSnapshotId("variant_values")
sql """
MERGE INTO variant_values t
USING (SELECT 11 AS id) s
@@ -528,6 +1001,31 @@ suite("test_iceberg_variant_read",
WHEN MATCHED THEN DELETE
"""
qt_variant_delete_only_merge "SELECT COUNT(*) FROM variant_values WHERE id
= 11"
+ order_qt_variant_position_delete_alignment """
+ SELECT id, CAST(v['name'] AS STRING), CAST(v['n'] AS INT), CAST(v AS
STRING)
+ FROM variant_values
+ WHERE v['n'] >= 40
+ ORDER BY id
+ """
+ order_qt_variant_before_position_delete """
+ SELECT id, CAST(v['name'] AS STRING), CAST(v['n'] AS INT), CAST(v AS
STRING)
+ FROM variant_values FOR VERSION AS OF ${beforePositionDeleteSnapshot}
+ WHERE v['n'] >= 40
+ ORDER BY id
+ """
+ String positionDeleteToken =
+ "iceberg_variant_position_delete_" + UUID.randomUUID().toString()
+ sql """
+ SELECT '${positionDeleteToken}', COUNT(*)
+ FROM variant_values
+ WHERE v['n'] >= 40
+ """
+ String positionDeleteProfile = getProfileByToken(positionDeleteToken,
+ ["VariantDirectLeafPathMisses",
"VariantReconstructedRows"]).toString()
+ assertTrue(counterSum(positionDeleteProfile,
"VariantDirectLeafPathMisses") > 0,
+ "Position-delete filtering did not preserve the unshredded
Variant fallback")
+ assertTrue(counterSum(positionDeleteProfile, "VariantReconstructedRows") >
0,
+ "Position-delete filtering did not reconstruct its Variant
rows")
// Files written before the Variant field existed have no physical Variant
payload. Schema
// evolution must synthesize NULL instead of rejecting their non-Parquet
file format.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
