This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new e2a44039c [server] Support adding aggregation columns (#3459)
e2a44039c is described below
commit e2a44039c303882f58bf9c671501eda1799e1bb1
Author: Liurnly <[email protected]>
AuthorDate: Fri Jul 3 16:23:46 2026 +0800
[server] Support adding aggregation columns (#3459)
---
.../fluss/client/utils/ClientRpcMessageUtils.java | 10 +++
.../fluss/client/admin/FlussAdminITCase.java | 74 ++++++++++++++++++++++
.../org/apache/fluss/metadata/TableChange.java | 55 ++++++++++++----
fluss-rpc/src/main/proto/FlussApi.proto | 2 +
.../fluss/server/coordinator/MetadataManager.java | 2 +
.../fluss/server/coordinator/SchemaUpdate.java | 7 +-
.../fluss/server/utils/ServerRpcMessageUtils.java | 43 ++++++++++---
.../server/utils/TableDescriptorValidation.java | 31 +++++++++
8 files changed, 201 insertions(+), 23 deletions(-)
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java
b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java
index 2df5b91ef..70f940c3c 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java
@@ -41,6 +41,7 @@ import org.apache.fluss.config.cluster.ConfigEntry;
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.fs.FsPathAndFileName;
import org.apache.fluss.fs.token.ObtainedSecurityToken;
+import org.apache.fluss.metadata.AggFunction;
import org.apache.fluss.metadata.DatabaseChange;
import org.apache.fluss.metadata.DatabaseSummary;
import org.apache.fluss.metadata.PartitionInfo;
@@ -690,6 +691,15 @@ public class ClientRpcMessageUtils {
if (addColumn.getComment() != null) {
pbAddColumn.setComment(addColumn.getComment());
}
+ if (addColumn.getAggFunction().isPresent()) {
+ AggFunction aggFunction = addColumn.getAggFunction().get();
+ pbAddColumn.setAggFunctionType(aggFunction.getType().toString());
+ aggFunction
+ .getParameters()
+ .forEach(
+ (key, value) ->
+
pbAddColumn.addAggFunctionParam().setKey(key).setValue(value));
+ }
return pbAddColumn;
}
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
index 68181844e..3d6b48dfc 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
@@ -120,6 +120,7 @@ import static
org.apache.fluss.metadata.DataLakeFormat.PAIMON;
import static
org.apache.fluss.record.TestData.DATA1_PARTITIONED_TABLE_DESCRIPTOR;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.apache.fluss.testutils.InternalRowAssert.assertThatRow;
import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -568,6 +569,73 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
.hasMessageContaining("Column nested_row already exists");
}
+ @Test
+ void testAlterAggregationTableColumnWithAggFunction() throws Exception {
+ TablePath tablePath = TablePath.of("test_db",
"alter_aggregation_table_column");
+ Map<String, String> properties = new HashMap<>();
+ properties.put(ConfigOptions.TABLE_MERGE_ENGINE.key(), "aggregation");
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("value", DataTypes.BIGINT(),
AggFunctions.SUM())
+ .primaryKey("id")
+ .build())
+ .distributedBy(3, "id")
+ .properties(properties)
+ .build();
+ admin.createTable(tablePath, tableDescriptor, false).get();
+
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_value",
+ DataTypes.BIGINT(),
+ "new aggregate column",
+ TableChange.ColumnPosition.last(),
+ AggFunctions.SUM())),
+ false)
+ .get();
+
+ SchemaInfo schemaInfo = admin.getTableSchema(tablePath).get();
+
assertThat(schemaInfo.getSchema().getAggFunction("new_value")).hasValue(AggFunctions.SUM());
+
+ try (Table table = conn.getTable(tablePath)) {
+ UpsertWriter upsertWriter = table.newUpsert().createWriter();
+ upsertWriter.upsert(row(1, 10L, 100L));
+ upsertWriter.upsert(row(1, 20L, 200L));
+ upsertWriter.flush();
+
+ assertThatRow(lookupRow(table.newLookup().createLookuper(),
row(1)))
+ .withSchema(schemaInfo.getSchema().getRowType())
+ .isEqualTo(row(1, 30L, 300L));
+ }
+ }
+
+ @Test
+ void testAlterNonAggregationTableColumnWithAggFunction() throws Exception {
+ TablePath tablePath = TablePath.of("test_db",
"alter_non_aggregation_table_column");
+ admin.createTable(tablePath, DEFAULT_TABLE_DESCRIPTOR, false).get();
+
+ assertThatThrownBy(
+ () ->
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.addColumn(
+ "new_value",
+
DataTypes.BIGINT(),
+ "new aggregate
column",
+
TableChange.ColumnPosition.last(),
+
AggFunctions.SUM())),
+ false)
+ .get())
+ .hasMessageContaining(
+ "Aggregation function is only supported for
aggregation merge engine table");
+ }
+
@Test
void testCreateInvalidDatabaseAndTable() throws Exception {
assertThatThrownBy(
@@ -675,6 +743,12 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
TableInfo tableInfoAggregate =
admin.getTableInfo(tablePathAggregate).join();
assertThat(tableInfoAggregate.getTableConfig().getDeleteBehavior())
.hasValue(DeleteBehavior.IGNORE);
+ assertThat(tableInfoAggregate.getSchema().getAggFunction("count"))
+ .hasValue(AggFunctions.SUM());
+ assertThat(tableInfoAggregate.getProperties().toMap())
+ .doesNotContainKey("fields.count.agg");
+ assertThat(tableInfoAggregate.getCustomProperties().toMap())
+ .doesNotContainKey("fields.count.agg");
// Test 2.6: AGGREGATION merge engine with delete behavior explicitly
set to ALLOW - should
// be allowed
diff --git
a/fluss-common/src/main/java/org/apache/fluss/metadata/TableChange.java
b/fluss-common/src/main/java/org/apache/fluss/metadata/TableChange.java
index bb027d690..2890ca82e 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metadata/TableChange.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metadata/TableChange.java
@@ -22,6 +22,7 @@ import org.apache.fluss.types.DataType;
import javax.annotation.Nullable;
import java.util.Objects;
+import java.util.Optional;
/** {@link TableChange} represents the modification of the Fluss Table. */
public interface TableChange {
@@ -42,7 +43,21 @@ public interface TableChange {
DataType dataType,
@Nullable String comment,
ColumnPosition position) {
- return new AddColumn(columnName, dataType, comment, position);
+ return new AddColumn(columnName, dataType, comment, position, null);
+ }
+
+ /**
+ * A table change to add the column with specified position and
aggregation function.
+ *
+ * @return a TableChange represents the modification.
+ */
+ static AddColumn addColumn(
+ String columnName,
+ DataType dataType,
+ @Nullable String comment,
+ ColumnPosition position,
+ @Nullable AggFunction aggFunction) {
+ return new AddColumn(columnName, dataType, comment, position,
aggFunction);
}
/**
@@ -230,15 +245,21 @@ public interface TableChange {
private final String name;
private final DataType dataType;
private final @Nullable String comment;
+ private final @Nullable AggFunction aggFunction;
private final ColumnPosition position;
private AddColumn(
- String name, DataType dataType, @Nullable String comment,
ColumnPosition position) {
+ String name,
+ DataType dataType,
+ @Nullable String comment,
+ ColumnPosition position,
+ @Nullable AggFunction aggFunction) {
this.name = name;
this.dataType = dataType;
this.comment = comment;
this.position = position;
+ this.aggFunction = aggFunction;
}
public String getName() {
@@ -258,20 +279,26 @@ public interface TableChange {
return position;
}
+ public Optional<AggFunction> getAggFunction() {
+ return Optional.ofNullable(aggFunction);
+ }
+
@Override
public String toString() {
- return "AddColumn{"
- + "name='"
- + name
- + '\''
- + ", dataType="
- + dataType
- + ", comment='"
- + comment
- + '\''
- + ", position="
- + position
- + '}';
+ String result =
+ "AddColumn{"
+ + "name='"
+ + name
+ + '\''
+ + ", dataType="
+ + dataType
+ + ", comment='"
+ + comment
+ + '\'';
+ if (aggFunction != null) {
+ result += ", aggFunction=" + aggFunction;
+ }
+ return result + ", position=" + position + '}';
}
}
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto
b/fluss-rpc/src/main/proto/FlussApi.proto
index 41c738777..57e96092d 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -1193,6 +1193,8 @@ message PbAddColumn {
required bytes data_type_json = 2;
optional string comment = 3;
required int32 column_position_type = 4; // LAST=0,FIRST=1,AFTER=3
+ optional string agg_function_type = 5;
+ repeated PbKeyValue agg_function_params = 6;
}
message PbDropColumn {
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
index 106d4ef66..65b2ee24b 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
@@ -76,6 +76,7 @@ import java.util.Set;
import java.util.concurrent.Callable;
import static
org.apache.fluss.server.utils.TableDescriptorValidation.validateAlterTableProperties;
+import static
org.apache.fluss.server.utils.TableDescriptorValidation.validateAlterTableSchema;
/** A manager for metadata. */
public class MetadataManager {
@@ -440,6 +441,7 @@ public class MetadataManager {
if (!schemaChanges.isEmpty()) {
Schema newSchema =
SchemaUpdate.applySchemaChanges(table.getSchema(),
schemaChanges);
+ validateAlterTableSchema(table, newSchema);
LakeCatalog.Context lakeCatalogContext =
new CoordinatorService.DefaultLakeCatalogContext(
false,
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/SchemaUpdate.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/SchemaUpdate.java
index d1ad0c16c..e85219a3b 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/SchemaUpdate.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/SchemaUpdate.java
@@ -82,7 +82,12 @@ public class SchemaUpdate {
}
// Delegate the actual addition to the builder
- builder.column(addColumn.getName(), addColumn.getDataType());
+ if (addColumn.getAggFunction().isPresent()) {
+ builder.column(
+ addColumn.getName(), addColumn.getDataType(),
addColumn.getAggFunction().get());
+ } else {
+ builder.column(addColumn.getName(), addColumn.getDataType());
+ }
// Fixed: Use null check for the String comment
String comment = addColumn.getComment();
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
index 266abe3bc..1e5e914c0 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -27,9 +27,13 @@ import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.cluster.AlterConfigOpType;
import org.apache.fluss.config.cluster.ColumnPositionType;
import org.apache.fluss.config.cluster.ConfigEntry;
+import org.apache.fluss.exception.InvalidConfigException;
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.fs.token.ObtainedSecurityToken;
import org.apache.fluss.lake.committer.LakeCommitResult;
+import org.apache.fluss.metadata.AggFunction;
+import org.apache.fluss.metadata.AggFunctionType;
+import org.apache.fluss.metadata.AggFunctions;
import org.apache.fluss.metadata.DatabaseChange;
import org.apache.fluss.metadata.DatabaseSummary;
import org.apache.fluss.metadata.PartitionSpec;
@@ -358,17 +362,40 @@ public class ServerRpcMessageUtils {
return addColumns.stream()
.filter(Objects::nonNull)
.map(
- pbAddColumn ->
- TableChange.addColumn(
- pbAddColumn.getColumnName(),
- JsonSerdeUtils.readValue(
- pbAddColumn.getDataTypeJson(),
- DataTypeJsonSerde.INSTANCE),
- pbAddColumn.hasComment() ?
pbAddColumn.getComment() : null,
-
toColumnPosition(pbAddColumn.getColumnPositionType())))
+ pbAddColumn -> {
+ AggFunction aggFunction =
toAggFunction(pbAddColumn);
+ return TableChange.addColumn(
+ pbAddColumn.getColumnName(),
+ JsonSerdeUtils.readValue(
+ pbAddColumn.getDataTypeJson(),
+ DataTypeJsonSerde.INSTANCE),
+ pbAddColumn.hasComment() ?
pbAddColumn.getComment() : null,
+
toColumnPosition(pbAddColumn.getColumnPositionType()),
+ aggFunction);
+ })
.collect(Collectors.toList());
}
+ private static AggFunction toAggFunction(PbAddColumn pbAddColumn) {
+ if (!pbAddColumn.hasAggFunctionType()) {
+ return null;
+ }
+
+ AggFunctionType type =
AggFunctionType.fromString(pbAddColumn.getAggFunctionType());
+ if (type == null) {
+ throw new InvalidConfigException(
+ String.format(
+ "Unknown aggregation function type: %s",
+ pbAddColumn.getAggFunctionType()));
+ }
+
+ Map<String, String> parameters = new HashMap<>();
+ for (PbKeyValue parameter : pbAddColumn.getAggFunctionParamsList()) {
+ parameters.put(parameter.getKey(), parameter.getValue());
+ }
+ return AggFunctions.of(type, parameters);
+ }
+
public static List<TableChange.SchemaChange>
toDropColumns(List<PbDropColumn> dropColumns) {
return dropColumns.stream()
.filter(Objects::nonNull)
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
index 949afd8c6..6b60a9409 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
@@ -17,6 +17,7 @@
package org.apache.fluss.server.utils;
+import org.apache.fluss.annotation.Internal;
import org.apache.fluss.config.ConfigOption;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
@@ -129,6 +130,19 @@ public class TableDescriptorValidation {
checkTableLakeFormatMatchesCluster(tableConf, clusterDataLakeFormat);
}
+ /** Validates the schema after altering table columns. */
+ @Internal
+ public static void validateAlterTableSchema(TableInfo table, Schema
newSchema) {
+ if (table.getTableConfig()
+ .getMergeEngineType()
+ .map(MergeEngineType.AGGREGATION::equals)
+ .orElse(false)) {
+ validateAggregationFunctionParameters(newSchema);
+ } else {
+ validateNoAggregationFunctions(newSchema);
+ }
+ }
+
private static void checkTableLakeFormatMatchesCluster(
Configuration tableConf, @Nullable DataLakeFormat
clusterDataLakeFormat) {
if (clusterDataLakeFormat == null) {
@@ -323,6 +337,9 @@ public class TableDescriptorValidation {
private static void checkMergeEngine(
Configuration tableConf, boolean hasPrimaryKey, Schema schema) {
MergeEngineType mergeEngine =
tableConf.get(ConfigOptions.TABLE_MERGE_ENGINE);
+ if (mergeEngine != MergeEngineType.AGGREGATION) {
+ validateNoAggregationFunctions(schema);
+ }
if (mergeEngine != null) {
if (!hasPrimaryKey) {
throw new InvalidConfigException(
@@ -377,6 +394,20 @@ public class TableDescriptorValidation {
}
}
+ /** Validates that the schema doesn't contain any aggregation functions. */
+ private static void validateNoAggregationFunctions(Schema schema) {
+ for (Schema.Column column : schema.getColumns()) {
+ Optional<AggFunction> aggFunction = column.getAggFunction();
+ if (aggFunction.isPresent()) {
+ throw new InvalidConfigException(
+ String.format(
+ "Aggregation function is only supported for
aggregation merge engine table, "
+ + "but column '%s' has aggregation
function '%s'.",
+ column.getName(), aggFunction.get()));
+ }
+ }
+ }
+
/**
* Validates aggregation function parameters in the schema.
*
