feat(variant): support variant data type

docs/source/user_guide/data_types.rst

@@ -217,3 +217,41 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
 
        The type can be declared using ``ROW<n0 t0 'd0', n1 t1 'd1', ...>`` where n
        is the unique name of a field, t is the logical type of a field, d is the description of a field.
+
+   * - ``VARIANT``
+     - Struct
+     - Data type of semi-structured data (e.g. JSON). A variant value contains one of:
+       a primitive (e.g. integer, string), an array of variant values, or an object
+       mapping string keys to variant values.
+
+       Variant values are encoded with two binaries following the parquet-format
+       `Variant Binary Encoding <https://github.com/apache/parquet-format/blob/master/VariantEncoding.md>`_
+       specification (compatible with the Java Paimon / Spark implementation). In
+       C++ Paimon, a variant field is represented in an Arrow schema as
+       ``Struct{value: Binary NOT NULL, metadata: Binary NOT NULL}`` marked with
+       Paimon-specific field metadata; use ``paimon::Variant::ArrowField`` to
+       construct such a field, and ``paimon::Variant`` (``FromJson``/``ToJson``/
+       ``VariantGet``) to build and inspect values.
+
+       Only the parquet file format supports VARIANT columns. VARIANT cannot be
+       used as a primary key, partition key or bucket key, and no predicate
+       pushdown applies to it.
+
+       When writing, variant columns can optionally be *shredded* into typed
+       parquet columns per the parquet-format
+       `Variant Shredding <https://github.com/apache/parquet-format/blob/master/VariantShredding.md>`_
+       specification by setting ``variant.shreddingSchema`` to a ROW type JSON
+       whose fields map variant column names to their shredding types.
+       Alternatively, setting ``variant.inferShreddingSchema`` to ``true``
+       infers a shredding schema per file from the first written rows
+       (tuned by ``variant.shredding.maxSchemaWidth``,
+       ``variant.shredding.maxSchemaDepth``,
+       ``variant.shredding.minFieldCardinalityRatio`` and
+       ``variant.shredding.maxInferBufferRow``). Readers reassemble shredded
+       columns transparently.
+
+       When reading, instead of the full variant, specific paths can be
+       extracted by replacing the variant column in the read schema with a
+       projection built by ``paimon::VariantAccessBuilder`` (e.g. ``$.a.b`` as
+       BIGINT). For shredded files only the required typed sub-columns are
+       read.

include/paimon/data/variant.h

