From 97bb75eec315272b8a2cb5f35eba5011cabee8b1 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 4 Aug 2026 14:33:36 +0800 Subject: [PATCH 1/2] test: expand Iceberg Variant read coverage --- .../parquet/variant_column_reader_test.cpp | 718 ++++++++++++++++++ .../iceberg/test_iceberg_variant_read.out | 59 +- .../iceberg/test_iceberg_variant_read.groovy | 444 ++++++++++- 3 files changed, 1196 insertions(+), 25 deletions(-) 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 40c18635aad339..5bf2d33c10f1af 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 #include +#include +#include +#include +#include #include #include +#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(); + 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(); @@ -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(); + value->name = "value"; + value->kind = ParquetColumnSchemaKind::PRIMITIVE; + value->type = make_nullable(std::make_shared()); + element->children.insert(element->children.begin(), std::move(value)); + return schema; +} + MutableColumnPtr shredded_int64_physical(const std::vector& values) { const std::array ignored {0}; const StringRef metadata(VARIANT_EMPTY_METADATA.data(), VARIANT_EMPTY_METADATA.size()); @@ -157,6 +197,20 @@ MutableColumnPtr shredded_int64_physical(const std::vector& 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 ignored {0}; + const StringRef metadata(VARIANT_EMPTY_METADATA.data(), VARIANT_EMPTY_METADATA.size()); + MutableColumns fields; + fields.push_back(nullable_strings(std::vector(rows, metadata), + std::vector(rows, 0))); + fields.push_back(nullable_strings(std::vector(rows, {ignored.data(), 0}), + std::vector(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& values, const IColumn** decoded_leaf = nullptr) { auto integers = ColumnInt64::create(); @@ -182,6 +236,37 @@ MutableColumnPtr projected_shredded_object_physical(const std::vector& 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& values, + const std::vector& 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& first, const std::vector& 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())->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( + assert_cast(*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& values, + const std::vector& 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 +MutableColumnPtr nullable_fixed(std::initializer_list values, + std::initializer_list 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 +MutableColumnPtr nullable_decimal(uint32_t scale, std::initializer_list 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& verify) { + auto output = make_nullable(std::make_shared())->create_column(); + const Status status = materialize_variant_rows( + schema, shredded_primitive_physical(std::move(typed)), output); + ASSERT_TRUE(status.ok()) << status; + verify(assert_cast( + assert_cast(*output).get_nested_column())); + }; + + decode(shredded_primitive_schema(std::make_shared()), + nullable_fixed({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(), + nullable_fixed( + {std::numeric_limits::min(), std::numeric_limits::max()}, {0, 0}), + 8, std::numeric_limits::min(), std::numeric_limits::max()); + verify_integer( + std::make_shared(), + nullable_fixed( + {std::numeric_limits::min(), std::numeric_limits::max()}, {0, 0}), + 16, std::numeric_limits::min(), std::numeric_limits::max()); + verify_integer( + std::make_shared(), + nullable_fixed( + {std::numeric_limits::min(), std::numeric_limits::max()}, {0, 0}), + 32, std::numeric_limits::min(), std::numeric_limits::max()); + verify_integer( + std::make_shared(), + nullable_fixed( + {std::numeric_limits::min(), std::numeric_limits::max()}, {0, 0}), + 64, std::numeric_limits::min(), std::numeric_limits::max()); + + decode(shredded_primitive_schema(std::make_shared()), + nullable_fixed({std::numeric_limits::quiet_NaN(), + std::numeric_limits::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()), + nullable_fixed({-std::numeric_limits::infinity(), 1.25}, + {0, 0}), + [](const auto& values) { + EXPECT_EQ(values.get_value_ref(0).get_double(), + -std::numeric_limits::infinity()); + EXPECT_EQ(values.get_value_ref(1).get_double(), 1.25); + }); + + { + auto schema = shredded_primitive_schema(std::make_shared(9, 2)); + schema.children.back()->type_descriptor.decimal_precision = 9; + schema.children.back()->type_descriptor.decimal_scale = 2; + decode(std::move(schema), + nullable_decimal(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(18, 3)); + schema.children.back()->type_descriptor.decimal_precision = 18; + schema.children.back()->type_descriptor.decimal_scale = 3; + decode(std::move(schema), + nullable_decimal( + 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(38, 4)); + schema.children.back()->type_descriptor.decimal_precision = 38; + schema.children.back()->type_descriptor.decimal_scale = 4; + decode(std::move(schema), + nullable_decimal( + 4, {Decimal128V3 {static_cast(1234567890123456789LL)}}), + [](const auto& values) { + EXPECT_EQ(values.get_value_ref(0).get_decimal(), + (VariantDecimal {1234567890123456789LL, 4, 16})); + }); + } + + const auto date = DateV2Value::create_from_olap_date( + (static_cast(1970) << 9) | (static_cast(1) << 5) | 2); + decode(shredded_primitive_schema(std::make_shared()), + nullable_fixed>({date}, {0}), + [](const auto& values) { EXPECT_EQ(values.get_value_ref(0).get_date(), 1); }); + + auto datetime = DateV2Value::create_from_olap_datetime(19700101000001ULL); + datetime.set_microsecond(234567); + { + auto schema = shredded_primitive_schema(std::make_shared(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>({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(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({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 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()); + 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(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()); + 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())->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( + assert_cast(*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 ignored {0}; const StringRef metadata(VARIANT_EMPTY_METADATA.data(), VARIANT_EMPTY_METADATA.size()); @@ -776,6 +1109,273 @@ TEST(VariantColumnReaderTest, MaterializesShreddedArrayElements) { EXPECT_EQ(value.array_at(1).get_int(), 4); } +TEST(VariantColumnReaderTest, RejectsCorruptShreddedWrappersWithoutCrashing) { + const std::array int_seven { + static_cast(static_cast(VariantPrimitiveId::INT8) + << VARIANT_VALUE_HEADER_SHIFT), + 7}; + const std::array invalid_value {static_cast(0xff)}; + const std::array 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())->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( + assert_cast(*output).get_nested_column()); + EXPECT_EQ(variants.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 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 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())->create_column(); + ASSERT_TRUE(materialize_variant_rows(shredded_mixed_array_schema(), *physical, output).ok()); + const auto& nullable = assert_cast(*output); + EXPECT_EQ(nullable.get_null_map_data(), (NullMap {0, 0, 0, 1})); + const auto& variants = assert_cast(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 int_seven { static_cast(static_cast(VariantPrimitiveId::INT8) @@ -817,6 +1417,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(); + 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(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( + DataTypes {make_nullable(std::make_shared())}, + Strings {"v"}) + ->create_column(); + ASSERT_TRUE(materialize_variant_columns(plan, *physical, output).ok()); + const auto& variants = assert_cast( + assert_cast( + assert_cast(*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(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( + make_nullable(std::make_shared())) + ->create_column(); + ASSERT_TRUE(materialize_variant_columns(plan, *physical, output).ok()); + const auto& variants = assert_cast( + assert_cast( + assert_cast(*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(); + key_schema->name = "key"; + key_schema->kind = ParquetColumnSchemaKind::PRIMITIVE; + key_schema->type = std::make_shared(); + root_schema.children.push_back(std::move(key_schema)); + root_schema.children.push_back( + std::make_unique(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(std::make_shared(), + make_nullable(std::make_shared())) + ->create_column(); + ASSERT_TRUE(materialize_variant_columns(plan, *physical, output).ok()); + const auto& variants = assert_cast( + assert_cast( + assert_cast(*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())->create_column(); + ASSERT_TRUE(materialize_variant_columns(plan, projected_shredded_object_physical({17}), output) + .ok()); + const auto& variants = assert_cast( + assert_cast(*output).get_nested_column()); + EXPECT_THROW((void)variants.get_value_ref(0), Exception); +} + TEST(VariantColumnReaderTest, AlignsNestedPrimitiveNullabilityAroundVariant) { const std::array int_seven { static_cast(static_cast(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 a7c41169d8dcb3..88408fc6dea795 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 7b3bc3e833d165..9e147b26697bfc 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> rows = spark_iceberg """ @@ -100,6 +133,115 @@ suite("test_iceberg_variant_read", (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'='true', + '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'='true', + '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'='true'); + 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'='true', + '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'='true', + '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'='true', + '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'='true', + '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; @@ -156,10 +298,83 @@ suite("test_iceberg_variant_read", INSERT INTO demo.${dbName}.variant_write_guard VALUES (1); """ + List> 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 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 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 + """ // 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 +390,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,18 +428,6 @@ 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()) - } - } - Thread.sleep(500) - } - throw new IllegalStateException("Missing profile for token: " + token) - } def counterSum = { String profile, String counterName -> Pattern pattern = Pattern.compile(Pattern.quote(counterName) + ":\\s*([0-9,]+)") Matcher matcher = pattern.matcher(profile) @@ -219,6 +437,25 @@ suite("test_iceberg_variant_read", } return sum } + def getProfileByToken = { String token, List positiveCounters = [] -> + String lastProfile = "" + for (int retry = 0; retry < 20; ++retry) { + List profileData = profileAction.getProfileList() + for (final def profileItem in profileData) { + if (profileItem["Sql Statement"].toString().contains(token)) { + lastProfile = profileAction.getProfile( + profileItem["Profile ID"].toString()).toString() + if (positiveCounters.every { counterSum(lastProfile, it) > 0 }) { + return lastProfile + } + } + } + Thread.sleep(500) + } + throw new IllegalStateException( + "Profile did not expose positive counters ${positiveCounters} for token ${token}: " + + lastProfile) + } String evolutionInitial = latestSnapshotId("variant_evolution") sql """ALTER TABLE variant_evolution CREATE TAG variant_initial""" @@ -306,6 +543,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 +576,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 +586,167 @@ 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" + 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 + """ + + 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['keep'] = true + """ + 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 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", + "FilteredRowsByLazyRead"]).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") + assertTrue(counterSum(pagePruningProfile, "FilteredRowsByLazyRead") > 0, + "The mixed Variant scan did not delay output materialization") 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 +893,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 +901,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. From 8877f9bb6f018c51a49568ab7cc79a117a34c84f Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 4 Aug 2026 16:47:16 +0800 Subject: [PATCH 2/2] fix: preserve nested Variant append atomicity --- .../parquet/reader/variant_column_reader.cpp | 101 +++++++--- .../parquet/variant_column_reader_test.cpp | 176 ++++++++++++++++++ .../iceberg/test_iceberg_variant_read.groovy | 170 +++++++++++++---- 3 files changed, 390 insertions(+), 57 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 26e98e5128b950..43ff808cf5b7a0 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 #include #include +#include #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(output)) { - if (const auto* converted_nullable = check_and_get_column(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(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 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 5bf2d33c10f1af..24ace262faba37 100644 --- a/be/test/format_v2/parquet/variant_column_reader_test.cpp +++ b/be/test/format_v2/parquet/variant_column_reader_test.cpp @@ -1300,6 +1300,182 @@ TEST(VariantColumnReaderTest, ImmediateCorruptionLeavesDestinationUnchanged) { EXPECT_EQ(variants.get_value_ref(0).get_int(), 7); } +TEST(VariantColumnReaderTest, LazyNestedCorruptionLeavesDestinationUnchanged) { + const std::array invalid_value {static_cast(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(); + schema->name = "label"; + schema->kind = ParquetColumnSchemaKind::PRIMITIVE; + schema->type = make_nullable(std::make_shared()); + return schema; + }; + auto make_plan = [](const ParquetColumnSchema& root) { + auto build = [&](auto&& self, const ParquetColumnSchema* schema) + -> std::unique_ptr { + auto node = std::make_unique(); + 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(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( + DataTypes {make_nullable(std::make_shared()), + make_nullable(std::make_shared())}, + 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(*output); + const auto& label = assert_cast(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(structure.get_column(1)); + EXPECT_EQ(payload.size(), 1); + EXPECT_EQ(payload.get_null_map_data(), (NullMap {0})); + EXPECT_EQ(assert_cast(payload.get_nested_column()) + .get_value_ref(0) + .get_int(), + 7); + } + + { + auto output = std::make_shared(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(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(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(*output); + EXPECT_EQ(array.get_offsets(), (ColumnArray::Offsets64 {1})); + const auto& element = assert_cast(array.get_data()); + EXPECT_EQ(element.get_null_map_data(), (NullMap {0})); + const auto& structure = assert_cast(element.get_nested_column()); + const auto& label = assert_cast(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(structure.get_column(1)); + EXPECT_EQ(payload.size(), 1); + EXPECT_EQ(payload.get_null_map_data(), (NullMap {0})); + EXPECT_EQ(assert_cast(payload.get_nested_column()) + .get_value_ref(0) + .get_int(), + 7); + } + + { + auto output = + std::make_shared(make_nullable(std::make_shared()), + make_nullable(std::make_shared())) + ->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(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(*output); + EXPECT_EQ(map.get_offsets(), (ColumnArray::Offsets64 {1})); + const auto& keys = assert_cast(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(map.get_values()); + EXPECT_EQ(values.size(), 1); + EXPECT_EQ(values.get_null_map_data(), (NullMap {0})); + EXPECT_EQ(assert_cast(values.get_nested_column()) + .get_value_ref(0) + .get_int(), + 7); + } +} + TEST(VariantColumnReaderTest, MaterializesMixedRootArraysAndNullKinds) { VariantBatchBuilder residual_builder; { 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 9e147b26697bfc..7161574b962851 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 @@ -123,11 +123,11 @@ 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"}}')), @@ -138,7 +138,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_root_arrays VALUES @@ -158,7 +158,7 @@ suite("test_iceberg_variant_read", 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'='true', + 'write.parquet.shred-variants'='false', 'write.parquet.variant-inference-buffer-size'='1' ); INSERT INTO demo.${dbName}.variant_multi_file @@ -170,7 +170,7 @@ suite("test_iceberg_variant_read", 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'='true'); + ('write.parquet.shred-variants'='false'); INSERT INTO demo.${dbName}.variant_multi_file VALUES (5, parse_json('{"shared":50,"b":5,"new_field":{"k":500}}')); @@ -179,7 +179,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_type_matrix SELECT 1, to_variant_object(named_struct( @@ -202,7 +202,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', 'write.parquet.row-group-size-bytes'='4096' ); @@ -217,7 +217,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', 'write.delete.mode'='merge-on-read', 'read.parquet.vectorization.enabled'='false', @@ -234,7 +234,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_equality_delete VALUES @@ -254,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 @@ -276,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 @@ -298,6 +298,13 @@ suite("test_iceberg_variant_read", INSERT INTO demo.${dbName}.variant_write_guard VALUES (1); """ + List> 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> multiRowGroupFiles = spark_iceberg """ SELECT COUNT(*) FROM demo.${dbName}.variant_multi_row_group.files WHERE content = 0 """ @@ -375,6 +382,21 @@ public class AppendVariantEqualityDelete { spark_iceberg """ DELETE FROM demo.${dbName}.variant_deletion_vector WHERE id % 2 = 1 """ + List> 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 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. @@ -428,33 +450,60 @@ public class AppendVariantEqualityDelete { sql """set profile_level=2""" def profileAction = new ProfileAction(context) + 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) + } + } + 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 positiveCounters = [] -> - String lastProfile = "" - for (int retry = 0; retry < 20; ++retry) { - List profileData = profileAction.getProfileList() - for (final def profileItem in profileData) { - if (profileItem["Sql Statement"].toString().contains(token)) { - lastProfile = profileAction.getProfile( - profileItem["Profile ID"].toString()).toString() - if (positiveCounters.every { counterSum(lastProfile, it) > 0 }) { - return lastProfile - } - } - } - Thread.sleep(500) + String lastProfile = profileAction.getProfileBySql(token, positiveCounters) + if (positiveCounters.every { String counter -> counterSum(lastProfile, counter) > 0 }) { + return lastProfile } - throw new IllegalStateException( - "Profile did not expose positive counters ${positiveCounters} for token ${token}: " + - 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") @@ -599,6 +648,7 @@ public class AppendVariantEqualityDelete { """ 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), @@ -610,6 +660,36 @@ public class AppendVariantEqualityDelete { WHERE v['shared'] >= 20 ORDER BY id """ + String parallelScanToken = + "iceberg_variant_parallel_scan_" + UUID.randomUUID().toString() + List> 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), @@ -665,7 +745,7 @@ public class AppendVariantEqualityDelete { qt_variant_deletion_vector_current """ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT)) FROM variant_deletion_vector - WHERE v['keep'] = true + WHERE v['n'] >= 0 """ qt_variant_deletion_vector_before_delete """ SELECT COUNT(*), MIN(id), MAX(id), SUM(CAST(v['n'] AS BIGINT)) @@ -705,6 +785,29 @@ public class AppendVariantEqualityDelete { 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> 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() @@ -716,8 +819,7 @@ public class AppendVariantEqualityDelete { """ String pagePruningProfile = getProfileByToken(pagePruningToken, ["FilteredRowsByPage", "VariantLeafProjections", "VariantDirectLeafPathMisses", - "VariantDirectLeafRows", "VariantReconstructedRows", - "FilteredRowsByLazyRead"]).toString() + "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 @@ -730,8 +832,6 @@ public class AppendVariantEqualityDelete { "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") - assertTrue(counterSum(pagePruningProfile, "FilteredRowsByLazyRead") > 0, - "The mixed Variant scan did not delay output materialization") String leafProjectionToken = "iceberg_variant_leaf_projection_" + UUID.randomUUID().toString() sql """