This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 43b98b935f [core] Support plain Variant values in ROW tables (#10650)
43b98b935f is described below
commit 43b98b935f8d9ee09da2ff6babc45b9d00abafbd
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 18:48:44 2026 +0800
[core] Support plain Variant values in ROW tables (#10650)
---
docs/docs/multimodal-table/variant.mdx | 10 +-
docs/docs/pypaimon/data-types.md | 7 +-
.../apache/paimon/utils/NestedProjectedRow.java | 58 ++++++-
.../paimon/utils/NestedProjectedRowTest.java | 144 +++++++++++++++++
.../org/apache/paimon/AppendOnlyFileStore.java | 9 ++
.../java/org/apache/paimon/KeyValueFileStore.java | 4 +
.../apache/paimon/operation/RawFileSplitRead.java | 15 +-
.../org/apache/paimon/schema/SchemaValidation.java | 172 ++++++---------------
.../paimon/operation/RawFileSplitReadTest.java | 102 ++++++++++++
.../apache/paimon/schema/SchemaValidationTest.java | 95 +++---------
.../paimon/table/AppendOnlySimpleTableTest.java | 149 ++++++++++++++++++
.../flink/source/TestChangelogDataReadWrite.java | 4 +
paimon-python/pypaimon/read/split_read.py | 3 +-
.../pypaimon/tests/test_format_row_table.py | 98 ++++++++++++
14 files changed, 660 insertions(+), 210 deletions(-)
diff --git a/docs/docs/multimodal-table/variant.mdx
b/docs/docs/multimodal-table/variant.mdx
index 67491d50b4..bcce4b644a 100644
--- a/docs/docs/multimodal-table/variant.mdx
+++ b/docs/docs/multimodal-table/variant.mdx
@@ -37,7 +37,7 @@ instead of storing its JSON text. The data is parsed when it
is converted to `VA
can extract a path as a requested SQL type.
`VARIANT` does not require Data Evolution mode. SQL support requires Spark 4.0
or later, or Flink
-2.1 or later. Variant data files must use Parquet; ORC and Avro do not support
this type.
+2.1 or later. Variant data files can use Parquet or ROW; ORC and Avro do not
support this type.
## Variant vs. JSON String
@@ -121,6 +121,11 @@ The binary representation avoids storing and reparsing
JSON syntax. It also reta
each scalar value. Extracting a sub-field from a plain Variant still requires
reading the binary
`value` field for that Variant column.
+The [ROW format](../concepts/spec/rowformat.md) also supports plain Variant
values. Set
+`file.format` to `row` to store the same `value` and `metadata` bytes inside
compressed row blocks.
+ROW reads and decompresses the selected blocks in full; it does not support
Variant shredding or
+sub-field I/O pruning.
+
### Shredded Variant
Shredding materializes selected Variant paths as typed Parquet sub-columns
while keeping the
@@ -263,7 +268,8 @@ full value.
## Limitations
-- Variant data files must use Parquet. ORC and Avro are not supported.
+- Plain Variant data files can use Parquet or ROW. Variant shredding requires
Parquet. ORC and
+ Avro are not supported.
- `VARIANT` cannot be used as a primary key or partition key.
- Variant extraction does not by itself create an index or guarantee predicate
pushdown. Use
shredding to reduce sub-column I/O; use an appropriate index or modeled
table column when a path
diff --git a/docs/docs/pypaimon/data-types.md b/docs/docs/pypaimon/data-types.md
index 63893d72fe..63b8572063 100644
--- a/docs/docs/pypaimon/data-types.md
+++ b/docs/docs/pypaimon/data-types.md
@@ -72,6 +72,10 @@ Paimon supports two Parquet storage layouts for VARIANT:
- **Shredded VARIANT** — typed sub-columns are stored alongside overflow
bytes, enabling column-skipping
inside the Parquet file. Controlled by the `variant.shreddingSchema` table
option.
+Plain VARIANT also supports `file.format=row`. The ROW format stores the same
`value` and
+`metadata` bytes inside compressed row blocks and exposes the same Arrow
struct when read.
+ROW does not support Variant shredding or sub-field I/O pruning.
+
<Tabs groupId="variant-read-write">
<TabItem value="plain-variant" label="Plain VARIANT">
@@ -301,5 +305,6 @@ creates intermediate objects or extends arrays.
**Limitations:**
-- `VARIANT` is only supported with Parquet file format. ORC and Avro are not
supported.
+- Plain `VARIANT` supports Parquet and ROW file formats. Shredded `VARIANT`
requires Parquet.
+ ORC and Avro are not supported.
- `VARIANT` cannot be used as a primary key or partition key.
diff --git
a/paimon-common/src/main/java/org/apache/paimon/utils/NestedProjectedRow.java
b/paimon-common/src/main/java/org/apache/paimon/utils/NestedProjectedRow.java
index add679e76a..4961d9b924 100644
---
a/paimon-common/src/main/java/org/apache/paimon/utils/NestedProjectedRow.java
+++
b/paimon-common/src/main/java/org/apache/paimon/utils/NestedProjectedRow.java
@@ -21,16 +21,23 @@ package org.apache.paimon.utils;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.Blob;
import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalMap;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.InternalVector;
import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.data.variant.GenericVariant;
+import org.apache.paimon.data.variant.PaimonShreddingUtils;
+import org.apache.paimon.data.variant.PaimonShreddingUtils.FieldToExtract;
import org.apache.paimon.data.variant.Variant;
+import org.apache.paimon.data.variant.VariantMetadataUtils;
+import org.apache.paimon.data.variant.VariantSchema;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.MapType;
import org.apache.paimon.types.MultisetType;
import org.apache.paimon.types.RowKind;
@@ -54,17 +61,20 @@ public class NestedProjectedRow implements InternalRow {
@Nullable private final NestedProjectedRow[] nestedProjections;
@Nullable private final int[] nestedArity;
@Nullable private final ElementProjection[] elementProjections;
+ @Nullable private final VariantProjection[] variantProjections;
private InternalRow row;
private NestedProjectedRow(
int[] indexMapping,
@Nullable NestedProjectedRow[] nestedProjections,
@Nullable int[] nestedArity,
- @Nullable ElementProjection[] elementProjections) {
+ @Nullable ElementProjection[] elementProjections,
+ @Nullable VariantProjection[] variantProjections) {
this.indexMapping = indexMapping;
this.nestedProjections = nestedProjections;
this.nestedArity = nestedArity;
this.elementProjections = elementProjections;
+ this.variantProjections = variantProjections;
}
public NestedProjectedRow replaceRow(InternalRow row) {
@@ -93,6 +103,7 @@ public class NestedProjectedRow implements InternalRow {
NestedProjectedRow[] nestedProjections = null;
int[] nestedArity = null;
ElementProjection[] elementProjections = null;
+ VariantProjection[] variantProjections = null;
for (int i = 0; i < projectedSize; i++) {
DataField projected = projectedFields.get(i);
@@ -106,6 +117,15 @@ public class NestedProjectedRow implements InternalRow {
projected.name());
indexMapping[i] = dataIdx;
+ if (dataField.type().is(DataTypeRoot.VARIANT)
+ &&
VariantMetadataUtils.isVariantRowType(projected.type())) {
+ if (variantProjections == null) {
+ variantProjections = new VariantProjection[projectedSize];
+ }
+ variantProjections[i] = new VariantProjection((RowType)
projected.type());
+ continue;
+ }
+
DataTypeRoot typeRoot = projected.type().getTypeRoot();
if (typeRoot == DataTypeRoot.ROW) {
RowType dataNestedType = (RowType) dataField.type();
@@ -131,7 +151,11 @@ public class NestedProjectedRow implements InternalRow {
}
return new NestedProjectedRow(
- indexMapping, nestedProjections, nestedArity,
elementProjections);
+ indexMapping,
+ nestedProjections,
+ nestedArity,
+ elementProjections,
+ variantProjections);
}
@Nullable
@@ -318,6 +342,11 @@ public class NestedProjectedRow implements InternalRow {
@Override
public InternalRow getRow(int pos, int numFields) {
+ if (variantProjections != null && variantProjections[pos] != null) {
+ return row.isNullAt(indexMapping[pos])
+ ? null
+ :
variantProjections[pos].project(row.getVariant(indexMapping[pos]));
+ }
if (nestedProjections != null && nestedProjections[pos] != null) {
InternalRow inner = row.getRow(indexMapping[pos],
nestedArity[pos]);
return nestedProjections[pos].replaceRow(inner);
@@ -325,6 +354,31 @@ public class NestedProjectedRow implements InternalRow {
return row.getRow(indexMapping[pos], numFields);
}
+ /** Extracts requested fields from an ordinary Variant using the shared
cast semantics. */
+ private static class VariantProjection {
+
+ private final VariantSchema schema;
+ private final FieldToExtract[] fields;
+
+ private VariantProjection(RowType projectedType) {
+ this.schema =
+ PaimonShreddingUtils.buildVariantSchema(
+
PaimonShreddingUtils.variantShreddingSchema(DataTypes.VARIANT()));
+ this.fields =
PaimonShreddingUtils.getFieldsToExtract(projectedType, schema);
+ }
+
+ private InternalRow project(Variant variant) {
+ GenericVariant value =
+ variant instanceof GenericVariant
+ ? (GenericVariant) variant
+ : new GenericVariant(variant.valueBuffer(),
variant.metadataBuffer());
+ GenericRow physicalRow = new GenericRow(schema.numFields);
+ physicalRow.setField(schema.variantIdx, value.value());
+ physicalRow.setField(schema.topLevelMetadataIdx, value.metadata());
+ return PaimonShreddingUtils.assembleVariantStruct(physicalRow,
schema, fields);
+ }
+ }
+
// ======================== ElementProjection ========================
/**
diff --git
a/paimon-common/src/test/java/org/apache/paimon/utils/NestedProjectedRowTest.java
b/paimon-common/src/test/java/org/apache/paimon/utils/NestedProjectedRowTest.java
index 4a12756bb9..0c242a7b01 100644
---
a/paimon-common/src/test/java/org/apache/paimon/utils/NestedProjectedRowTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/utils/NestedProjectedRowTest.java
@@ -25,10 +25,15 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalMap;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.variant.BufferOnlyVariant;
+import org.apache.paimon.data.variant.GenericVariant;
+import org.apache.paimon.data.variant.GenericVariantBuilder;
+import org.apache.paimon.data.variant.VariantMetadataUtils;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.BooleanType;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.DoubleType;
import org.apache.paimon.types.FloatType;
import org.apache.paimon.types.IntType;
@@ -52,6 +57,145 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link NestedProjectedRow}. */
public class NestedProjectedRowTest {
+ @Test
+ void testVariantExtractionProjection() {
+ RowType dataSchema =
+ DataTypes.ROW(
+ DataTypes.FIELD(7, "id", DataTypes.INT()),
+ DataTypes.FIELD(41, "payload", DataTypes.VARIANT()));
+ RowType extractedType =
+ VariantMetadataUtils.VariantRowTypeBuilder.builder()
+ .field(DataTypes.BIGINT(), "$.a.b")
+ .field(DataTypes.INT(), "$.items[1]")
+ .field(DataTypes.STRING(), "$.missing.deep")
+ .field(DataTypes.INT(), "$.items[10]")
+ .field(DataTypes.BIGINT(), "$.wrong", false, "UTC")
+ .field(DataTypes.STRING(), "$.number")
+ .field(DataTypes.VARIANT(), "$.a")
+ .build();
+ RowType projectionSchema =
+ DataTypes.ROW(
+ DataTypes.FIELD(41, "payload", extractedType),
+ DataTypes.FIELD(7, "id", DataTypes.INT()));
+ NestedProjectedRow projection = NestedProjectedRow.create(dataSchema,
projectionSchema);
+ GenericVariant variant =
+ GenericVariant.fromJson(
+
"{\"a\":{\"b\":42},\"items\":[1,7],\"wrong\":\"invalid\",\"number\":123}");
+ InternalRow projected =
+ projection.replaceRow(GenericRow.of(5, new
BufferOnlyVariant(variant)));
+ assertThat(projected.getInt(1)).isEqualTo(5);
+ InternalRow extracted = projected.getRow(0,
extractedType.getFieldCount());
+ assertThat(extracted.getLong(0)).isEqualTo(42L);
+ assertThat(extracted.getInt(1)).isEqualTo(7);
+ assertThat(extracted.isNullAt(2)).isTrue();
+ assertThat(extracted.isNullAt(3)).isTrue();
+ assertThat(extracted.isNullAt(4)).isTrue();
+ assertThat(extracted.getString(5).toString()).isEqualTo("123");
+ assertThat(extracted.getVariant(6).toJson()).isEqualTo("{\"b\":42}");
+ }
+
+ @Test
+ void testVariantProjectionDistinguishesSqlNullAndVariantNull() {
+ RowType dataSchema = DataTypes.ROW(DataTypes.FIELD(0, "v",
DataTypes.VARIANT()));
+ RowType extractedType =
+ VariantMetadataUtils.VariantRowTypeBuilder.builder()
+ .field(DataTypes.STRING(), "$.field")
+ .field(DataTypes.VARIANT(), "$")
+ .build();
+ NestedProjectedRow projection =
+ NestedProjectedRow.create(
+ dataSchema, DataTypes.ROW(DataTypes.FIELD(0, "v",
extractedType)));
+ InternalRow sqlNull = projection.replaceRow(GenericRow.of((Object)
null));
+ assertThat(sqlNull.isNullAt(0)).isTrue();
+
+ InternalRow variantNull =
+
projection.replaceRow(GenericRow.of(GenericVariant.fromJson("null")));
+ assertThat(variantNull.isNullAt(0)).isFalse();
+ InternalRow extracted = variantNull.getRow(0, 2);
+ assertThat(extracted.isNullAt(0)).isTrue();
+ assertThat(extracted.isNullAt(1)).isFalse();
+ assertThat(extracted.getVariant(1).toJson()).isEqualTo("null");
+ }
+
+ @Test
+ void testNestedVariantProjectionCanBeReused() {
+ RowType nestedType =
+ DataTypes.ROW(
+ DataTypes.FIELD(8, "skipped", DataTypes.INT()),
+ DataTypes.FIELD(21, "v", DataTypes.VARIANT()));
+ RowType dataSchema =
+ DataTypes.ROW(
+ DataTypes.FIELD(1, "nested", nestedType),
+ DataTypes.FIELD(2, "unmodified", DataTypes.VARIANT()));
+ RowType extractedType =
+ VariantMetadataUtils.VariantRowTypeBuilder.builder()
+ .field(DataTypes.INT(), "$.x")
+ .build();
+ RowType projectedNested = DataTypes.ROW(DataTypes.FIELD(21, "v",
extractedType));
+ RowType projectedSchema =
+ DataTypes.ROW(
+ DataTypes.FIELD(2, "unmodified", DataTypes.VARIANT()),
+ DataTypes.FIELD(1, "nested", projectedNested));
+ NestedProjectedRow projection = NestedProjectedRow.create(dataSchema,
projectedSchema);
+ GenericVariant unchanged = GenericVariant.fromJson("[1,2]");
+ InternalRow first =
+ projection.replaceRow(
+ GenericRow.of(
+ GenericRow.of(100,
GenericVariant.fromJson("{\"x\":41}")),
+ unchanged));
+ assertThat(first.getVariant(0)).isSameAs(unchanged);
+ InternalRow firstExtracted = first.getRow(1, 1).getRow(0, 1);
+ assertThat(firstExtracted.getInt(0)).isEqualTo(41);
+ InternalRow second =
+ projection.replaceRow(
+ GenericRow.of(
+ GenericRow.of(200,
GenericVariant.fromJson("{\"x\":3}")),
+ unchanged));
+ assertThat(second.getRow(1, 1).getRow(0, 1).getInt(0)).isEqualTo(3);
+ assertThat(firstExtracted.getInt(0)).isEqualTo(41);
+ assertThat(projection.replaceRow(GenericRow.of(null,
unchanged)).isNullAt(1)).isTrue();
+ InternalRow nullVariant =
+ projection.replaceRow(GenericRow.of(GenericRow.of(0, null),
unchanged));
+ assertThat(nullVariant.getRow(1, 1).isNullAt(0)).isTrue();
+ }
+
+ @Test
+ void testVariantProjectionPreservesStrictCastPolicy() {
+ RowType dataSchema = DataTypes.ROW(DataTypes.FIELD(0, "v",
DataTypes.VARIANT()));
+ RowType extractedType =
+ VariantMetadataUtils.VariantRowTypeBuilder.builder()
+ .field(DataTypes.INT(), "$.x")
+ .build();
+ NestedProjectedRow projection =
+ NestedProjectedRow.create(
+ dataSchema, DataTypes.ROW(DataTypes.FIELD(0, "v",
extractedType)));
+ InternalRow projected =
+ projection.replaceRow(
+
GenericRow.of(GenericVariant.fromJson("{\"x\":\"invalid\"}")));
+ assertThatThrownBy(() -> projected.getRow(0, 1))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Invalid cast")
+ .hasMessageContaining("INT");
+ }
+
+ @Test
+ void testVariantProjectionUsesRequestedTimeZone() {
+ RowType dataSchema = DataTypes.ROW(DataTypes.FIELD(0, "v",
DataTypes.VARIANT()));
+ RowType extractedType =
+ VariantMetadataUtils.VariantRowTypeBuilder.builder()
+ .field(DataTypes.STRING(), "$", true, "UTC")
+ .field(DataTypes.STRING(), "$", true, "Asia/Shanghai")
+ .build();
+ NestedProjectedRow projection =
+ NestedProjectedRow.create(
+ dataSchema, DataTypes.ROW(DataTypes.FIELD(0, "v",
extractedType)));
+ GenericVariantBuilder builder = new GenericVariantBuilder(false);
+ builder.appendTimestamp(0);
+ InternalRow extracted =
projection.replaceRow(GenericRow.of(builder.result())).getRow(0, 2);
+ assertThat(extracted.getString(0).toString()).isEqualTo("1970-01-01
00:00:00");
+ assertThat(extracted.getString(1).toString()).isEqualTo("1970-01-01
08:00:00");
+ }
+
@Test
void testReturnNullWhenSchemasAreEqual() {
RowType schema =
diff --git
a/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
index f0f5e1c6ff..b60e91906a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
@@ -74,6 +74,15 @@ public class AppendOnlyFileStore extends
AbstractFileStore<InternalRow> {
schemaManager,
schema,
rowType,
+ dataSchema -> {
+ if (options.rowTrackingEnabled()) {
+ // The row id and sequence number may be in the
manifest entry.
+ return SpecialFields.rowTypeWithRowTracking(
+ dataSchema.logicalRowType(), true,
true)
+ .getFields();
+ }
+ return dataSchema.fields();
+ },
FileFormatDiscover.of(options),
pathFactory(),
options);
diff --git a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
index 8f84f5e504..2f99d2367e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
@@ -120,6 +120,10 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
schemaManager,
schema,
valueType,
+ dataSchema ->
+ KeyValue.createKeyValueFields(
+ keyValueFieldsExtractor.keyFields(dataSchema),
+
keyValueFieldsExtractor.valueFields(dataSchema)),
FileFormatDiscover.of(options),
pathFactory(),
options);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
index 84b32eaa1d..ea652001a0 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
@@ -51,6 +51,7 @@ import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.IncrementalSplit;
import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.FormatReaderMapping;
@@ -68,9 +69,9 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.function.Function;
import static org.apache.paimon.predicate.PredicateBuilder.splitAnd;
-import static org.apache.paimon.table.SpecialFields.rowTypeWithRowTracking;
/** A {@link SplitRead} to read raw file directly from {@link DataSplit}. */
public class RawFileSplitRead implements SplitRead<InternalRow> {
@@ -80,6 +81,7 @@ public class RawFileSplitRead implements
SplitRead<InternalRow> {
private final FileIO fileIO;
private final SchemaManager schemaManager;
private final TableSchema schema;
+ private final Function<TableSchema, List<DataField>> fieldsExtractor;
private final FileFormatDiscover formatDiscover;
private final FileStorePathFactory pathFactory;
private final Map<FormatKey, FormatReaderMapping> formatReaderMappings;
@@ -100,12 +102,14 @@ public class RawFileSplitRead implements
SplitRead<InternalRow> {
SchemaManager schemaManager,
TableSchema schema,
RowType rowType,
+ Function<TableSchema, List<DataField>> fieldsExtractor,
FileFormatDiscover formatDiscover,
FileStorePathFactory pathFactory,
CoreOptions coreOptions) {
this.fileIO = fileIO;
this.schemaManager = schemaManager;
this.schema = schema;
+ this.fieldsExtractor = fieldsExtractor;
this.formatDiscover = formatDiscover;
this.pathFactory = pathFactory;
this.formatReaderMappings = new HashMap<>();
@@ -263,14 +267,7 @@ public class RawFileSplitRead implements
SplitRead<InternalRow> {
return new Builder(
formatDiscover,
outputRowType.getFields(),
- schema -> {
- if (rowTrackingEnabled) {
- // maybe file has no row id and sequence number, but
in manifest entry
- return rowTypeWithRowTracking(schema.logicalRowType(),
true, true)
- .getFields();
- }
- return schema.fields();
- },
+ fieldsExtractor,
filters,
pushDownTopN,
pushDownLimit,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index ab73ca53f2..c98600f7ee 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -256,8 +256,7 @@ public class SchemaValidation {
FileFormat fileFormat =
FileFormat.fromIdentifier(options.formatType(), new
Options(schema.options()));
RowType tableRowType = new RowType(schema.fields());
- validateGeospatialTypes(schema, options, tableRowType);
- validateVariantTypes(schema, options);
+ validateIcebergGeospatialTypes(tableRowType, options);
validateIcebergTimestampPrecisions(tableRowType, options);
validateIcebergTimePrecisions(tableRowType, options);
validateIcebergPublishableTypes(tableRowType, options);
@@ -417,6 +416,24 @@ public class SchemaValidation {
validateRowTracking(schema, options);
+ List<String> geospatialClusteringColumns =
+ schema.fields().stream()
+ .filter(field ->
options.clusteringColumns().contains(field.name()))
+ .filter(
+ field ->
+ containsType(
+ field.type(),
+ type ->
+ type.isAnyOf(
+
DataTypeRoot.GEOMETRY,
+
DataTypeRoot.GEOGRAPHY)))
+ .map(DataField::name)
+ .collect(Collectors.toList());
+ checkArgument(
+ geospatialClusteringColumns.isEmpty(),
+ "Geometry and geography columns cannot be clustering columns:
%s.",
+ geospatialClusteringColumns);
+
validateIncrementalClustering(schema, options);
validateChainTable(schema, options);
@@ -500,125 +517,6 @@ public class SchemaValidation {
}
}
- /**
- * Only the Parquet format can store {@code VARIANT}, so every format the
table may write data
- * files with has to be parquet: the main format, each per-level override
and the changelog
- * format. Checking here reports the column and the option at DDL time,
rather than the bare
- * "unsupported type" a non-parquet writer raises on the first row.
- */
- private static void validateVariantTypes(TableSchema schema, CoreOptions
options) {
- List<String> variantColumns =
- schema.fields().stream()
- .filter(
- field ->
- containsType(
- field.type(),
- type ->
type.is(DataTypeRoot.VARIANT)))
- .map(DataField::name)
- .collect(Collectors.toList());
- if (variantColumns.isEmpty()) {
- return;
- }
-
- checkArgument(
- CoreOptions.FILE_FORMAT_PARQUET.equals(options.formatType()),
- "Variant columns %s require '%s'='parquet', but was '%s'.",
- variantColumns,
- CoreOptions.FILE_FORMAT.key(),
- options.formatType());
- options.fileFormatPerLevel()
- .forEach(
- (level, format) ->
- checkArgument(
-
CoreOptions.FILE_FORMAT_PARQUET.equals(format),
- "Variant columns %s require parquet at
every level, but '%s' contains '%s:%s'.",
- variantColumns,
-
CoreOptions.FILE_FORMAT_PER_LEVEL.key(),
- level,
- format));
- checkArgument(
- options.changelogFileFormat() == null
- ||
CoreOptions.FILE_FORMAT_PARQUET.equals(options.changelogFileFormat()),
- "Variant columns %s require '%s' to be parquet, but was '%s'.",
- variantColumns,
- CoreOptions.CHANGELOG_FILE_FORMAT.key(),
- options.changelogFileFormat());
- }
-
- private static void validateGeospatialTypes(
- TableSchema schema, CoreOptions options, RowType rowType) {
- boolean hasGeospatial =
- containsType(
- rowType,
- type -> type.isAnyOf(DataTypeRoot.GEOMETRY,
DataTypeRoot.GEOGRAPHY));
- if (!hasGeospatial) {
- return;
- }
-
- checkArgument(
- CoreOptions.FILE_FORMAT_PARQUET.equals(options.formatType()),
- "Geometry and geography columns require '%s'='parquet', but
was '%s'.",
- CoreOptions.FILE_FORMAT.key(),
- options.formatType());
- options.fileFormatPerLevel()
- .forEach(
- (level, format) ->
- checkArgument(
-
CoreOptions.FILE_FORMAT_PARQUET.equals(format),
- "Geometry and geography columns
require parquet at every level, but '%s' contains '%s:%s'.",
-
CoreOptions.FILE_FORMAT_PER_LEVEL.key(),
- level,
- format));
- checkArgument(
- options.changelogFileFormat() == null
- ||
CoreOptions.FILE_FORMAT_PARQUET.equals(options.changelogFileFormat()),
- "Geometry and geography columns require '%s' to be parquet,
but was '%s'.",
- CoreOptions.CHANGELOG_FILE_FORMAT.key(),
- options.changelogFileFormat());
- validateIcebergGeospatialTypes(rowType, options);
-
- List<String> geospatialClusteringColumns =
- schema.fields().stream()
- .filter(field ->
options.clusteringColumns().contains(field.name()))
- .filter(
- field ->
- containsType(
- field.type(),
- type ->
- type.isAnyOf(
-
DataTypeRoot.GEOMETRY,
-
DataTypeRoot.GEOGRAPHY)))
- .map(DataField::name)
- .collect(Collectors.toList());
- checkArgument(
- geospatialClusteringColumns.isEmpty(),
- "Geometry and geography columns cannot be clustering columns:
%s.",
- geospatialClusteringColumns);
-
- Set<String> geospatialFields =
- schema.fields().stream()
- .filter(
- field ->
- field.type()
- .isAnyOf(
- DataTypeRoot.GEOMETRY,
-
DataTypeRoot.GEOGRAPHY))
- .map(DataField::name)
- .collect(Collectors.toSet());
- Set<String> geospatialBucketKeys = new HashSet<>(schema.bucketKeys());
- geospatialBucketKeys.retainAll(geospatialFields);
- checkArgument(
- geospatialBucketKeys.isEmpty(),
- "Geometry and geography columns cannot be bucket keys: %s.",
- geospatialBucketKeys);
- Set<String> geospatialSequenceFields = new
HashSet<>(options.sequenceField());
- geospatialSequenceFields.retainAll(geospatialFields);
- checkArgument(
- geospatialSequenceFields.isEmpty(),
- "Geometry and geography columns cannot be sequence fields:
%s.",
- geospatialSequenceFields);
- }
-
/**
* Refuses the timestamp precisions the Iceberg mirror cannot publish. A
higher precision is
* written as Parquet INT96, which Iceberg reads as a microsecond zoned
timestamp rather than
@@ -1603,6 +1501,22 @@ public class SchemaValidation {
schema.primaryKeys(), schema.partitionKeys()));
}
}
+
+ Set<String> geospatialSequenceFields =
+ schema.fields().stream()
+ .filter(field -> sequenceField.contains(field.name()))
+ .filter(
+ field ->
+ field.type()
+ .isAnyOf(
+ DataTypeRoot.GEOMETRY,
+
DataTypeRoot.GEOGRAPHY))
+ .map(DataField::name)
+ .collect(Collectors.toSet());
+ checkArgument(
+ geospatialSequenceFields.isEmpty(),
+ "Geometry and geography columns cannot be sequence fields:
%s.",
+ geospatialSequenceFields);
}
private static void validateBucket(TableSchema schema, CoreOptions
options) {
@@ -1650,6 +1564,22 @@ public class SchemaValidation {
}
}
}
+
+ Set<String> geospatialBucketKeys =
+ schema.fields().stream()
+ .filter(field ->
schema.bucketKeys().contains(field.name()))
+ .filter(
+ field ->
+ field.type()
+ .isAnyOf(
+ DataTypeRoot.GEOMETRY,
+
DataTypeRoot.GEOGRAPHY))
+ .map(DataField::name)
+ .collect(Collectors.toSet());
+ checkArgument(
+ geospatialBucketKeys.isEmpty(),
+ "Geometry and geography columns cannot be bucket keys: %s.",
+ geospatialBucketKeys);
}
private static boolean isPostponeBucketTable(TableSchema schema, int
bucket) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/RawFileSplitReadTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/RawFileSplitReadTest.java
index 048fb5fb0a..1e8e462db4 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/RawFileSplitReadTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/RawFileSplitReadTest.java
@@ -32,10 +32,12 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.ReadBatchSizer;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.FileSystemSchemaManager;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.SchemaUtils;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
@@ -45,19 +47,26 @@ import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.IOExceptionSupplier;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.apache.paimon.predicate.PredicateBuilder.and;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link RawFileSplitRead}. */
@@ -65,6 +74,98 @@ class RawFileSplitReadTest {
@TempDir java.nio.file.Path tempDir;
+ @ParameterizedTest
+ @CsvSource({
+ "row, false",
+ "avro, false",
+ "avro, true",
+ "parquet, false",
+ "parquet, true",
+ "orc, false",
+ "orc, true"
+ })
+ void testRawConvertibleReadUsesKeyValueSchema(String format, boolean
thinMode)
+ throws Exception {
+ Path tablePath = new Path(tempDir.resolve("primary-key-raw").toUri());
+ Options options = new Options();
+ options.set(CoreOptions.PATH, tablePath.toString());
+ options.set(CoreOptions.BUCKET, 1);
+ options.set(CoreOptions.WRITE_ONLY, true);
+ options.set(CoreOptions.FILE_FORMAT, format);
+ options.set(CoreOptions.DATA_FILE_THIN_MODE, thinMode);
+ Schema schema =
+ Schema.newBuilder()
+ .column("pt", DataTypes.INT())
+ .column("a", DataTypes.INT())
+ .column("b", DataTypes.BIGINT())
+ .partitionKeys("pt")
+ .primaryKey("pt", "a")
+ .options(options.toMap())
+ .build();
+ TableSchema tableSchema =
+ SchemaUtils.forceCommit(
+ new FileSystemSchemaManager(LocalFileIO.create(),
tablePath), schema);
+ FileStoreTable table =
+ FileStoreTableFactory.create(LocalFileIO.create(), tablePath,
tableSchema);
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(GenericRow.of(1, 10, 100L));
+ write.write(GenericRow.of(1, 20, 200L));
+ write.write(GenericRow.of(1, 30, 300L));
+ write.write(GenericRow.of(1, 40, 400L));
+ write.write(GenericRow.of(2, 20, 220L));
+ commit.commit(write.prepareCommit());
+ }
+
+ List<DataSplit> dataSplits =
table.newSnapshotReader().read().dataSplits();
+ assertThat(dataSplits).hasSize(2);
+ for (DataSplit split : dataSplits) {
+ assertThat(split.rawConvertible()).isTrue();
+ assertThat(split.dataFiles()).hasSize(1);
+ }
+ List<Split> splits = new ArrayList<>(dataSplits);
+ for (int schemaVersion = 0; schemaVersion < 2; schemaVersion++) {
+ RowType readType =
+ new RowType(
+ Arrays.asList(
+ table.rowType().getField(2),
+ table.rowType().getField(1),
+ table.rowType().getField(0)));
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ TableRead read =
+ table.newRead()
+ .withReadType(readType)
+ .withFilter(
+ and(
+ builder.equal(0, 1),
+ builder.greaterOrEqual(1, 20),
+ builder.lessOrEqual(2, 300L)))
+ .executeFilter();
+ List<String> actual = new ArrayList<>();
+ try (RecordReader<InternalRow> reader = read.createReader(splits))
{
+ reader.forEachRemaining(
+ row -> {
+ assertThat(row.getFieldCount()).isEqualTo(3);
+
assertThat(row.getRowKind()).isEqualTo(RowKind.INSERT);
+ actual.add(row.getLong(0) + "|" + row.getInt(1) +
"|" + row.getInt(2));
+ });
+ }
+ assertThat(actual).containsExactlyInAnyOrder("200|20|1",
"300|30|1");
+
+ if (schemaVersion == 0) {
+ table.schemaManager()
+ .commitChanges(
+ Arrays.asList(
+ SchemaChange.renameColumn("a",
"renamed_a"),
+ SchemaChange.renameColumn("b",
"renamed_b")));
+ table = table.copyWithLatestSchema();
+ assertThat(dataSplits.get(0).dataFiles().get(0).schemaId())
+ .isLessThan(table.schema().id());
+ }
+ }
+ }
+
@Test
void testReaderMappingIsNotSharedBetweenReadTypes() throws Exception {
FileStoreTable table = createTable("mapping-cache");
@@ -117,6 +218,7 @@ class RawFileSplitReadTest {
table.schemaManager(),
table.schema(),
table.rowType(),
+ TableSchema::fields,
ignored -> countingFormat,
table.store().pathFactory(),
table.coreOptions());
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index 50bf9e8006..54a7088950 100644
---
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -1843,35 +1843,18 @@ class SchemaValidationTest {
emptyList(),
new HashMap<>())));
- Map<String, String> avroOptions = new HashMap<>();
- avroOptions.put(CoreOptions.FILE_FORMAT.key(), "avro");
- assertThatThrownBy(
- () ->
- validateTableSchema(
- geospatialSchema(
- fields, emptyList(),
emptyList(), avroOptions)))
- .hasMessageContaining("require 'file.format'='parquet'");
-
- Map<String, String> perLevelOptions = new HashMap<>();
- perLevelOptions.put(CoreOptions.FILE_FORMAT_PER_LEVEL.key(), "0:orc");
- assertThatThrownBy(
- () ->
- validateTableSchema(
- geospatialSchema(
- fields, emptyList(),
emptyList(), perLevelOptions)))
- .hasMessageContaining("require parquet at every level");
-
- Map<String, String> changelogOptions = new HashMap<>();
- changelogOptions.put(CoreOptions.CHANGELOG_FILE_FORMAT.key(), "orc");
- assertThatThrownBy(
- () ->
- validateTableSchema(
- geospatialSchema(
- fields,
- emptyList(),
- emptyList(),
- changelogOptions)))
- .hasMessageContaining("require 'changelog-file.format' to be
parquet");
+ for (String format : Arrays.asList("parquet", "row")) {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.FILE_FORMAT.key(), format);
+ options.put(CoreOptions.FILE_FORMAT_PER_LEVEL.key(),
"0:orc,1:avro");
+ options.put(CoreOptions.CHANGELOG_FILE_FORMAT.key(), "orc");
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ validateTableSchema(
+ geospatialSchema(
+ fields, emptyList(),
emptyList(), options)));
+ }
Map<String, String> icebergV2Options = new HashMap<>();
icebergV2Options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
"table-location");
@@ -2013,9 +1996,9 @@ class SchemaValidationTest {
"Geometry and geography columns cannot be clustering
columns: [nested].");
}
- @Test
- public void testVariantTypeValidation() {
- // a top-level variant and one nested inside a struct
+ @ParameterizedTest
+ @ValueSource(strings = {"parquet", "row"})
+ public void testVariantTypeValidation(String format) {
List<DataField> fields =
Arrays.asList(
new DataField(0, "id", DataTypes.INT()),
@@ -2024,56 +2007,20 @@ class SchemaValidationTest {
2,
"nested",
DataTypes.ROW(DataTypes.FIELD(3, "inner",
DataTypes.VARIANT()))));
-
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.FILE_FORMAT.key(), format);
+ options.put(CoreOptions.FILE_FORMAT_PER_LEVEL.key(), "0:orc,1:avro");
+ options.put(CoreOptions.CHANGELOG_PRODUCER.key(), "input");
+ options.put(CoreOptions.CHANGELOG_FILE_FORMAT.key(), "avro");
assertThatNoException()
.isThrownBy(
- () ->
- validateTableSchema(
- geospatialSchema(
- fields,
- emptyList(),
- emptyList(),
- new HashMap<>())));
-
- // the main format used to be caught by the format's own type
converter, with a bare
- // "Unsupported type: VARIANT" that named neither the column nor the
option
- Map<String, String> orcOptions = new HashMap<>();
- orcOptions.put(CoreOptions.FILE_FORMAT.key(), "orc");
- assertThatThrownBy(
- () ->
- validateTableSchema(
- geospatialSchema(
- fields, emptyList(),
emptyList(), orcOptions)))
- .hasMessageContaining(
- "Variant columns [v, nested] require
'file.format'='parquet', but was 'orc'");
-
- // these two used to pass DDL and fail on the first row written with
that format
- Map<String, String> perLevelOptions = new HashMap<>();
- perLevelOptions.put(CoreOptions.FILE_FORMAT_PER_LEVEL.key(), "0:orc");
- assertThatThrownBy(
() ->
validateTableSchema(
geospatialSchema(
fields,
emptyList(),
singletonList("id"),
- perLevelOptions)))
- .hasMessageContaining(
- "Variant columns [v, nested] require parquet at every
level, but 'file.format.per.level' contains '0:orc'");
-
- Map<String, String> changelogOptions = new HashMap<>();
- changelogOptions.put(CoreOptions.CHANGELOG_PRODUCER.key(), "input");
- changelogOptions.put(CoreOptions.CHANGELOG_FILE_FORMAT.key(), "avro");
- assertThatThrownBy(
- () ->
- validateTableSchema(
- geospatialSchema(
- fields,
- emptyList(),
- singletonList("id"),
- changelogOptions)))
- .hasMessageContaining(
- "Variant columns [v, nested] require
'changelog-file.format' to be parquet, but was 'avro'");
+ options)));
}
private TableSchema geospatialSchema(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
index ce4bd35f3e..df9323bf46 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java
@@ -28,6 +28,8 @@ import org.apache.paimon.data.GenericMap;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.data.variant.GenericVariant;
+import org.apache.paimon.data.variant.VariantMetadataUtils;
import org.apache.paimon.fileindex.FileIndexOptions;
import org.apache.paimon.fileindex.bitmap.BitmapFileIndexFactory;
import org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndexFactory;
@@ -36,6 +38,7 @@ import
org.apache.paimon.fileindex.rangebitmap.RangeBitmapFileIndexFactory;
import org.apache.paimon.fs.FileIOFinder;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.io.BundleRecords;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileEntry;
@@ -80,6 +83,7 @@ import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.BranchMergeHandler;
import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringBitmap32;
import org.apache.commons.math3.random.RandomDataGenerator;
@@ -135,6 +139,151 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
/** Tests for {@link AppendOnlyFileStoreTable}. */
public class AppendOnlySimpleTableTest extends SimpleTableTestBase {
+ @ParameterizedTest
+ @ValueSource(strings = {"append", "primary-key-raw", "primary-key-merge"})
+ public void testRowFormatVariantReadWrite(String tableKind) throws
Exception {
+ boolean primaryKey = !"append".equals(tableKind);
+ boolean merge = "primary-key-merge".equals(tableKind);
+ RowType rowType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "payload", DataTypes.VARIANT()),
+ DataTypes.FIELD(
+ 2,
+ "nested",
+ DataTypes.ROW(
+ DataTypes.FIELD(3, "label",
DataTypes.STRING()),
+ DataTypes.FIELD(4, "detail",
DataTypes.VARIANT()))));
+ Options options = new Options();
+ options.set(BUCKET, primaryKey ? 1 : -1);
+ options.set(FILE_FORMAT, "row");
+ options.set(WRITE_ONLY, true);
+ new FileSystemSchemaManager(FileIOFinder.find(tablePath), tablePath)
+ .createTable(
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ primaryKey
+ ? Collections.singletonList("id")
+ : Collections.emptyList(),
+ options.toMap(),
+ ""));
+ FileStoreTable table =
+ FileStoreTableFactory.create(FileIOFinder.find(tablePath),
tablePath);
+ List<InternalRow> expected =
+ Arrays.asList(
+ GenericRow.of(
+ 1,
+ GenericVariant.fromJson(
+
"{\"a\":[1,\"two\",null],\"b\":{\"c\":true}}"),
+ GenericRow.of(
+ BinaryString.fromString("one"),
+
GenericVariant.fromJson("[1,{\"x\":2}]"))),
+ GenericRow.of(
+ 2, null, GenericRow.of(null,
GenericVariant.fromJson("null"))),
+ GenericRow.of(3, GenericVariant.fromJson("null"),
null),
+ GenericRow.of(
+ 4,
+ GenericVariant.fromJson("\"text\""),
+ GenericRow.of(BinaryString.fromString("two"),
null)));
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ for (int commitIndex = 0; commitIndex < (merge ? 2 : 1);
commitIndex++) {
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ for (InternalRow row : expected) {
+ write.write(
+ merge && commitIndex == 0
+ ? GenericRow.of(
+ row.getInt(0),
+
GenericVariant.fromJson("{\"a\":[-1]}"),
+ null)
+ : row);
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ ReadBuilder readBuilder = table.newReadBuilder();
+ List<Split> splits = readBuilder.newScan().plan().splits();
+ if (primaryKey) {
+ assertThat(splits).hasSize(1);
+ DataSplit split = (DataSplit) splits.get(0);
+ assertThat(split.rawConvertible()).isEqualTo(!merge);
+ assertThat(split.dataFiles()).hasSize(merge ? 2 : 1);
+ }
+ InternalRowSerializer serializer = new InternalRowSerializer(rowType);
+ List<InternalRow> actual = new ArrayList<>();
+ for (Split split : splits) {
+ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(split)) {
+ reader.forEachRemaining(row ->
actual.add(serializer.copy(row)));
+ }
+ }
+ assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
+
+ RowType payloadProjection =
+ VariantMetadataUtils.VariantRowTypeBuilder.builder()
+ .field(DataTypes.INT(), "$.a[0]", false, "UTC")
+ .field(DataTypes.BOOLEAN(), "$.b.c", false, "UTC")
+ .build();
+ RowType detailProjection =
+ VariantMetadataUtils.VariantRowTypeBuilder.builder()
+ .field(DataTypes.INT(), "$[1].x", false, "UTC")
+ .build();
+ RowType readType =
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ 2,
+ "nested",
+ DataTypes.ROW(DataTypes.FIELD(4, "detail",
detailProjection))),
+ DataTypes.FIELD(1, "payload", payloadProjection),
+ DataTypes.FIELD(0, "id", DataTypes.INT()));
+ ReadBuilder projectedRead =
table.newReadBuilder().withReadType(readType);
+ InternalRowSerializer projectedSerializer = new
InternalRowSerializer(readType);
+ List<InternalRow> projected = new ArrayList<>();
+ for (Split split : projectedRead.newScan().plan().splits()) {
+ try (RecordReader<InternalRow> reader =
projectedRead.newRead().createReader(split)) {
+ reader.forEachRemaining(row ->
projected.add(projectedSerializer.copy(row)));
+ }
+ }
+ assertThat(projected)
+ .containsExactlyInAnyOrder(
+ GenericRow.of(GenericRow.of(GenericRow.of(2)),
GenericRow.of(1, true), 1),
+ GenericRow.of(GenericRow.of(GenericRow.of((Object)
null)), null, 2),
+ GenericRow.of(null, GenericRow.of(null, null), 3),
+ GenericRow.of(GenericRow.of((Object) null),
GenericRow.of(null, null), 4));
+
+ if ("primary-key-raw".equals(tableKind)) {
+ new FileSystemSchemaManager(FileIOFinder.find(tablePath),
tablePath)
+ .commitChanges(
+ Arrays.asList(
+ SchemaChange.renameColumn("id",
"renamed_id"),
+ SchemaChange.renameColumn("payload",
"renamed_payload")));
+ FileStoreTable evolvedTable =
+ FileStoreTableFactory.create(FileIOFinder.find(tablePath),
tablePath);
+ RowType evolvedReadType =
+ DataTypes.ROW(
+ readType.getField(2),
+ readType.getField(1).newName("renamed_payload"),
+ readType.getField(0).newName("renamed_id"));
+ ReadBuilder evolvedRead =
evolvedTable.newReadBuilder().withReadType(evolvedReadType);
+ IndexedSplit selected =
+ new IndexedSplit(
+ (DataSplit) splits.get(0),
+ Arrays.asList(new Range(0, 0), new Range(2, 2)),
+ null);
+ InternalRowSerializer evolvedSerializer = new
InternalRowSerializer(evolvedReadType);
+ List<InternalRow> historical = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
evolvedRead.newRead().createReader(selected)) {
+ reader.forEachRemaining(row ->
historical.add(evolvedSerializer.copy(row)));
+ }
+ assertThat(historical)
+ .containsExactlyInAnyOrder(
+ GenericRow.of(
+ GenericRow.of(GenericRow.of(2)),
GenericRow.of(1, true), 1),
+ GenericRow.of(null, GenericRow.of(null, null), 3));
+ }
+ }
+
@Test
public void testOverwriteSameFiles() throws Exception {
FileStoreTable table = createFileStoreTable();
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
index 920f297028..5e5eb53015 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
@@ -150,6 +150,10 @@ public class TestChangelogDataReadWrite {
schemaManager,
schema,
VALUE_TYPE,
+ dataSchema ->
+ KeyValue.createKeyValueFields(
+ EXTRACTOR.keyFields(dataSchema),
+ EXTRACTOR.valueFields(dataSchema)),
FileFormatDiscover.of(options),
pathFactory,
options);
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index a50deffb09..bc8d30206f 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -95,7 +95,8 @@ from pypaimon.globalindex.indexed_split import IndexedSplit
from pypaimon.utils.data_evolution_utils import retrieve_anchor_file
KEY_PREFIX = "_KEY_"
-KEY_FIELD_ID_START = 1000000
+# Match Java's reserved key-field range without overlapping valid user IDs.
+KEY_FIELD_ID_START = SpecialFields.SYSTEM_FIELD_ID_START
NULL_FIELD_INDEX = -1
diff --git a/paimon-python/pypaimon/tests/test_format_row_table.py
b/paimon-python/pypaimon/tests/test_format_row_table.py
index c483144547..b2598fa78c 100644
--- a/paimon-python/pypaimon/tests/test_format_row_table.py
+++ b/paimon-python/pypaimon/tests/test_format_row_table.py
@@ -23,8 +23,10 @@ import tempfile
import unittest
import pyarrow as pa
+import pytest
from pypaimon import CatalogFactory, Schema
+from pypaimon.data.generic_variant import GenericVariant
class RowFormatAppendOnlyTest(unittest.TestCase):
@@ -499,5 +501,101 @@ class RowFormatVectorTableTest(unittest.TestCase):
self.assertEqual(sorted(result.column('id').to_pylist()), [1, 2, 3])
[email protected]_read
[email protected]_write
[email protected]('table_kind', [
+ 'append', 'primary-key', 'primary-key-high-id', 'data-evolution',
+])
+def test_row_table_preserves_variant_values(tmp_path, table_kind):
+ variants = GenericVariant.to_arrow_array([
+ GenericVariant.from_python({'a': 1, 'nested': {'b': [True, None,
'text']}}),
+ GenericVariant.from_python([1, {'x': 'value'}, None]),
+ GenericVariant.from_python(None),
+ None,
+ GenericVariant.from_python('scalar'),
+ ])
+ payloads = variants.to_pylist()
+ nested_type = pa.struct([pa.field('payload', variants.type),
pa.field('tag', pa.string())])
+ data = pa.table({
+ 'id': pa.array(range(5), type=pa.int32()),
+ 'payload': variants,
+ 'nested': pa.array([
+ {'payload': payloads[0], 'tag': 'object'},
+ {'payload': payloads[1], 'tag': 'array'},
+ {'payload': payloads[2], 'tag': 'variant-null'},
+ None,
+ {'payload': None, 'tag': 'sql-null'},
+ ], type=nested_type),
+ 'items': pa.array([
+ [payloads[0], None, payloads[2]], [], None, [payloads[1]],
[payloads[4]],
+ ], type=pa.list_(variants.type)),
+ 'mapping': pa.array([
+ [('first', payloads[0]), ('missing', None)], [], None,
+ [('array', payloads[1])], [('null', payloads[2])],
+ ], type=pa.map_(pa.string(), variants.type)),
+ })
+ options = {
+ 'file.format': 'row', 'write.native.enabled': 'false',
'read.native.enabled': 'false',
+ }
+ primary_keys = []
+ if table_kind.startswith('primary-key'):
+ primary_keys = ['id']
+ options['bucket'] = '1'
+ elif table_kind == 'data-evolution':
+ options.update({'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true'})
+
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'warehouse')})
+ catalog.create_database('default', True)
+ schema = Schema.from_pyarrow_schema(
+ data.schema, primary_keys=primary_keys, options=options)
+ if table_kind == 'primary-key-high-id':
+ schema.fields[1].id = 1_000_000
+ catalog.create_table('default.variants', schema, False)
+ table = catalog.get_table('default.variants')
+ batches = [data]
+ if table_kind == 'primary-key-high-id':
+ stale = GenericVariant.to_arrow_array([
+ GenericVariant.from_python({'stale': True}) for _ in range(5)
+ ])
+ batches.insert(0, data.set_column(1, 'payload', stale))
+ builder = table.new_batch_write_builder()
+ for batch in batches:
+ writer, committer = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(batch)
+ messages = writer.prepare_commit()
+ files = [file for message in messages for file in
message.new_files]
+ assert files
+ for file in files:
+ assert file.file_name.endswith('.row')
+ with table.file_io.new_input_stream(file.file_path) as stream:
+ stream.seek(file.file_size - 8)
+ assert stream.read(1) == b'\x01'
+ committer.commit(messages)
+ finally:
+ writer.close()
+ committer.close()
+
+ read_builder = table.new_read_builder()
+ splits = read_builder.new_scan().plan().splits()
+ result = read_builder.new_read().to_arrow(splits).sort_by('id')
+ assert result.to_pydict() == data.to_pydict()
+ restored_payloads = result.column('payload').to_pylist()
+ assert GenericVariant.from_arrow_struct(restored_payloads[0]).to_python()
== {
+ 'a': 1, 'nested': {'b': [True, None, 'text']},
+ }
+ assert GenericVariant.from_arrow_struct(restored_payloads[1]).to_python()
== [
+ 1, {'x': 'value'}, None,
+ ]
+ assert restored_payloads[2] is not None
+ assert GenericVariant.from_arrow_struct(restored_payloads[2]).to_python()
is None
+ assert restored_payloads[3] is None
+
+ projected = table.new_read_builder().with_projection(['payload', 'id'])
+ projected_result = projected.new_read().to_arrow(
+ projected.new_scan().plan().splits()).sort_by('id')
+ assert projected_result.to_pydict() == data.select(['payload',
'id']).to_pydict()
+
+
if __name__ == '__main__':
unittest.main()
