feat: add Iceberg v3 variant Parquet support

.github/workflows/cpp-linter.yml

@@ -72,6 +72,8 @@ jobs:
         run: |
           mkdir build && cd build
           cmake .. -G Ninja -DCMAKE_EXPORT_COMPILE_COMMANDS=ON \
+            -DICEBERG_REQUIRE_ARROW_VARIANT=ON \
+            -DICEBERG_USE_BUNDLED_ARROW=ON \
             -DCMAKE_C_COMPILER_LAUNCHER=sccache \
             -DCMAKE_CXX_COMPILER_LAUNCHER=sccache \
             -DICEBERG_BUILD_SQL_CATALOG=ON \

.github/workflows/sanitizer_test.yml

@@ -63,6 +63,8 @@ jobs:
         run: |
           mkdir build && cd build
           cmake .. -G Ninja -DCMAKE_BUILD_TYPE=Debug -DICEBERG_ENABLE_ASAN=ON -DICEBERG_ENABLE_UBSAN=ON \
+            -DICEBERG_REQUIRE_ARROW_VARIANT=ON \
+            -DICEBERG_USE_BUNDLED_ARROW=ON \
             -DCMAKE_C_COMPILER_LAUNCHER=sccache -DCMAKE_CXX_COMPILER_LAUNCHER=sccache
           cmake --build . --verbose
       - name: Save sccache

CMakeLists.txt

@@ -53,6 +53,10 @@ option(ICEBERG_BUILD_BENCHMARKS "Build benchmarks" OFF)
 option(ICEBERG_BUILD_BUNDLE "Build the battery included library" ON)
 option(ICEBERG_ENCRYPTION
        "Build table encryption support (Parquet modular encryption, AES GCM streams)" OFF)
+option(ICEBERG_USE_BUNDLED_ARROW
+       "Use the pinned Arrow source instead of an installed Arrow" OFF)
+option(ICEBERG_REQUIRE_ARROW_VARIANT
+       "Require Arrow variant APIs when building the bundle" ON)
 option(ICEBERG_BUILD_REST "Build rest catalog client" ON)
 option(ICEBERG_BUILD_REST_INTEGRATION_TESTS "Build rest catalog integration tests" OFF)
 option(ICEBERG_BUILD_HIVE "Build hive (HMS) catalog client" OFF)

ci/scripts/build_iceberg.sh

@@ -168,6 +168,10 @@ if [[ -n "${ICEBERG_EXTRA_CMAKE_ARGS:-}" ]]; then
     CMAKE_ARGS+=("${EXTRA_CMAKE_ARGS[@]}")
 fi
 
+# CI must compile and run the complete variant tests, even if a runner happens
+# to discover an older system Arrow installation.
+CMAKE_ARGS+=("-DICEBERG_USE_BUNDLED_ARROW=ON" "-DICEBERG_REQUIRE_ARROW_VARIANT=ON")
+
 cmake "${CMAKE_ARGS[@]}" ${source_dir}
 
 cmake --build . --target install

cmake_modules/IcebergThirdpartyToolchain.cmake

@@ -89,22 +89,21 @@ endfunction()
 # ICEBERG_CPR_URL            - cpr tarball URL
 #
 # Example usage:
-#   export ICEBERG_ARROW_URL="https://your-mirror.com/apache-arrow-25.0.0.tar.gz"
+#   export ICEBERG_ARROW_URL="https://your-mirror.com/arrow-0f16412bb1c39611aeed7393eba09e637f4d86fa.tar.gz"
 #   cmake -S . -B build
 #
 
-set(ICEBERG_ARROW_BUILD_VERSION "25.0.0")
+# Temporary pin for apache/arrow#50252. Replace with an Apache Arrow release
+# containing the variant APIs once that work has merged and been released.
+set(ICEBERG_ARROW_BUILD_COMMIT "0f16412bb1c39611aeed7393eba09e637f4d86fa")
 set(ICEBERG_ARROW_BUILD_SHA256_CHECKSUM
-    "12afc2dc8137bdd4a68876cec939f664c9d55cfc7b75f55b45163ebb4e344d81")
+    "4e184655c31b4b28cf7d0cd1034f9b3891d5d861520d427a90726b28e52e72f9")
 
 if(DEFINED ENV{ICEBERG_ARROW_URL})
   set(ARROW_SOURCE_URL "$ENV{ICEBERG_ARROW_URL}")
 else()
   set(ARROW_SOURCE_URL
-      "https://www.apache.org/dyn/closer.lua?action=download&filename=/arrow/arrow-${ICEBERG_ARROW_BUILD_VERSION}/apache-arrow-${ICEBERG_ARROW_BUILD_VERSION}.tar.gz"
-      "https://downloads.apache.org/arrow/arrow-${ICEBERG_ARROW_BUILD_VERSION}/apache-arrow-${ICEBERG_ARROW_BUILD_VERSION}.tar.gz"
-      "https://archive.apache.org/dist/arrow/arrow-${ICEBERG_ARROW_BUILD_VERSION}/apache-arrow-${ICEBERG_ARROW_BUILD_VERSION}.tar.gz"
-  )
+      "https://codeload.github.com/HuaHuaY/arrow/tar.gz/${ICEBERG_ARROW_BUILD_COMMIT}")
 endif()
 
 set(ICEBERG_NANOARROW_BUILD_VERSION "0.9.0")
