Skip to content

[core] Validate aggregate function types at DDL time #10983

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Nov 6, 2026
+228 −3
Merged

[core] Validate aggregate function types at DDL time #10983

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions paimon-core/src/main/java/org/apache/paimon/schema/FileSystemSchemaManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -276,6 +276,7 @@ public boolean commit(TableSchema newSchema) throws Exception {
SchemaValidation.validateGeneratedColumnEvolution(previous.get(), newSchema);
}
SchemaValidation.validateTableSchema(newSchema);
SchemaValidation.validateAggregationFunctionTypes(newSchema);
SchemaValidation.validateHistoricalIcebergTypes(
this::listAll, new CoreOptions(newSchema.options()));
SchemaValidation.validateFallbackBranch(this, newSchema);
Expand Down
70 changes: 70 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.csv.CsvOptions;
import org.apache.paimon.iceberg.IcebergOptions;
import org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction;
import org.apache.paimon.mergetree.compact.aggregate.AggregateMergeFunction;
import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
import org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
import org.apache.paimon.mergetree.compact.aggregate.factory.FieldLastValueAggFactory;
Expand Down Expand Up @@ -59,6 +61,8 @@
import org.apache.paimon.utils.SetUtils;
import org.apache.paimon.utils.StringUtils;

import javax.annotation.Nullable;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
Expand Down Expand Up @@ -1391,6 +1395,72 @@ private static void validateMergeFunctionFactory(TableSchema schema) {
}
}

/**
* Rejects type-incompatible aggregation functions (e.g. 'collect' on a non-array column) when
* the schema is committed. Only called on the schema commit path, not when loading tables.
*/
public static void validateAggregationFunctionTypes(TableSchema schema) {
if (schema.primaryKeys().isEmpty()) {
return;
}
CoreOptions options = new CoreOptions(schema.options());
switch (options.mergeEngine()) {
case DEDUPLICATE:
case FIRST_ROW:
return;
default:
}

RowType rowType = schema.logicalRowType();
List<String> primaryKeys = schema.primaryKeys();
List<String> sequenceFields = options.sequenceField();
List<String> orderingFields = new ArrayList<>();
List<String> protectedFields = new ArrayList<>();
for (Map.Entry<String, String> entry : options.toMap().entrySet()) {
if (isSequenceGroupOption(entry.getKey())) {
orderingFields.addAll(sequenceGroupOrderingFields(entry.getKey()));
protectedFields.addAll(sequenceGroupProtectedFields(entry.getValue()));
}
}

for (int i = 0; i < rowType.getFieldNames().size(); i++) {
String fieldName = rowType.getFieldNames().get(i);
String aggFuncName =
resolveAggFuncName(
options,
fieldName,
primaryKeys,
sequenceFields,
orderingFields,
protectedFields);
if (aggFuncName == null) {
continue;
}
FieldAggregatorFactory.create(

@JingsongLi JingsongLi Nov 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Keep valid historical projections readable after aggregate schema evolution

AbstractFileStoreTable.runtimeStore() also runs this check, before the read projection is applied. I created a primary-key partial-update table with label VARCHAR(16) protected by fields.seq.sequence-group=label, wrote snapshot 1, and widened label to STRING while enabling listagg through ALTER. Both DDL operations succeed. A read of only id at snapshot 1 passes on the base and now fails before planning with list agg column must be STRING ... VARCHAR(16), because time travel combines the old type with current options. The same eager check also blocks previously working projections on pre-upgrade tables with incompatible aggregators on unused columns. Please run the eager type check on CREATE/ALTER and preserve the runtime projection behavior for stored schemas. Add the historical projection as a regression test.

@ZZZxDong ZZZxDong Nov 5, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed — I'll restrict the eager check to the CREATE/ALTER schema-commit path so reads of stored schemas keep the previous projection-scoped behavior, and add a regression test for the time-travel projection you described.

@ZZZxDong ZZZxDong Nov 6, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Applied both points in bd984a2: the eager type check now runs only from FileSystemSchemaManager.commit() so stored schemas keep the previous projection-scoped read behavior, and TimeTravelSchemaEvolutionTest#testTimeTravelProjectionWithEvolvedAggregateFunction covers the VARCHAR(16) → STRING + listagg scenario you described (verified it fails without the change).

rowType.getFieldTypes().get(i), fieldName, aggFuncName, options);
}
}

