Skip to content

Commit c6ceefb

Browse files
SteNicholasclaude
andcommitted
feat(variant): support variant data type
Add the VARIANT data type for storing and efficiently querying semi-structured data (e.g. JSON), following the parquet-format VariantEncoding.md / VariantShredding.md specifications. - Type system: VARIANT is parsed from / serialized to "VARIANT" and is represented in Arrow as struct<value: binary, metadata: binary> marked with paimon.extension.type=paimon.type.variant; field-id assignment and schema validation treat variant structs as leaves. - Public API: paimon::Variant builds variants from JSON, renders them back to JSON, and extracts values by path (e.g. $.a[0]) cast to scalar targets or, via VariantGetArrow, to nested STRUCT / LIST / MAP / VARIANT targets. - Parquet: a variant column is written as an unannotated group of REQUIRED BINARY value (field id 0) and metadata (field id 1); ORC/Avro report NotImplemented for VARIANT columns. - Shredding writes: a generic write-plan framework shared with MAP shared-shredding rewrites variant columns into typed parquet sub-columns, either from the configured variant.shreddingSchema or inferred per file from sampled rows when variant.inferShreddingSchema is enabled (tuned by the variant.shredding.* options). - Shredding reads: shredded files are detected structurally and reassembled back into variants transparently; a variant column can also be read as a variant-access projection that extracts specific paths while pruning the scan to metadata and the requested typed_value sub-columns. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent ffbda4f commit c6ceefb

96 files changed

Lines changed: 10699 additions & 75 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

docs/source/user_guide/data_types.rst

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -217,3 +217,41 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
217217

218218
The type can be declared using ``ROW<n0 t0 'd0', n1 t1 'd1', ...>`` where n
219219
is the unique name of a field, t is the logical type of a field, d is the description of a field.
220+
221+
* - ``VARIANT``
222+
- Struct
223+
- Data type of semi-structured data (e.g. JSON). A variant value contains one of:
224+
a primitive (e.g. integer, string), an array of variant values, or an object
225+
mapping string keys to variant values.
226+
227+
Variant values are encoded with two binaries following the parquet-format
228+
`Variant Binary Encoding <https://github.com/apache/parquet-format/blob/master/VariantEncoding.md>`_
229+
specification (compatible with the Java Paimon / Spark implementation). In
230+
C++ Paimon, a variant field is represented in an Arrow schema as
231+
``Struct{value: Binary NOT NULL, metadata: Binary NOT NULL}`` marked with
232+
Paimon-specific field metadata; use ``paimon::Variant::ArrowField`` to
233+
construct such a field, and ``paimon::Variant`` (``FromJson``/``ToJson``/
234+
``VariantGet``) to build and inspect values.
235+
236+
Only the parquet file format supports VARIANT columns. VARIANT cannot be
237+
used as a primary key, partition key or bucket key, and no predicate
238+
pushdown applies to it.
239+
240+
When writing, variant columns can optionally be *shredded* into typed
241+
parquet columns per the parquet-format
242+
`Variant Shredding <https://github.com/apache/parquet-format/blob/master/VariantShredding.md>`_
243+
specification by setting ``variant.shreddingSchema`` to a ROW type JSON
244+
whose fields map variant column names to their shredding types.
245+
Alternatively, setting ``variant.inferShreddingSchema`` to ``true``
246+
infers a shredding schema per file from the first written rows
247+
(tuned by ``variant.shredding.maxSchemaWidth``,
248+
``variant.shredding.maxSchemaDepth``,
249+
``variant.shredding.minFieldCardinalityRatio`` and
250+
``variant.shredding.maxInferBufferRow``). Readers reassemble shredded
251+
columns transparently.
252+
253+
When reading, instead of the full variant, specific paths can be
254+
extracted by replacing the variant column in the read schema with a
255+
projection built by ``paimon::VariantAccessBuilder`` (e.g. ``$.a.b`` as
256+
BIGINT). For shredded files only the required typed sub-columns are
257+
read.

include/paimon/data/variant.h