@@ -0,0 +1,197 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <unordered_map>
+
+#include "paimon/memory/memory_pool.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+
+struct ArrowArray;
+struct ArrowSchema;
+
+namespace paimon {
+
+/// Arguments controlling how a variant value is cast to a target type in `Variant::VariantGet`.
+struct PAIMON_EXPORT VariantCastArgs {
+    /// Whether an invalid cast fails the call (true) or yields SQL NULL (false).
+    bool fail_on_error = true;
+    /// The time zone used when rendering TIMESTAMP values. Supported forms are `UTC`/`Z`/`GMT`,
+    /// fixed offsets such as `+08:00`, and IANA region ids such as `Asia/Shanghai`.
+    std::string zone_id = "UTC";
+};
+
+/// A Variant represents a type that contains one of: 1) Primitive: A type and corresponding
+/// value (e.g. INT, STRING); 2) Array: An ordered list of Variant values; 3) Object: An
+/// unordered collection of string/Variant pairs (i.e. key/value pairs). An object may not
+/// contain duplicate keys.
+///
+/// A Variant is encoded with 2 binaries: the value and the metadata, following the parquet
+/// Variant Binary Encoding specification (compatible with the Java / Spark implementation). The
+/// encoding allows representation of semi-structured data (e.g. JSON) in a form that can be
+/// efficiently queried by path.
+///
+/// In an Arrow schema, a Variant field is represented as
+/// `struct<value: binary not null, metadata: binary not null>` marked with Paimon-specific field
+/// metadata; use `Variant::ArrowField` to construct such a field.
+class PAIMON_EXPORT Variant {
+ public:
+    ~Variant();
+
+    /// Parses a JSON string as a Variant (duplicate object keys are rejected).
+    ///
+    /// @param json The JSON document text.
+    /// @param pool The memory pool used for the variant buffers.
+    /// @return A result containing the created variant or an error.
+    static Result<std::unique_ptr<Variant>> FromJson(const std::string& json,
+                                                     const std::shared_ptr<MemoryPool>& pool);
+
+    /// Creates a Variant from already-encoded value and metadata binaries (copied into `pool`).
+    ///
+    /// @param value The variant value binary.
+    /// @param value_length The length of the value binary.
+    /// @param metadata The variant metadata binary.
+    /// @param metadata_length The length of the metadata binary.
+    /// @param pool The memory pool used for the variant buffers.
+    /// @return A result containing the created variant or an error.
+    static Result<std::unique_ptr<Variant>> Create(const char* value, uint64_t value_length,
+                                                   const char* metadata, uint64_t metadata_length,
+                                                   const std::shared_ptr<MemoryPool>& pool);
+
+    /// The variant value binary. The view remains valid as long as this variant exists.
+    std::string_view Value() const;
+
+    /// The variant metadata binary. The view remains valid as long as this variant exists.
+    std::string_view Metadata() const;
+
+    /// The size of the variant in bytes (value size + metadata size).
+    int64_t SizeInBytes() const;
+
+    /// Stringifies the variant in JSON format.
+    ///
+    /// @param zone_id The time zone used when rendering TIMESTAMP values.
+    /// @return A result containing the JSON text or an error.
+    Result<std::string> ToJson(const std::string& zone_id = "UTC") const;
+
+    /// Extracts a sub-variant value according to a path which starts with a `$`, e.g. `$.key`,
+    /// `$['key']`, `$["key"]`, `$.array[0]`, and casts the value to the target type.
+    ///
+    /// @param path The extraction path.
+    /// @param target_type The target Arrow type (C data interface, consumed by the call); only
+    ///        scalar types are supported. Use `VariantGetArrow` for nested targets.
+    /// @param cast_args Cast behavior arguments.
+    /// @return A result containing the extracted literal, or nullopt for SQL NULL (unmatched
+    ///         path, variant null, or an invalid cast with `fail_on_error == false`).
+    Result<std::optional<Literal>> VariantGet(const std::string& path,
+                                              struct ArrowSchema* target_type,
+                                              const VariantCastArgs& cast_args) const;
+
+    /// Extracts a sub-variant value according to a path which starts with a `$` and casts the
+    /// value to the (possibly nested) target field type.
+    ///
+    /// In addition to scalar types, the target may be a STRUCT (cast from a variant object,
+    /// children matched by field name; unmatched children are null), a MAP with string keys
+    /// (cast from a variant object), a LIST (cast from a variant array), or a variant-marked
+    /// field created by `Variant::ArrowField` (the sub-variant is deeply re-encoded).
+    ///
+    /// @param path The extraction path.
+    /// @param target_field The target Arrow field (C data interface, consumed by the call).
+    /// @param cast_args Cast behavior arguments.
+    /// @return A result containing a length-1 Arrow array of the target type whose single slot
+    ///         holds the result; a null slot represents SQL NULL (unmatched path, variant null,
+    ///         or an invalid cast with `fail_on_error == false`).
+    Result<std::unique_ptr<struct ArrowArray>> VariantGetArrow(
+        const std::string& path, struct ArrowSchema* target_field,
+        const VariantCastArgs& cast_args) const;
+
+    /// Extracts a sub-variant value according to a path which starts with a `$` and renders it
+    /// as JSON text.
+    ///
+    /// @param path The extraction path.
+    /// @param zone_id The time zone used when rendering TIMESTAMP values.
+    /// @return A result containing the JSON text, or nullopt if the path does not match.
+    Result<std::optional<std::string>> VariantGetJson(const std::string& path,
+                                                      const std::string& zone_id = "UTC") const;
+
+    /// Creates an Arrow field definition for the Variant type.
+    ///
+    /// This function constructs an Arrow Field (internally
+    /// `struct<value: binary not null, metadata: binary not null>`) and exports it to the C data
+    /// interface structure `::ArrowSchema`. It automatically injects Paimon-specific metadata to
+    /// identify the field as a VARIANT.
+    ///
+    /// @param field_name The name of the Arrow field.
+    /// @param nullable Whether the field is nullable.
+    /// @param metadata A map of key-value metadata to be attached to the field.
+    /// @return A result containing a unique pointer to the generated `::ArrowSchema` or an error.
+    static Result<std::unique_ptr<struct ArrowSchema>> ArrowField(
+        const std::string& field_name, bool nullable = true,
+        std::unordered_map<std::string, std::string> metadata = {});
+
+ private:
+    class Impl;
+
+    explicit Variant(std::unique_ptr<Impl>&& impl);
+
+    std::unique_ptr<Impl> impl_;
+};
+
+/// Builds a variant-access projection field: a struct field that replaces a VARIANT column in
+/// the read schema so that, instead of the full variant, only the described paths are extracted
+/// at read time (reading only the required shredded sub-columns from shredded files).
+///
+/// Example: read `$.age` as INT64 and `$.city` as STRING from variant column `v`:
+///
+///     VariantAccessBuilder builder;
+///     builder.AddField(age_type, "$.age");
+///     builder.AddField(city_type, "$.city");
+///     auto field = builder.Build("v");  // use in ReadContextBuilder::SetReadSchema
+///
+/// The resulting struct has one child per added field, named by its position ("0", "1", ...).
+class PAIMON_EXPORT VariantAccessBuilder {
+ public:
+    VariantAccessBuilder();
+    ~VariantAccessBuilder();
+
+    /// Adds an extracted field.
+    ///
+    /// @param target_type The Arrow type the extracted value is cast to (C data interface,
+    ///        consumed by the call). Nested targets follow `Variant::VariantGetArrow` semantics.
+    /// @param path The extraction path, e.g. `$.a.b` or `$.array[0]`.
+    /// @param fail_on_error Whether an invalid cast fails the read (true) or yields SQL NULL.
+    /// @param zone_id The time zone used when rendering TIMESTAMP values.
+    Status AddField(struct ArrowSchema* target_type, const std::string& path,
+                    bool fail_on_error = true, const std::string& zone_id = "UTC");
+
+    /// Builds the projection field named `field_name`, to be used in the read schema in place
+    /// of the variant column with the same name.
+    Result<std::unique_ptr<struct ArrowSchema>> Build(const std::string& field_name) const;
+
+ private:
+    class Impl;
+
+    std::unique_ptr<Impl> impl_;
+};
+
+}  // namespace paimon