/** Mirrors how each merge engine resolves the aggregation function of a field. */
@Nullable
private static String resolveAggFuncName(
CoreOptions options,
String fieldName,
List<String> primaryKeys,
List<String> sequenceFields,
List<String> orderingFields,
List<String> protectedFields) {
if (options.mergeEngine() == MergeEngine.AGGREGATE) {
return AggregateMergeFunction.getAggFuncName(
fieldName, options, primaryKeys, sequenceFields);
}

// partial-update only excludes sequence-group ordering fields from aggregation; a
// sequence.field column can still be configured with an aggregate function.
return PartialUpdateMergeFunction.getAggFuncName(
fieldName, options, primaryKeys, orderingFields, protectedFields);
}

/** Rejects changes which invalidate stored generated values or their dependencies. */
public static void validateGeneratedColumnEvolution(TableSchema previous, TableSchema next) {
for (DataField field : previous.fields()) {
Expand Down
2 changes: 1 addition & 1 deletion paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@ public class SchemaManagerTest {
private final Map<String, String> options = Collections.singletonMap("key", "value");
private final RowType rowType = RowType.of(new IntType(), new BigIntType(), new VarCharType());
private final RowType rowTypeWithSequenceField =
RowType.of(new IntType(), new BigIntType(), new VarCharType(), new BigIntType());
RowType.of(new IntType(), new BigIntType(), new BigIntType(), new BigIntType());
private final Schema schema =
new Schema(rowType.getFields(), partitionKeys, primaryKeys, options, "");

Expand Down
123 changes: 121 additions & 2 deletions paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
import static org.apache.paimon.CoreOptions.SCAN_SNAPSHOT_ID;
import static org.apache.paimon.CoreOptions.VECTOR_FIELD;
import static org.apache.paimon.CoreOptions.VECTOR_FILE_FORMAT;
import static org.apache.paimon.schema.SchemaValidation.validateAggregationFunctionTypes;
import static org.apache.paimon.schema.SchemaValidation.validateTableSchema;
import static org.apache.paimon.schema.TableSchema.CURRENT_VERSION;
import static org.apache.paimon.schema.TableSchema.PAIMON_07_VERSION;
Expand Down Expand Up @@ -75,6 +76,115 @@ void testNullablePrimaryKeyRequiresPrimaryKeyTable() {
"Option 'primary-key.nullable' can only be enabled for a table with primary keys.");
}

@Test
public void testPartialUpdateSequenceGroupOrderingFieldSkipped() {
// A sequence-group ordering field never gets an aggregator, so an incompatible
// aggregate-function configured on it must not be rejected at DDL time.
Map<String, String> options = new HashMap<>();
options.put("merge-engine", "partial-update");
options.put("fields.f2.sequence-group", "f3");
// 'collect' on the INT ordering field would be invalid if it were checked.
options.put("fields.f2.aggregate-function", "collect");
assertThatNoException().isThrownBy(() -> validateTableSchemaExec(options));
}

@Test
public void testPartialUpdateGlobalSequenceFieldStillValidated() {
// A global sequence.field column is not a sequence-group ordering field, so it still
// takes part in aggregation resolution and must be validated.
Map<String, String> options = new HashMap<>();
options.put("merge-engine", "partial-update");
options.put("sequence.field", "f2");
options.put("fields.default-aggregate-function", "collect");
options.put(BUCKET.key(), String.valueOf(-1));
TableSchema schema =
new TableSchema(
1,
Arrays.asList(
new DataField(0, "f0", DataTypes.INT()),
new DataField(2, "f2", DataTypes.BIGINT())),
10,
emptyList(),
singletonList("f0"),
options,
"");

assertThatThrownBy(() -> validateTableSchema(schema))
.hasMessageContaining(
"Must use sequence group for aggregation functions but not found for field f2");
}

@Test
public void testAggregateFunctionTypeMismatch() {
Map<String, String> options = new HashMap<>();
options.put("merge-engine", "aggregation");
options.put("fields.f3.aggregate-function", "collect");
assertThatThrownBy(() -> validateAggregationFunctionExec(options))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining(
"Data type for collect column must be 'Array' but was 'STRING'");

options.put("fields.f3.aggregate-function", "sum");
assertThatThrownBy(() -> validateAggregationFunctionExec(options))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Data type for sum column must be 'NumericType'");
}

@Test
public void testPartialUpdateAggregateFunctionTypeMismatch() {
Map<String, String> options = new HashMap<>();
options.put("merge-engine", "partial-update");
options.put("fields.f2.sequence-group", "f3");
options.put("fields.f3.aggregate-function", "collect");
assertThatThrownBy(() -> validateAggregationFunctionExec(options))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining(
"Data type for collect column must be 'Array' but was 'STRING'");

options.put("fields.f3.aggregate-function", "sum");
assertThatThrownBy(() -> validateAggregationFunctionExec(options))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Data type for sum column must be 'NumericType'");
}

@Test
public void testAggregateFunctionTypeCheckOnlyOnSchemaCommit() {
// A stored schema (possibly combined with current options by time travel) must stay
// readable, so loading a table only checks the function name, not the field type.
Map<String, String> options = new HashMap<>();
options.put("merge-engine", "aggregation");
options.put("fields.f3.aggregate-function", "collect");
assertThatNoException().isThrownBy(() -> validateTableSchemaExec(options));
}

@Test
public void testAggregateFunctionCompatibleTypePasses() {
Map<String, String> options = new HashMap<>(2);
options.put("merge-engine", "aggregation");
options.put("fields.f2.aggregate-function", "sum");
assertThatCode(() -> validateAggregationFunctionExec(options)).doesNotThrowAnyException();
}

@Test
public void testPartialUpdateAggregateFunctionCompatibleTypePasses() {
Map<String, String> options = new HashMap<>();
options.put("merge-engine", "partial-update");
options.put("fields.f2.sequence-group", "f3");
options.put("fields.f3.aggregate-function", "collect");
options.put(BUCKET.key(), String.valueOf(-1));

List<DataField> fields =
Arrays.asList(
new DataField(0, "f0", DataTypes.INT()),
new DataField(1, "f1", DataTypes.INT()),
new DataField(2, "f2", DataTypes.INT()),
new DataField(3, "f3", DataTypes.ARRAY(DataTypes.STRING())));
TableSchema schema =
new TableSchema(1, fields, 10, emptyList(), singletonList("f1"), options, "");

assertThatNoException().isThrownBy(() -> validateTableSchema(schema));
}

@Test
public void testSequenceGroupSortOrderValidation() {
Map<String, String> options = new HashMap<>();
Expand Down Expand Up @@ -140,6 +250,16 @@ public void testRowSequenceGroupSortOrder(String sortOrder) {
}

private void validateTableSchemaExec(Map<String, String> options) {
validateTableSchema(schemaForAggregation(options));
}

private void validateAggregationFunctionExec(Map<String, String> options) {
TableSchema schema = schemaForAggregation(options);
validateTableSchema(schema);
validateAggregationFunctionTypes(schema);
}

private TableSchema schemaForAggregation(Map<String, String> options) {
List<DataField> fields =
Arrays.asList(
new DataField(0, "f0", DataTypes.INT()),
Expand All @@ -149,8 +269,7 @@ private void validateTableSchemaExec(Map<String, String> options) {
List<String> partitionKeys = singletonList("f0");
List<String> primaryKeys = singletonList("f1");
options.put(BUCKET.key(), String.valueOf(-1));
validateTableSchema(
new TableSchema(1, fields, 10, partitionKeys, primaryKeys, options, ""));
return new TableSchema(1, fields, 10, partitionKeys, primaryKeys, options, "");
}

private void validateBlobSchema(Map<String, String> options, List<String> partitions) {
Expand Down
35 changes: 35 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/table/TimeTravelSchemaEvolutionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.apache.paimon.table;

import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.Blob;
import org.apache.paimon.data.BlobDescriptor;
import org.apache.paimon.data.GenericRow;
Expand Down Expand Up @@ -218,6 +219,40 @@ public void testExplicitVectorOptionsArePreserved() throws Exception {
.doesNotContainKey(CoreOptions.VECTOR_FIELD.key());
}

@Test
public void testTimeTravelProjectionWithEvolvedAggregateFunction() throws Exception {
// Time travel combines the historical column type with the current aggregation options,
// so a projection that does not read the aggregated column must stay readable.
Schema schema =
Schema.newBuilder()
.column("id", DataTypes.INT())
.column("seq", DataTypes.BIGINT())
.column("label", DataTypes.VARCHAR(16))
.primaryKey("id")
.option(CoreOptions.MERGE_ENGINE.key(), "partial-update")
.option(CoreOptions.BUCKET.key(), "1")
.option("fields.seq.sequence-group", "label")
.build();
catalog.createTable(identifier(), schema, false);
write(getTableDefault(), GenericRow.of(1, 1L, BinaryString.fromString("a")));

catalog.alterTable(
identifier(), SchemaChange.updateColumnType("label", DataTypes.STRING()), false);
catalog.alterTable(
identifier(),
SchemaChange.setOption("fields.label.aggregate-function", "listagg"),
false);

FileStoreTable historicalTable =
getTableDefault()
.copy(Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), "1"));
assertThat(historicalTable.schema().fields().get(2).type())
.isEqualTo(DataTypes.VARCHAR(16));
assertThat(read(historicalTable, new int[] {0}))
.extracting(row -> row.getInt(0))
.containsExactly(1);
}

private void assertHistoricalTable(FileStoreTable historicalTable, TableSchema historicalSchema)
throws Exception {
assertThat(historicalTable.schema().id()).isEqualTo(historicalSchema.id());
Expand Down
Loading