Lines changed: 197 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,197 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#pragma once
18+
19+
#include <cstdint>
20+
#include <memory>
21+
#include <optional>
22+
#include <string>
23+
#include <string_view>
24+
#include <unordered_map>
25+
26+
#include "paimon/memory/memory_pool.h"
27+
#include "paimon/predicate/literal.h"
28+
#include "paimon/result.h"
29+
#include "paimon/visibility.h"
30+
31+
struct ArrowArray;
32+
struct ArrowSchema;
33+
34+
namespace paimon {
35+
36+
/// Arguments controlling how a variant value is cast to a target type in `Variant::VariantGet`.
37+
struct PAIMON_EXPORT VariantCastArgs {
38+
/// Whether an invalid cast fails the call (true) or yields SQL NULL (false).
39+
bool fail_on_error = true;
40+
/// The time zone used when rendering TIMESTAMP values. Supported forms are `UTC`/`Z`/`GMT`,
41+
/// fixed offsets such as `+08:00`, and IANA region ids such as `Asia/Shanghai`.
42+
std::string zone_id = "UTC";
43+
};
44+
45+
/// A Variant represents a type that contains one of: 1) Primitive: A type and corresponding
46+
/// value (e.g. INT, STRING); 2) Array: An ordered list of Variant values; 3) Object: An
47+
/// unordered collection of string/Variant pairs (i.e. key/value pairs). An object may not
48+
/// contain duplicate keys.
49+
///
50+
/// A Variant is encoded with 2 binaries: the value and the metadata, following the parquet
51+
/// Variant Binary Encoding specification (compatible with the Java / Spark implementation). The
52+
/// encoding allows representation of semi-structured data (e.g. JSON) in a form that can be
53+
/// efficiently queried by path.
54+
///
55+
/// In an Arrow schema, a Variant field is represented as
56+
/// `struct<value: binary not null, metadata: binary not null>` marked with Paimon-specific field
57+
/// metadata; use `Variant::ArrowField` to construct such a field.
58+
class PAIMON_EXPORT Variant {
59+
public:
60+
~Variant();
61+
62+
/// Parses a JSON string as a Variant (duplicate object keys are rejected).
63+
///
64+
/// @param json The JSON document text.
65+
/// @param pool The memory pool used for the variant buffers.
66+
/// @return A result containing the created variant or an error.
67+
static Result<std::unique_ptr<Variant>> FromJson(const std::string& json,
68+
const std::shared_ptr<MemoryPool>& pool);
69+
70+
/// Creates a Variant from already-encoded value and metadata binaries (copied into `pool`).
71+
///
72+
/// @param value The variant value binary.
73+
/// @param value_length The length of the value binary.
74+
/// @param metadata The variant metadata binary.
75+
/// @param metadata_length The length of the metadata binary.
76+
/// @param pool The memory pool used for the variant buffers.
77+
/// @return A result containing the created variant or an error.
78+
static Result<std::unique_ptr<Variant>> Create(const char* value, uint64_t value_length,
79+
const char* metadata, uint64_t metadata_length,
80+
const std::shared_ptr<MemoryPool>& pool);
81+
82+
/// The variant value binary. The view remains valid as long as this variant exists.
83+
std::string_view Value() const;
84+
85+
/// The variant metadata binary. The view remains valid as long as this variant exists.
86+
std::string_view Metadata() const;
87+
88+
/// The size of the variant in bytes (value size + metadata size).
89+
int64_t SizeInBytes() const;
90+
91+
/// Stringifies the variant in JSON format.
92+
///
93+
/// @param zone_id The time zone used when rendering TIMESTAMP values.
94+
/// @return A result containing the JSON text or an error.
95+
Result<std::string> ToJson(const std::string& zone_id = "UTC") const;
96+
97+
/// Extracts a sub-variant value according to a path which starts with a `$`, e.g. `$.key`,
98+
/// `$['key']`, `$["key"]`, `$.array[0]`, and casts the value to the target type.
99+
///
100+
/// @param path The extraction path.
101+
/// @param target_type The target Arrow type (C data interface, consumed by the call); only
102+
/// scalar types are supported. Use `VariantGetArrow` for nested targets.
103+
/// @param cast_args Cast behavior arguments.
104+
/// @return A result containing the extracted literal, or nullopt for SQL NULL (unmatched
105+
/// path, variant null, or an invalid cast with `fail_on_error == false`).
106+
Result<std::optional<Literal>> VariantGet(const std::string& path,
107+
struct ArrowSchema* target_type,
108+
const VariantCastArgs& cast_args) const;
109+
110+
/// Extracts a sub-variant value according to a path which starts with a `$` and casts the
111+
/// value to the (possibly nested) target field type.
112+
///
113+
/// In addition to scalar types, the target may be a STRUCT (cast from a variant object,
114+
/// children matched by field name; unmatched children are null), a MAP with string keys
115+
/// (cast from a variant object), a LIST (cast from a variant array), or a variant-marked
116+
/// field created by `Variant::ArrowField` (the sub-variant is deeply re-encoded).
117+
///
118+
/// @param path The extraction path.
119+
/// @param target_field The target Arrow field (C data interface, consumed by the call).
120+
/// @param cast_args Cast behavior arguments.
121+
/// @return A result containing a length-1 Arrow array of the target type whose single slot
122+
/// holds the result; a null slot represents SQL NULL (unmatched path, variant null,
123+
/// or an invalid cast with `fail_on_error == false`).
124+
Result<std::unique_ptr<struct ArrowArray>> VariantGetArrow(
125+
const std::string& path, struct ArrowSchema* target_field,
126+
const VariantCastArgs& cast_args) const;
127+
128+
/// Extracts a sub-variant value according to a path which starts with a `$` and renders it
129+
/// as JSON text.
130+
///
131+
/// @param path The extraction path.
132+
/// @param zone_id The time zone used when rendering TIMESTAMP values.
133+
/// @return A result containing the JSON text, or nullopt if the path does not match.
134+
Result<std::optional<std::string>> VariantGetJson(const std::string& path,
135+
const std::string& zone_id = "UTC") const;
136+
137+
/// Creates an Arrow field definition for the Variant type.
138+
///
139+
/// This function constructs an Arrow Field (internally
140+
/// `struct<value: binary not null, metadata: binary not null>`) and exports it to the C data
141+
/// interface structure `::ArrowSchema`. It automatically injects Paimon-specific metadata to
142+
/// identify the field as a VARIANT.
143+
///
144+
/// @param field_name The name of the Arrow field.
145+
/// @param nullable Whether the field is nullable.
146+
/// @param metadata A map of key-value metadata to be attached to the field.
147+
/// @return A result containing a unique pointer to the generated `::ArrowSchema` or an error.
148+
static Result<std::unique_ptr<struct ArrowSchema>> ArrowField(
149+
const std::string& field_name, bool nullable = true,
150+
std::unordered_map<std::string, std::string> metadata = {});
151+
152+
private:
153+
class Impl;
154+
155+
explicit Variant(std::unique_ptr<Impl>&& impl);
156+
157+
std::unique_ptr<Impl> impl_;
158+
};
159+
160+
/// Builds a variant-access projection field: a struct field that replaces a VARIANT column in
161+
/// the read schema so that, instead of the full variant, only the described paths are extracted
162+
/// at read time (reading only the required shredded sub-columns from shredded files).
163+
///
164+
/// Example: read `$.age` as INT64 and `$.city` as STRING from variant column `v`:
165+
///
166+
/// VariantAccessBuilder builder;
167+
/// builder.AddField(age_type, "$.age");
168+
/// builder.AddField(city_type, "$.city");
169+
/// auto field = builder.Build("v"); // use in ReadContextBuilder::SetReadSchema
170+
///
171+
/// The resulting struct has one child per added field, named by its position ("0", "1", ...).
172+
class PAIMON_EXPORT VariantAccessBuilder {
173+
public:
174+
VariantAccessBuilder();
175+
~VariantAccessBuilder();
176+
177+
/// Adds an extracted field.
178+
///
179+
/// @param target_type The Arrow type the extracted value is cast to (C data interface,
180+
/// consumed by the call). Nested targets follow `Variant::VariantGetArrow` semantics.
181+
/// @param path The extraction path, e.g. `$.a.b` or `$.array[0]`.
182+
/// @param fail_on_error Whether an invalid cast fails the read (true) or yields SQL NULL.
183+
/// @param zone_id The time zone used when rendering TIMESTAMP values.
184+
Status AddField(struct ArrowSchema* target_type, const std::string& path,
185+
bool fail_on_error = true, const std::string& zone_id = "UTC");
186+
187+
/// Builds the projection field named `field_name`, to be used in the read schema in place
188+
/// of the variant column with the same name.
189+
Result<std::unique_ptr<struct ArrowSchema>> Build(const std::string& field_name) const;
190+
191+
private:
192+
class Impl;
193+
194+
std::unique_ptr<Impl> impl_;
195+
};
196+
197+
} // namespace paimon