include/paimon/defs.h

@@ -46,6 +46,7 @@ enum class FieldType {
     MAP = 14,
     STRUCT = 15,
     BLOB = 16,
+    VARIANT = 17,
     UNKNOWN = 128,
 };
 
@@ -391,6 +392,28 @@ struct PAIMON_EXPORT Options {
     /// Only effective when map.storage-layout = shared-shredding.
     static const char MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY[];
 
+    /// "variant.shreddingSchema" - The Variant shredding schema for writing: a ROW type JSON
+    /// whose fields map variant column names to their shredding types. No default value.
+    static const char VARIANT_SHREDDING_SCHEMA[];
+    /// "parquet.variant.shreddingSchema" - Fallback key of "variant.shreddingSchema".
+    static const char PARQUET_VARIANT_SHREDDING_SCHEMA[];
+    /// "variant.inferShreddingSchema" - Whether to automatically infer the shredding schema when
+    /// writing Variant columns. Default value is "false".
+    static const char VARIANT_INFER_SHREDDING_SCHEMA[];
+    /// "variant.shredding.maxSchemaWidth" - Maximum number of shredded fields allowed in an
+    /// inferred schema. Default value is 300.
+    static const char VARIANT_SHREDDING_MAX_SCHEMA_WIDTH[];
+    /// "variant.shredding.maxSchemaDepth" - Maximum traversal depth in Variant values during
+    /// schema inference. Default value is 50.
+    static const char VARIANT_SHREDDING_MAX_SCHEMA_DEPTH[];
+    /// "variant.shredding.minFieldCardinalityRatio" - Minimum fraction of rows that must contain
+    /// a field for it to be shredded. Fields below this threshold stay in the un-shredded
+    /// Variant binary. Default value is 0.1.
+    static const char VARIANT_SHREDDING_MIN_FIELD_CARDINALITY_RATIO[];
+    /// "variant.shredding.maxInferBufferRow" - Maximum number of rows to buffer for schema
+    /// inference. Default value is 4096.
+    static const char VARIANT_SHREDDING_MAX_INFER_BUFFER_ROW[];
+
     /// "blob-as-descriptor" - Read blob field using blob descriptor rather than blob
     /// bytes. Default value is "false".
     static const char BLOB_AS_DESCRIPTOR[];

src/paimon/CMakeLists.txt

@@ -38,6 +38,23 @@ set(PAIMON_COMMON_SRCS
     common/data/serializer/row_compacted_serializer.cpp
     common/data/serializer/binary_serializer_utils.cpp
     common/data/timestamp.cpp
+    common/data/variant/generic_variant.cpp
+    common/data/variant/infer_variant_shredding_schema.cpp
+    common/data/variant/variant.cpp
+    common/data/variant/variant_binary_util.cpp
+    common/data/variant/variant_builder.cpp
+    common/data/variant/variant_get.cpp
+    common/data/variant/variant_json_utils.cpp
+    common/data/variant/variant_path_segment.cpp
+    common/data/variant/variant_reassembler.cpp
+    common/data/variant/variant_shredding_batch_converter.cpp
+    common/data/variant/variant_access_utils.cpp
+    common/data/variant/variant_shredding_read_plan_factory.cpp
+    common/data/variant/variant_shredding_utils.cpp
+    common/data/variant/variant_shredding_write_plan.cpp
+    common/data/variant/variant_shredding_write_plan_factory.cpp
+    common/data/variant/variant_shredding_writer.cpp
+    common/data/variant/variant_type_utils.cpp
     common/defs.cpp
     common/executor/executor.cpp
     common/factories/singleton.cpp
@@ -139,11 +156,14 @@ set(PAIMON_COMMON_SRCS
     common/utils/crc32c.cpp
     common/utils/decimal_utils.cpp
     common/data/shredding/map_shared_shredding_utils.cpp
+    common/data/shredding/map_shared_shredding_write_plan_factory.cpp
+    common/data/shredding/shredding_write_plan_factories.cpp
     common/data/shredding/map_shared_shredding_context.cpp
     common/data/shredding/map_shared_shredding_batch_converter.cpp
     common/data/shredding/map_shared_shredding_column_allocator.cpp
     common/data/shredding/lru_map_shared_shredding_column_allocator.cpp
     common/data/shredding/map_shared_shredding_file_reader.cpp
+    common/data/shredding/shredding_file_reader.cpp
     common/utils/delta_varint_compressor.cpp
     common/utils/fields_comparator.cpp
     common/utils/path_util.cpp
@@ -428,6 +448,13 @@ if(PAIMON_BUILD_TESTS)
                     common/data/blob_descriptor_test.cpp
                     common/data/blob_view_struct_test.cpp
                     common/data/blob_utils_test.cpp
+                    common/data/variant/generic_variant_test.cpp
+                    common/data/variant/infer_variant_shredding_schema_test.cpp
+                    common/data/variant/variant_shredding_write_plan_factory_test.cpp
+                    common/data/variant/variant_get_test.cpp
+                    common/data/variant/variant_shredding_test.cpp
+                    common/data/variant/variant_test.cpp
+                    common/data/variant/variant_type_utils_test.cpp
                     common/executor/default_executor_test.cpp
                     common/format/column_stats_test.cpp
                     common/fs/external_path_provider_test.cpp
@@ -620,6 +647,7 @@ if(PAIMON_BUILD_TESTS)
                     core/index/index_file_meta_serializer_test.cpp
                     core/index/index_file_handler_test.cpp
                     core/io/compact_increment_test.cpp
+                    core/io/infer_shredding_file_writer_test.cpp
                     core/io/concat_key_value_record_reader_test.cpp
                     core/io/data_file_meta_serializer_test.cpp
                     core/io/data_file_path_factory_test.cpp

src/paimon/common/data/shredding/map_shared_shredding_batch_converter.h

@@ -27,6 +27,7 @@
 #include "paimon/common/data/shredding/map_shared_shredding_column_allocator.h"
 #include "paimon/common/data/shredding/map_shared_shredding_field_dict.h"
 #include "paimon/common/data/shredding/map_shredding_defs.h"
+#include "paimon/common/data/shredding/shredding_batch_converter.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/result.h"
 #include "paimon/status.h"
@@ -44,7 +45,7 @@ class MapSharedShreddingContext;
 ///
 /// Non-shared-shredding columns are passed through unchanged.
 /// Each shared-shredding column has its own FieldDict and ColumnAllocator.
-class MapSharedShreddingBatchConverter {
+class MapSharedShreddingBatchConverter : public ShreddingBatchConverter {
  public:
     /// Creates a converter for one file write cycle.
     /// Computes per-file K from context, builds physical schema, and constructs the converter.
@@ -59,12 +60,12 @@ class MapSharedShreddingBatchConverter {
         const std::shared_ptr<MemoryPool>& pool);
 
     /// Returns the physical schema produced for this converter.
-    const std::shared_ptr<arrow::Schema>& GetPhysicalSchema() const;
+    const std::shared_ptr<arrow::Schema>& GetPhysicalSchema() const override;
 
     /// Converts a logical batch to a physical batch.
     /// @param logical_batch Input ArrowArray (C ABI) with logical schema. Consumed on success.
     /// @return Owned physical ArrowArray (C ABI) with physical schema.
-    Result<std::unique_ptr<ArrowArray>> Convert(ArrowArray* logical_batch);
+    Result<std::unique_ptr<ArrowArray>> Convert(ArrowArray* logical_batch) override;
 
     /// Builds MapSharedShreddingFieldMeta for one shredding column (by field name).
     /// Called at file close to serialize metadata.