@@ -180,12 +179,67 @@ macro(resolve_openssl_for_encryption)
   find_package(OpenSSL REQUIRED)
 endmacro()
 
+function(iceberg_check_arrow_variant)
+  if(NOT ICEBERG_REQUIRE_ARROW_VARIANT)
+    return()
+  endif()
+
+  include(CheckCXXSourceCompiles)
+  # This also works before bundled Arrow has been built: compile the API probe
+  # without linking against its libraries.
+  set(CMAKE_TRY_COMPILE_TARGET_TYPE STATIC_LIBRARY)
+  set(CMAKE_REQUIRED_INCLUDES)
+  foreach(dependency Arrow Parquet)
+    if(dependency STREQUAL "Arrow")
+      set(component arrow)
+    else()
+      set(component parquet)
+    endif()
+    if(TARGET ${dependency}::${component}_static)
+      set(dependency_target ${dependency}::${component}_static)
+    else()
+      set(dependency_target ${dependency}::${component}_shared)
+    endif()
+    get_target_property(dependency_includes ${dependency_target}
+                        INTERFACE_INCLUDE_DIRECTORIES)
+    list(APPEND CMAKE_REQUIRED_INCLUDES ${dependency_includes})
+  endforeach()
+  # Recheck when an existing build switches Arrow installations or source pins.
+  unset(ICEBERG_ARROW_HAS_VARIANT CACHE)
+  check_cxx_source_compiles("#include <arrow/extension/parquet_variant.h>
+     #include <parquet/properties.h>
+     #include <parquet/variant/builder.h>
+     #include <parquet/variant/decoding.h>
+     #include <parquet/variant/shred.h>
+     #include <parquet/variant/unshred.h>
+     #include <parquet/variant/validate.h>
+     using Shred = decltype(&parquet::variant::ShredVariantArray);
+     using Unshred = decltype(&parquet::variant::UnshredVariantArray);
+     using Validate = decltype(&parquet::variant::ValidateVariants<true>);
+     using ReaderValidation = decltype(&parquet::ArrowReaderProperties::set_variant_validation_enabled);
+     using WriterValidation = decltype(&parquet::ArrowWriterProperties::Builder::set_variant_validation_enabled);
+     using IsShredded = decltype(&arrow::extension::VariantArray::is_shredded);
+     int main() { return 0; }"
+                            ICEBERG_ARROW_HAS_VARIANT)
+  if(NOT ICEBERG_ARROW_HAS_VARIANT)
+    message(FATAL_ERROR "Arrow/Parquet lacks the variant APIs required by ICEBERG_REQUIRE_ARROW_VARIANT. "
+                        "Use the pinned bundled Arrow (ICEBERG_USE_BUNDLED_ARROW=ON) "
+                        "or a compatible Arrow installation. Set ICEBERG_REQUIRE_ARROW_VARIANT=OFF "
+                        "only for builds that do not need variant I/O.")
+  endif()
+endfunction()
+
 function(resolve_arrow_dependency)
   if(ICEBERG_ENCRYPTION)
     resolve_openssl_for_encryption()
   endif()
 
   prepare_fetchcontent()
+  if(ICEBERG_USE_BUNDLED_ARROW)
+    # This override is scoped to Arrow; other dependencies can still use
+    # installed packages, including the Windows CI runner's vcpkg packages.
+    set(FETCHCONTENT_TRY_FIND_PACKAGE_MODE NEVER)
+  endif()
 
   # Prevent Arrow from injecting -Werror into CMAKE_CXX_FLAGS_DEBUG via
   # arrow_add_werror_if_debug(). PRODUCTION level only adds standard warnings.