include/paimon/defs.h

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@ enum class FieldType {
4646
MAP = 14,
4747
STRUCT = 15,
4848
BLOB = 16,
49+
VARIANT = 17,
4950
UNKNOWN = 128,
5051
};
5152

@@ -391,6 +392,28 @@ struct PAIMON_EXPORT Options {
391392
/// Only effective when map.storage-layout = shared-shredding.
392393
static const char MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY[];
393394

395+
/// "variant.shreddingSchema" - The Variant shredding schema for writing: a ROW type JSON
396+
/// whose fields map variant column names to their shredding types. No default value.
397+
static const char VARIANT_SHREDDING_SCHEMA[];
398+
/// "parquet.variant.shreddingSchema" - Fallback key of "variant.shreddingSchema".
399+
static const char PARQUET_VARIANT_SHREDDING_SCHEMA[];
400+
/// "variant.inferShreddingSchema" - Whether to automatically infer the shredding schema when
401+
/// writing Variant columns. Default value is "false".
402+
static const char VARIANT_INFER_SHREDDING_SCHEMA[];
403+
/// "variant.shredding.maxSchemaWidth" - Maximum number of shredded fields allowed in an
404+
/// inferred schema. Default value is 300.
405+
static const char VARIANT_SHREDDING_MAX_SCHEMA_WIDTH[];
406+
/// "variant.shredding.maxSchemaDepth" - Maximum traversal depth in Variant values during
407+
/// schema inference. Default value is 50.
408+
static const char VARIANT_SHREDDING_MAX_SCHEMA_DEPTH[];
409+
/// "variant.shredding.minFieldCardinalityRatio" - Minimum fraction of rows that must contain
410+
/// a field for it to be shredded. Fields below this threshold stay in the un-shredded
411+
/// Variant binary. Default value is 0.1.
412+
static const char VARIANT_SHREDDING_MIN_FIELD_CARDINALITY_RATIO[];
413+
/// "variant.shredding.maxInferBufferRow" - Maximum number of rows to buffer for schema
414+
/// inference. Default value is 4096.
415+
static const char VARIANT_SHREDDING_MAX_INFER_BUFFER_ROW[];
416+
394417
/// "blob-as-descriptor" - Read blob field using blob descriptor rather than blob
395418
/// bytes. Default value is "false".
396419
static const char BLOB_AS_DESCRIPTOR[];

src/paimon/CMakeLists.txt

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,23 @@ set(PAIMON_COMMON_SRCS
3838
common/data/serializer/row_compacted_serializer.cpp
3939
common/data/serializer/binary_serializer_utils.cpp
4040
common/data/timestamp.cpp
41+
common/data/variant/generic_variant.cpp
42+
common/data/variant/infer_variant_shredding_schema.cpp
43+
common/data/variant/variant.cpp
44+
common/data/variant/variant_binary_util.cpp
45+
common/data/variant/variant_builder.cpp
46+
common/data/variant/variant_get.cpp
47+
common/data/variant/variant_json_utils.cpp
48+
common/data/variant/variant_path_segment.cpp
49+
common/data/variant/variant_reassembler.cpp
50+
common/data/variant/variant_shredding_batch_converter.cpp
51+
common/data/variant/variant_access_utils.cpp
52+
common/data/variant/variant_shredding_read_plan_factory.cpp
53+
common/data/variant/variant_shredding_utils.cpp
54+
common/data/variant/variant_shredding_write_plan.cpp
55+
common/data/variant/variant_shredding_write_plan_factory.cpp
56+
common/data/variant/variant_shredding_writer.cpp
57+
common/data/variant/variant_type_utils.cpp
4158
common/defs.cpp
4259
common/executor/executor.cpp
4360
common/factories/singleton.cpp
@@ -139,11 +156,14 @@ set(PAIMON_COMMON_SRCS
139156
common/utils/crc32c.cpp
140157
common/utils/decimal_utils.cpp
141158
common/data/shredding/map_shared_shredding_utils.cpp
159+
common/data/shredding/map_shared_shredding_write_plan_factory.cpp
160+
common/data/shredding/shredding_write_plan_factories.cpp
142161
common/data/shredding/map_shared_shredding_context.cpp
143162
common/data/shredding/map_shared_shredding_batch_converter.cpp
144163
common/data/shredding/map_shared_shredding_column_allocator.cpp
145164
common/data/shredding/lru_map_shared_shredding_column_allocator.cpp
146165
common/data/shredding/map_shared_shredding_file_reader.cpp
166+
common/data/shredding/shredding_file_reader.cpp
147167
common/utils/delta_varint_compressor.cpp
148168
common/utils/fields_comparator.cpp
149169
common/utils/path_util.cpp
@@ -428,6 +448,13 @@ if(PAIMON_BUILD_TESTS)
428448
common/data/blob_descriptor_test.cpp
429449
common/data/blob_view_struct_test.cpp
430450
common/data/blob_utils_test.cpp
451+
common/data/variant/generic_variant_test.cpp
452+
common/data/variant/infer_variant_shredding_schema_test.cpp
453+
common/data/variant/variant_shredding_write_plan_factory_test.cpp
454+
common/data/variant/variant_get_test.cpp
455+
common/data/variant/variant_shredding_test.cpp
456+
common/data/variant/variant_test.cpp
457+
common/data/variant/variant_type_utils_test.cpp
431458
common/executor/default_executor_test.cpp
432459
common/format/column_stats_test.cpp
433460
common/fs/external_path_provider_test.cpp
@@ -620,6 +647,7 @@ if(PAIMON_BUILD_TESTS)
620647
core/index/index_file_meta_serializer_test.cpp
621648
core/index/index_file_handler_test.cpp
622649
core/io/compact_increment_test.cpp
650+
core/io/infer_shredding_file_writer_test.cpp
623651
core/io/concat_key_value_record_reader_test.cpp
624652
core/io/data_file_meta_serializer_test.cpp
625653
core/io/data_file_path_factory_test.cpp

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

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
#include "paimon/common/data/shredding/map_shared_shredding_column_allocator.h"
2828
#include "paimon/common/data/shredding/map_shared_shredding_field_dict.h"
2929
#include "paimon/common/data/shredding/map_shredding_defs.h"
30+
#include "paimon/common/data/shredding/shredding_batch_converter.h"
3031
#include "paimon/memory/memory_pool.h"
3132
#include "paimon/result.h"
3233
#include "paimon/status.h"
@@ -44,7 +45,7 @@ class MapSharedShreddingContext;
4445
///
4546
/// Non-shared-shredding columns are passed through unchanged.
4647
/// Each shared-shredding column has its own FieldDict and ColumnAllocator.
47-
class MapSharedShreddingBatchConverter {
48+
class MapSharedShreddingBatchConverter : public ShreddingBatchConverter {
4849
public:
4950
/// Creates a converter for one file write cycle.
5051
/// Computes per-file K from context, builds physical schema, and constructs the converter.
@@ -59,12 +60,12 @@ class MapSharedShreddingBatchConverter {
5960
const std::shared_ptr<MemoryPool>& pool);
6061

6162
/// Returns the physical schema produced for this converter.
62-
const std::shared_ptr<arrow::Schema>& GetPhysicalSchema() const;
63+
const std::shared_ptr<arrow::Schema>& GetPhysicalSchema() const override;
6364

6465
/// Converts a logical batch to a physical batch.
6566
/// @param logical_batch Input ArrowArray (C ABI) with logical schema. Consumed on success.
6667
/// @return Owned physical ArrowArray (C ABI) with physical schema.
67-
Result<std::unique_ptr<ArrowArray>> Convert(ArrowArray* logical_batch);
68+
Result<std::unique_ptr<ArrowArray>> Convert(ArrowArray* logical_batch) override;
6869

6970
/// Builds MapSharedShreddingFieldMeta for one shredding column (by field name).
7071
/// Called at file close to serialize metadata.

0 commit comments

Comments
 (0)