This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit fa8ae00346866df0b006ca81322aaec2f5e0f958 Author: Xiangyi Zhu <[email protected]> AuthorDate: Fri Sep 18 11:59:15 2026 +0800 [core] Reject a non-parquet per-level or changelog format for variant columns at DDL time (#9942) --- .../org/apache/paimon/schema/SchemaValidation.java | 46 ++++++++++++++++ .../apache/paimon/schema/SchemaValidationTest.java | 63 ++++++++++++++++++++++ 2 files changed, 109 insertions(+) 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 3a31f7dbba..ed316a7223 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 @@ -251,6 +251,7 @@ public class SchemaValidation { FileFormat.fromIdentifier(options.formatType(), new Options(schema.options())); RowType tableRowType = new RowType(schema.fields()); validateGeospatialTypes(schema, options, tableRowType); + validateVariantTypes(schema, options); validateIcebergTimestampPrecisions(tableRowType, options); validateIcebergTimePrecisions(tableRowType, options); validateBlobFields(tableRowType, options); @@ -492,6 +493,51 @@ 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 = 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 238dbef999..681edbf1da 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 @@ -1920,6 +1920,69 @@ 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 + List<DataField> fields = + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField(1, "v", DataTypes.VARIANT()), + new DataField( + 2, + "nested", + DataTypes.ROW(DataTypes.FIELD(3, "inner", DataTypes.VARIANT())))); + + 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'"); + } + private TableSchema geospatialSchema( List<DataField> fields, List<String> partitionKeys,