@@ -340,6 +394,8 @@ function(resolve_arrow_dependency)
     list(APPEND ICEBERG_SYSTEM_DEPENDENCIES OpenSSL)
   endif()
 
+  iceberg_check_arrow_variant()
+
   set(ICEBERG_SYSTEM_DEPENDENCIES
       ${ICEBERG_SYSTEM_DEPENDENCIES}
       PARENT_SCOPE)

mkdocs/docs/getting-started.md

@@ -51,7 +51,7 @@ cmake --install build
 ### Bundle Library (with vendored dependencies)
 
 ```bash
-cmake -S . -B build -G Ninja -DCMAKE_INSTALL_PREFIX=/path/to/install -DICEBERG_BUILD_BUNDLE=ON
+cmake -S . -B build -G Ninja -DCMAKE_INSTALL_PREFIX=/path/to/install -DICEBERG_BUILD_BUNDLE=ON -DFETCHCONTENT_TRY_FIND_PACKAGE_MODE=NEVER
 cmake --build build
 cmake --install build
 ```
@@ -72,6 +72,8 @@ cmake --install build
 | `ICEBERG_BUILD_SHARED` | `OFF` | Build shared library |
 | `ICEBERG_BUILD_TESTS` | `ON` | Build tests |
 | `ICEBERG_BUILD_BUNDLE` | `ON` | Build the battery-included library |
+| `ICEBERG_USE_BUNDLED_ARROW` | `OFF` | Force the pinned Arrow source; CI enables this |
+| `ICEBERG_REQUIRE_ARROW_VARIANT` | `ON` | Require Arrow variant APIs in the bundle; CI enforces this |
 | `ICEBERG_BUILD_REST` | `ON` | Build REST catalog client |
 | `ICEBERG_BUILD_REST_INTEGRATION_TESTS` | `OFF` | Build REST catalog integration tests |
 | `ICEBERG_BUILD_HIVE` | `OFF` | Build Hive (HMS) catalog client |
@@ -159,6 +161,9 @@ If you experience network issues when downloading dependencies, you can override
 Example:
 
 ```bash
-export ICEBERG_ARROW_URL="https://your-mirror.com/apache-arrow-22.0.0.tar.gz"
+export ICEBERG_ARROW_URL="https://your-mirror.com/arrow-0f16412bb1c39611aeed7393eba09e637f4d86fa.tar.gz"
 cmake -S . -B build -G Ninja
 ```
+
+The Arrow mirror must serve the same pinned source archive: its SHA256 is checked
+against the value in `cmake_modules/IcebergThirdpartyToolchain.cmake`.

mkdocs/docs/library.md

@@ -42,6 +42,43 @@ are supplied by the `iceberg_bundle` library.
 This library combines `iceberg_data` with the Arrow C++, Avro, Parquet, and
 Arrow filesystem integrations. It gives applications one link-time dependency.
 
+Variant values use the `arrow.parquet.variant` extension with binary `metadata`
+and `value` storage. Call `iceberg::arrow::RegisterAll()` before importing these
+schemas into Arrow C++. Parquet readers and writers also register the extension
+when opened.
+
+Parquet variant I/O requires the Arrow variant builders, validation, and shredding
+utilities introduced by [Arrow PR #50252](https://github.com/apache/arrow/pull/50252).
+The bundled dependency temporarily pins Arrow revision
+`0f16412bb1c39611aeed7393eba09e637f4d86fa` from that PR and verifies the source
+archive's SHA256. It will be replaced by an Apache Arrow release containing these
+APIs. Use `ICEBERG_USE_BUNDLED_ARROW=ON` to force this Arrow source, or provide a
+compatible Arrow installation through `CMAKE_PREFIX_PATH`.
+
+`ICEBERG_REQUIRE_ARROW_VARIANT` defaults to `ON`: configuration fails when the
+selected Arrow/Parquet lacks the required APIs. CI explicitly uses the pinned
+Arrow source and keeps this option enabled; the variant tests fail to compile
+if their APIs are missing. To use an older Arrow installation for other
+functionality, explicitly set this option
+to `OFF`; Parquet variant I/O then returns `NotSupported`.
+
+Variants are written unshredded by default. To shred a field, supply a logical
+target keyed by its Iceberg field ID:
+
+```cpp
+iceberg::WriterOptions options;
+// Set the path, schema, and FileIO as usual.
+options.variant_shredding_types.emplace(7, iceberg::int64());
+```
+
+Targets can also contain structs and lists. Their field IDs are ignored; they
+describe the requested layout inside the variant. The layout stays fixed for
+the file, and incompatible values remain in residual storage. Input batches
+always use unshredded variants. Readers reconstruct the same unshredded schema
+from either file layout. Validation stays enabled, and a null column value is
+distinct from an encoded variant null. Avro variant data and variant bounds
+metrics are not supported.
+
 ### `iceberg_rest`, `iceberg_hive`, `iceberg_sql_catalog`
 
 These optional catalog libraries cover REST, Hive Metastore, and SQL databases.

src/iceberg/CMakeLists.txt

@@ -323,6 +323,7 @@ if(ICEBERG_BUILD_BUNDLE)
       parquet/parquet_reader.cc
       parquet/parquet_register.cc
       parquet/parquet_schema_util.cc
+      parquet/parquet_variant_util.cc
       parquet/parquet_writer.cc
       parquet/row_number_reader.cc)
 

src/iceberg/arrow/arrow_register.cc

@@ -23,6 +23,10 @@
 #include <string>
 #include <string_view>
 
+#include <arrow/extension/parquet_variant.h>
+#include <arrow/extension_type.h>
+#include <arrow/type.h>
+
 #include "iceberg/arrow/arrow_io_util.h"
 #include "iceberg/arrow/s3/s3_properties.h"
 #include "iceberg/file_io_registry.h"
@@ -57,6 +61,16 @@ void RegisterS3FileIO() {
 void EnsureArrowFileIOsRegistered() {
   static std::once_flag flag;
   std::call_once(flag, []() {
+    // Unlike UUID, Arrow does not register its variant extension by default.
+    // Register it so C Data imports reconstruct variant arrays from Iceberg's
+    // canonical metadata/value storage.
+    if (!::arrow::GetExtensionType("arrow.parquet.variant")) {
+      auto variant = ::arrow::extension::variant(
+          ::arrow::struct_({::arrow::field("metadata", ::arrow::binary(), false),
+                            ::arrow::field("value", ::arrow::binary(), false)}));
+      (void)::arrow::RegisterExtensionType(
+          std::static_pointer_cast<::arrow::ExtensionType>(variant));
+    }
     RegisterLocalFileIO();
     RegisterS3FileIO();
   });

src/iceberg/arrow/arrow_register.h

@@ -26,7 +26,7 @@
 
 namespace iceberg::arrow {
 
-/// \brief Register Arrow FileIOs and Arrow-backed C Data utilities.
+/// \brief Register Arrow FileIOs, the variant extension, and C Data utilities.
 ///
 /// This operation is idempotent and safe to call multiple times.
 ICEBERG_BUNDLE_EXPORT void RegisterAll();

src/iceberg/file_writer.h

@@ -25,6 +25,7 @@
 #include <functional>
 #include <memory>
 #include <optional>
+#include <unordered_map>
 #include <vector>
 
 #include "iceberg/arrow_c_data.h"
@@ -85,6 +86,11 @@ struct ICEBERG_EXPORT WriterOptions {
   std::shared_ptr<MetricsConfig> metrics_config = MetricsConfig::Default();
   /// \brief Format-specific or implementation-specific properties.
   WriterProperties properties;
+  /// \brief Optional logical shredding targets for Parquet variant fields, by field ID.
+  /// Targets may be supported primitive, struct, or list types. Their field IDs are
+  /// ignored. Unconfigured fields are written unshredded. Input arrays always use the
+  /// unshredded variant representation; the target layout stays fixed for this file.
+  std::unordered_map<int32_t, std::shared_ptr<Type>> variant_shredding_types;
 };
 
 /// \brief Base writer class to write data from different file formats.

src/iceberg/parquet/parquet_data_util.cc

@@ -30,7 +30,9 @@
 #include "iceberg/arrow/literal_util_internal.h"
 #include "iceberg/metadata_columns.h"
 #include "iceberg/parquet/parquet_data_util_internal.h"
+#include "iceberg/parquet/parquet_variant_util_internal.h"
 #include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
 #include "iceberg/schema_util.h"
 #include "iceberg/type.h"
 #include "iceberg/util/checked_cast.h"
@@ -117,16 +119,20 @@ Result<std::shared_ptr<::arrow::Array>> ProjectNestedArray(
 Result<std::shared_ptr<::arrow::Array>> MakeNullArray(
     const std::shared_ptr<::arrow::DataType>& type, int64_t length,
     ::arrow::MemoryPool* pool) {
-  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto builder, ::arrow::MakeBuilder(type, pool));
-  ICEBERG_ARROW_RETURN_NOT_OK(builder->AppendNulls(length));
-  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto array, builder->Finish());
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto array,
+                                 ::arrow::MakeArrayOfNull(type, length, pool));
   return array;
 }
 
 Result<std::shared_ptr<::arrow::Array>> ProjectPrimitiveArray(
     const std::shared_ptr<::arrow::Array>& array,
     const std::shared_ptr<::arrow::DataType>& output_arrow_type,
     ::arrow::MemoryPool* pool) {
+  if (output_arrow_type->id() == ::arrow::Type::EXTENSION &&
+      static_cast<const ::arrow::ExtensionType&>(*output_arrow_type).extension_name() ==
+          kArrowVariantExtensionName) {
+    return ProjectVariantArray(array, output_arrow_type, pool);
+  }
   if (array->type()->Equals(output_arrow_type)) {
     return array;
   }

src/iceberg/parquet/parquet_reader.cc

@@ -35,15 +35,18 @@
 #include <arrow/util/key_value_metadata.h>
 #include <parquet/arrow/reader.h>
 #include <parquet/arrow/schema.h>
+#include <parquet/exception.h>
 #include <parquet/file_reader.h>
 #include <parquet/properties.h>
 
 #include "iceberg/arrow/arrow_io_internal.h"
+#include "iceberg/arrow/arrow_register.h"
 #include "iceberg/arrow/arrow_status_internal.h"
 #include "iceberg/arrow/metadata_column_util_internal.h"
 #include "iceberg/parquet/parquet_data_util_internal.h"
 #include "iceberg/parquet/parquet_register.h"
 #include "iceberg/parquet/parquet_schema_util_internal.h"
+#include "iceberg/parquet/parquet_variant_util_internal.h"
 #include "iceberg/parquet/row_number_reader_internal.h"
 #include "iceberg/result.h"
 #include "iceberg/schema_internal.h"
@@ -277,9 +280,11 @@ class ParquetReader::Impl {
  public:
   // Open the Parquet reader with the given options
   Status Open(const ReaderOptions& options) {
+    arrow::RegisterAll();
     if (options.projection == nullptr) {
       return InvalidArgument("Projected schema is required by Parquet reader");
     }
+    ICEBERG_RETURN_UNEXPECTED(CheckVariantSupport(*options.projection));
 
     read_schema_ = options.projection;
 
@@ -289,6 +294,11 @@ class ParquetReader::Impl {
     arrow_reader_properties.set_batch_size(
         options.properties.Get(ReaderProperties::kBatchSize));
     arrow_reader_properties.set_arrow_extensions_enabled(true);
+#if ICEBERG_HAS_ARROW_VARIANT
+    // Arrow's per-column validator cannot see null ancestors of required variants.
+    // Validate the complete batch with its parent validity in Next() instead.
+    arrow_reader_properties.set_variant_validation_enabled(false);
+#endif
     use_large_list_ = options.properties.Get(ReaderProperties::kArrowUseLargeList);
     if (use_large_list_) {
       arrow_reader_properties.set_list_type(::arrow::Type::LARGE_LIST);
@@ -323,6 +333,7 @@ class ParquetReader::Impl {
     if (!batch) {
       return std::nullopt;
     }
+    ICEBERG_ASSIGN_OR_RAISE(batch, NormalizeVariantReadBatch(batch, pool_));
 
     std::shared_ptr<::arrow::Int64Array> row_positions;
     if (context_->row_number_reader_) {
@@ -496,9 +507,16 @@ class ParquetReader::Impl {
   std::unique_ptr<ReadContext> context_;
 };
 
+ParquetReader::ParquetReader() = default;
 ParquetReader::~ParquetReader() = default;
 
-Result<std::optional<ArrowArray>> ParquetReader::Next() { return impl_->Next(); }
+Result<std::optional<ArrowArray>> ParquetReader::Next() {
+  try {
+    return impl_->Next();
+  } catch (const ::parquet::ParquetException& error) {
+    return Invalid("Cannot read Parquet batch: {}", error.what());
+  }
+}
 
 Result<ArrowSchema> ParquetReader::Schema() { return impl_->Schema(); }
 
@@ -508,7 +526,11 @@ Result<std::unordered_map<std::string, std::string>> ParquetReader::Metadata() {
 
 Status ParquetReader::Open(const ReaderOptions& options) {
   impl_ = std::make_unique<Impl>();
-  return impl_->Open(options);
+  try {
+    return impl_->Open(options);
+  } catch (const ::parquet::ParquetException& error) {
+    return InvalidSchema("Cannot open Parquet reader: {}", error.what());
+  }
 }
 
 Status ParquetReader::Close() { return impl_->Close(); }