You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ya...@apache.org on 2026/10/12 09:51:59 UTC
(fluss) branch main updated: [server] Add last_n aggregate function to aggregation merge engine (#4570)
This is an automated email from the ASF dual-hosted git repository.
platinumhamburg 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 a85c6a13d [server] Add last_n aggregate function to aggregation merge engine (#4570)
a85c6a13d is described below
commit a85c6a13dddaa6d4723afdd125d52f744ab1bd8e
Author: Anton Borisov <14...@users.noreply.github.com>
AuthorDate: Mon Oct 12 10:51:52 2026 +0100
[server] Add last_n aggregate function to aggregation merge engine (#4570)
* [server] Add last_n aggregate function to aggregation merge engine
* address comments
* [server] Validate last_n size at DDL and add tests
---------
Co-authored-by: Giannis Polyzos <ip...@gmail.com>
---
.../org/apache/fluss/metadata/AggFunction.java | 11 +-
.../org/apache/fluss/metadata/AggFunctionType.java | 39 ++++-
.../org/apache/fluss/metadata/AggFunctions.java | 22 +++
.../apache/fluss/metadata/TableDescriptorTest.java | 9 ++
.../fluss/flink/sink/FlinkTableSinkITCase.java | 39 +++++
.../server/kv/rowmerger/AggregateRowMerger.java | 16 ++-
.../aggregate/AggregateFieldsProcessor.java | 21 +++
.../kv/rowmerger/aggregate/AggregationContext.java | 11 ++
.../aggregate/factory/FieldLastNAggFactory.java | 48 +++++++
.../aggregate/functions/FieldAggregator.java | 8 ++
.../aggregate/functions/FieldLastNAgg.java | 78 ++++++++++
...merger.aggregate.factory.FieldAggregatorFactory | 1 +
.../kv/rowmerger/AggregateRowMergerTest.java | 57 ++++++++
.../aggregate/functions/FieldLastNAggTest.java | 158 +++++++++++++++++++++
.../utils/TableDescriptorValidationTest.java | 30 ++++
.../docs/table-design/merge-engines/aggregation.md | 64 +++++++++
16 files changed, 608 insertions(+), 4 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunction.java b/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunction.java
index 1f31a8c22..ee517027b 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunction.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunction.java
@@ -121,13 +121,22 @@ public final class AggFunction implements Serializable {
* <p>This method checks that:
*
* <ul>
+ * <li>All parameters required by the function type are present
* <li>All parameter names are supported by the function type
* <li>All parameter values are valid
* </ul>
*
- * @throws IllegalArgumentException if any parameter is invalid
+ * @throws IllegalArgumentException if any parameter is missing or invalid
*/
public void validateParameters() {
+ for (String required : type.getRequiredParameters()) {
+ if (!parameters.containsKey(required)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Parameter '%s' is required for aggregation function '%s'.",
+ required, type));
+ }
+ }
for (Map.Entry<String, String> entry : parameters.entrySet()) {
type.validateParameter(entry.getKey(), entry.getValue());
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctionType.java b/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctionType.java
index f0026fbba..79ac0575d 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctionType.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctionType.java
@@ -27,6 +27,7 @@ import java.util.Locale;
import java.util.Set;
import static org.apache.fluss.metadata.AggFunctions.PARAM_DELIMITER;
+import static org.apache.fluss.metadata.AggFunctions.PARAM_SIZE;
import static org.apache.fluss.utils.Preconditions.checkArgument;
/**
@@ -59,7 +60,10 @@ public enum AggFunctionType {
// Roaring bitmap aggregation
RBM32,
- RBM64;
+ RBM64,
+
+ // Array aggregation
+ LAST_N;
// ------------------------------------------------------------------------------------------
@@ -102,12 +106,29 @@ public enum AggFunctionType {
case STRING_AGG:
// LISTAGG and STRING_AGG support optional "delimiter" parameter
return Collections.singleton(PARAM_DELIMITER);
+ case LAST_N:
+ return Collections.singleton(PARAM_SIZE);
default:
// All other functions do not accept parameters
return Collections.emptySet();
}
}
+ /**
+ * Returns the set of parameter names that must be provided for this aggregation function.
+ *
+ * @return an immutable set of required parameter names
+ */
+ public Set<String> getRequiredParameters() {
+ switch (this) {
+ case LAST_N:
+ // LAST_N has no default for "size"
+ return Collections.singleton(PARAM_SIZE);
+ default:
+ return Collections.emptySet();
+ }
+ }
+
/**
* Returns the supported data type roots for this aggregation function.
*
@@ -121,6 +142,8 @@ public enum AggFunctionType {
case RBM32:
case RBM64:
return new DataTypeRoot[] {DataTypeRoot.BYTES};
+ case LAST_N:
+ return new DataTypeRoot[] {DataTypeRoot.ARRAY};
case LISTAGG:
case STRING_AGG:
return new DataTypeRoot[] {DataTypeRoot.STRING, DataTypeRoot.CHAR};
@@ -177,6 +200,20 @@ public enum AggFunctionType {
}
}
break;
+ case LAST_N:
+ int size;
+ try {
+ size = Integer.parseInt(parameterValue);
+ } catch (NumberFormatException e) {
+ size = -1;
+ }
+ if (size <= 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Parameter '%s' for aggregation function '%s' must be a positive integer, but was '%s'",
+ parameterName, this, parameterValue));
+ }
+ break;
default:
// No validation needed for other functions (they don't have parameters)
break;
diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctions.java b/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctions.java
index 86a704ef6..e43f3ab3a 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctions.java
@@ -300,6 +300,28 @@ public final class AggFunctions {
return new AggFunction(AggFunctionType.RBM64, null);
}
+ // ===================================================================================
+ // Array Aggregation Functions
+ // ===================================================================================
+
+ /** Parameter name for the number of elements LAST_N keeps. */
+ public static final String PARAM_SIZE = "size";
+
+ /**
+ * Creates a LAST_N aggregation function that appends input arrays and keeps the last {@code
+ * size} elements.
+ *
+ * <p>Supported data types: ARRAY
+ *
+ * @param size the number of elements to keep, must be positive
+ * @return a LAST_N aggregation function
+ */
+ public static AggFunction LAST_N(int size) {
+ Map<String, String> params = new HashMap<>();
+ params.put(PARAM_SIZE, String.valueOf(size));
+ return new AggFunction(AggFunctionType.LAST_N, params);
+ }
+
// ===================================================================================
// Internal Factory Methods
// ===================================================================================
diff --git a/fluss-common/src/test/java/org/apache/fluss/metadata/TableDescriptorTest.java b/fluss-common/src/test/java/org/apache/fluss/metadata/TableDescriptorTest.java
index bd9961075..294d988ed 100644
--- a/fluss-common/src/test/java/org/apache/fluss/metadata/TableDescriptorTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/metadata/TableDescriptorTest.java
@@ -327,6 +327,15 @@ class TableDescriptorTest {
.hasMessageContaining("must be a non-empty string");
}
+ @Test
+ void testLastNRequiresSizeParameter() {
+ AggFunctions.LAST_N(5).validateParameters();
+
+ assertThatThrownBy(() -> AggFunctions.of(AggFunctionType.LAST_N).validateParameters())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("'size' is required");
+ }
+
@Test
void testInvalidListaggParameterUnknownParameter() {
// LISTAGG with unknown parameter - should fail
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
index 06eb79540..428d8ff56 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
@@ -1585,6 +1585,45 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
assertResultsIgnoreOrder(rowIter, expectedRows, true);
}
+ @Test
+ void testLastNAggregation() throws Exception {
+ tEnv.executeSql(
+ "create table last_n_agg ("
+ + "id int not null primary key not enforced, "
+ + "events array<string>, "
+ + "scores array<bigint>"
+ + ") with ("
+ + "'table.merge-engine' = 'aggregation', "
+ + "'fields.events.agg' = 'last_n', "
+ + "'fields.events.last_n.size' = '2', "
+ + "'fields.scores.agg' = 'last_n', "
+ + "'fields.scores.last_n.size' = '3')");
+
+ tEnv.executeSql("INSERT INTO last_n_agg VALUES (1, ARRAY['a'], ARRAY[CAST(1 AS BIGINT)])")
+ .await();
+ tEnv.executeSql("INSERT INTO last_n_agg VALUES (1, ARRAY['b'], ARRAY[CAST(2 AS BIGINT)])")
+ .await();
+ tEnv.executeSql(
+ "INSERT INTO last_n_agg VALUES "
+ + "(1, ARRAY['c', 'd'], CAST(NULL AS ARRAY<BIGINT>))")
+ .await();
+ tEnv.executeSql(
+ "INSERT INTO last_n_agg VALUES "
+ + "(2, ARRAY['x', 'y', 'z'], ARRAY[CAST(1 AS BIGINT)])")
+ .await();
+
+ CloseableIterator<Row> rowIter = tEnv.executeSql("SELECT * FROM last_n_agg").collect();
+ List<String> expectedRows =
+ Arrays.asList(
+ "+I[1, [a], [1]]",
+ "-U[1, [a], [1]]",
+ "+U[1, [a, b], [1, 2]]",
+ "-U[1, [a, b], [1, 2]]",
+ "+U[1, [c, d], [1, 2]]",
+ "+I[2, [y, z], [1]]");
+ assertResultsIgnoreOrder(rowIter, expectedRows, true);
+ }
+
@Test
void testPartialUpdateOnAggregationMergeEngine() throws Exception {
tEnv.executeSql(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java
index f84769eb2..3a3783cb5 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java
@@ -88,7 +88,7 @@ public class AggregateRowMerger implements RowMerger {
public BinaryValue merge(@Nullable BinaryValue oldValue, BinaryValue newValue) {
// First write: no existing row
if (oldValue == null || oldValue.row == null) {
- return newValue;
+ return aggregateFirstRow(contextCache, newValue);
}
// Get contexts for schema evolution support
@@ -108,6 +108,18 @@ public class AggregateRowMerger implements RowMerger {
return new BinaryValue(targetSchemaId, mergedRow);
}
+ private static BinaryValue aggregateFirstRow(
+ AggregationContextCache contextCache, BinaryValue value) {
+ AggregationContext context = contextCache.getContext(value.schemaId);
+ if (!context.requiresAggOnFirstWrite()) {
+ return value;
+ }
+ RowEncoder encoder = context.getRowEncoder();
+ encoder.startNewRow();
+ AggregateFieldsProcessor.aggregateFirstRow(value.row, context, encoder);
+ return new BinaryValue(value.schemaId, encoder.finishRow());
+ }
+
@Override
public BinaryValue delete(BinaryValue oldValue) {
// Remove the entire row (returns null to indicate deletion)
@@ -271,7 +283,7 @@ public class AggregateRowMerger implements RowMerger {
public BinaryValue merge(@Nullable BinaryValue oldValue, BinaryValue newValue) {
// First write: no existing row
if (oldValue == null || oldValue.row == null) {
- return newValue;
+ return aggregateFirstRow(contextCache, newValue);
}
// Get contexts for schema evolution support
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregateFieldsProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregateFieldsProcessor.java
index 0429a2015..75191d1af 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregateFieldsProcessor.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregateFieldsProcessor.java
@@ -147,6 +147,27 @@ public final class AggregateFieldsProcessor {
encoder.encodeField(targetIdx, fieldGetter.getFieldOrNull(oldRow));
}
+ /**
+ * Encodes the first row written for a key, passing the fields whose aggregator requires it
+ * through {@code agg(null, value)}.
+ *
+ * @param row the first row written for the key
+ * @param context context for the row schema
+ * @param encoder the row encoder to encode results (should match context)
+ */
+ public static void aggregateFirstRow(
+ BinaryRow row, AggregationContext context, RowEncoder encoder) {
+ InternalRow.FieldGetter[] fieldGetters = context.getFieldGetters();
+ FieldAggregator[] aggregators = context.getAggregators();
+ for (int idx = 0; idx < context.getFieldCount(); idx++) {
+ Object field = fieldGetters[idx].getFieldOrNull(row);
+ FieldAggregator aggregator = aggregators[idx];
+ encoder.encodeField(
+ idx,
+ aggregator.requiresAggOnFirstWrite() ? aggregator.agg(null, field) : field);
+ }
+ }
+
/**
* Aggregate target fields from old and new rows with explicit target schema (partial
* aggregation).
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContext.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContext.java
index 7cec63a4e..4a307e4d5 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContext.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContext.java
@@ -52,6 +52,7 @@ public class AggregationContext {
final FieldAggregator[] aggregators;
final RowEncoder rowEncoder;
final int fieldCount;
+ private final boolean requiresAggOnFirstWrite;
/**
* Mapping from column ID to field index in this schema. This is used for schema evolution to
@@ -77,6 +78,11 @@ public class AggregationContext {
this.aggregators = aggregators;
this.rowEncoder = rowEncoder;
this.fieldCount = rowType.getFieldCount();
+ boolean aggOnFirstWrite = false;
+ for (FieldAggregator aggregator : aggregators) {
+ aggOnFirstWrite |= aggregator.requiresAggOnFirstWrite();
+ }
+ this.requiresAggOnFirstWrite = aggOnFirstWrite;
// Build columnId to index mapping for schema evolution support
this.columnIdToIndex = new HashMap<>();
@@ -92,6 +98,11 @@ public class AggregationContext {
}
}
+ /** Whether any field must be aggregated when the first row for a key is written. */
+ public boolean requiresAggOnFirstWrite() {
+ return requiresAggOnFirstWrite;
+ }
+
public Schema getSchema() {
return schema;
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldLastNAggFactory.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldLastNAggFactory.java
new file mode 100644
index 000000000..22d4bc5d9
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/FieldLastNAggFactory.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.kv.rowmerger.aggregate.factory;
+
+import org.apache.fluss.metadata.AggFunction;
+import org.apache.fluss.metadata.AggFunctionType;
+import org.apache.fluss.metadata.AggFunctions;
+import org.apache.fluss.server.kv.rowmerger.aggregate.functions.FieldLastNAgg;
+import org.apache.fluss.types.ArrayType;
+import org.apache.fluss.types.DataType;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+
+/** Factory for {@link FieldLastNAgg}. */
+public class FieldLastNAggFactory implements FieldAggregatorFactory {
+
+ @Override
+ public FieldLastNAgg create(DataType fieldType, AggFunction aggFunction) {
+ checkArgument(
+ fieldType instanceof ArrayType,
+ "Data type for last_n column must be 'ArrayType' but was '%s'.",
+ fieldType);
+ String size = aggFunction.getParameter(AggFunctions.PARAM_SIZE);
+ checkArgument(size != null, "Aggregation function last_n requires the 'size' parameter.");
+ return new FieldLastNAgg((ArrayType) fieldType, Integer.parseInt(size));
+ }
+
+ @Override
+ public String identifier() {
+ return AggFunctionType.LAST_N.toString();
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldAggregator.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldAggregator.java
index f6ec77a49..73548c944 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldAggregator.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldAggregator.java
@@ -61,6 +61,14 @@ public abstract class FieldAggregator implements Serializable {
return agg(inputField, accumulator);
}
+ /**
+ * Whether the first value written for a key must go through {@code agg(null, value)} instead of
+ * being stored as is.
+ */
+ public boolean requiresAggOnFirstWrite() {
+ return false;
+ }
+
/** Resets the aggregator to a clean start state. */
public void reset() {}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldLastNAgg.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldLastNAgg.java
new file mode 100644
index 000000000..81ed69570
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldLastNAgg.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.kv.rowmerger.aggregate.functions;
+
+import org.apache.fluss.row.GenericArray;
+import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.types.ArrayType;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+
+/** Appends input arrays to the accumulator and keeps the last {@code size} elements. */
+public class FieldLastNAgg extends FieldAggregator {
+
+ private static final long serialVersionUID = 1L;
+
+ private final int size;
+ private final InternalArray.ElementGetter elementGetter;
+
+ /**
+ * Creates a LAST_N aggregator.
+ *
+ * @param dataType the array type of the aggregated column
+ * @param size the number of elements to keep, must be positive
+ */
+ public FieldLastNAgg(ArrayType dataType, int size) {
+ super(dataType);
+ checkArgument(size > 0, "size must be positive, but was %s", size);
+ this.size = size;
+ this.elementGetter = InternalArray.createElementGetter(dataType.getElementType());
+ }
+
+ @Override
+ public Object agg(Object accumulator, Object inputField) {
+ if (inputField == null) {
+ return accumulator;
+ }
+ InternalArray older = (InternalArray) accumulator;
+ InternalArray newer = (InternalArray) inputField;
+ if (older != null && newer.size() == 0) {
+ return accumulator;
+ }
+ if (older == null && newer.size() <= size) {
+ return inputField;
+ }
+ int olderSize = older == null ? 0 : older.size();
+ int total = olderSize + newer.size();
+ int skip = Math.max(0, total - size);
+ Object[] result = new Object[total - skip];
+ for (int i = skip; i < total; i++) {
+ result[i - skip] =
+ i < olderSize
+ ? elementGetter.getElementOrNull(older, i)
+ : elementGetter.getElementOrNull(newer, i - olderSize);
+ }
+ return new GenericArray(result);
+ }
+
+ @Override
+ public boolean requiresAggOnFirstWrite() {
+ return true;
+ }
+}
diff --git a/fluss-server/src/main/resources/META-INF/services/org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldAggregatorFactory b/fluss-server/src/main/resources/META-INF/services/org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldAggregatorFactory
index cb4475ff3..7fdfe2787 100644
--- a/fluss-server/src/main/resources/META-INF/services/org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldAggregatorFactory
+++ b/fluss-server/src/main/resources/META-INF/services/org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldAggregatorFactory
@@ -30,3 +30,4 @@ org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldListaggAggFactory
org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldStringAggFactory
org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldRoaringBitmap32AggFactory
org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldRoaringBitmap64AggFactory
+org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldLastNAggFactory
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMergerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMergerTest.java
index bd1bd3217..1a905a9f8 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMergerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMergerTest.java
@@ -27,6 +27,7 @@ import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.record.BinaryValue;
import org.apache.fluss.record.TestingSchemaGetter;
import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.GenericArray;
import org.apache.fluss.types.DataTypes;
import org.apache.fluss.types.RowType;
@@ -54,6 +55,14 @@ class AggregateRowMergerTest {
private static final RowType ROW_TYPE_SUM = SCHEMA_SUM.getRowType();
+ private static final Schema SCHEMA_LAST_N =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("events", DataTypes.ARRAY(DataTypes.BIGINT()), AggFunctions.LAST_N(2))
+ .column("count", DataTypes.BIGINT(), AggFunctions.SUM())
+ .primaryKey("id")
+ .build();
+
@Test
void testBasicAggregation() {
// Create a single schema with all aggregation types
@@ -999,6 +1008,54 @@ class AggregateRowMergerTest {
}
}
+ @Test
+ void testLastNTrimsFirstWrite() {
+ TableConfig tableConfig = new TableConfig(new Configuration());
+ AggregateRowMerger merger = createMerger(SCHEMA_LAST_N, tableConfig);
+ merger.configureTargetColumns(null, SCHEMA_ID, SCHEMA_LAST_N);
+ RowType rowType = SCHEMA_LAST_N.getRowType();
+
+ // first write for the key is trimmed to the last 2 elements, other fields kept as is
+ BinaryValue first =
+ merger.merge(
+ null,
+ toBinaryValue(compactedRow(rowType, new Object[] {1, longs(1, 2, 3), 5L})));
+ assertThat(first.schemaId).isEqualTo(SCHEMA_ID);
+ assertThat(first.row.getInt(0)).isEqualTo(1);
+ assertThat(first.row.getArray(1).toLongArray()).containsExactly(2L, 3L);
+ assertThat(first.row.getLong(2)).isEqualTo(5L);
+
+ BinaryValue second =
+ merger.merge(
+ first,
+ toBinaryValue(compactedRow(rowType, new Object[] {1, longs(4), 1L})));
+ assertThat(second.row.getArray(1).toLongArray()).containsExactly(3L, 4L);
+ assertThat(second.row.getLong(2)).isEqualTo(6L);
+ }
+
+ @Test
+ void testLastNTrimsFirstWriteWithPartialUpdate() {
+ TableConfig tableConfig = new TableConfig(new Configuration());
+ AggregateRowMerger merger = createMerger(SCHEMA_LAST_N, tableConfig);
+ RowMerger partialMerger =
+ merger.configureTargetColumns(new int[] {0, 1}, SCHEMA_ID, SCHEMA_LAST_N);
+ RowType rowType = SCHEMA_LAST_N.getRowType();
+
+ BinaryValue first =
+ partialMerger.merge(
+ null,
+ toBinaryValue(
+ compactedRow(rowType, new Object[] {1, longs(1, 2, 3), null})));
+ assertThat(first.schemaId).isEqualTo(SCHEMA_ID);
+ assertThat(first.row.getInt(0)).isEqualTo(1);
+ assertThat(first.row.getArray(1).toLongArray()).containsExactly(2L, 3L);
+ assertThat(first.row.isNullAt(2)).isTrue();
+ }
+
+ private static GenericArray longs(long... values) {
+ return new GenericArray(values);
+ }
+
private AggregateRowMerger createMerger(Schema schema, TableConfig tableConfig) {
TestingSchemaGetter schemaGetter =
new TestingSchemaGetter(new SchemaInfo(schema, SCHEMA_ID));
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldLastNAggTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldLastNAggTest.java
new file mode 100644
index 000000000..4467f30ec
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldLastNAggTest.java
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.kv.rowmerger.aggregate.functions;
+
+import org.apache.fluss.metadata.AggFunctionType;
+import org.apache.fluss.metadata.AggFunctions;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.GenericArray;
+import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldAggregatorFactory;
+import org.apache.fluss.types.ArrayType;
+import org.apache.fluss.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link FieldLastNAgg}. */
+class FieldLastNAggTest {
+
+ private static final ArrayType BIGINTS = DataTypes.ARRAY(DataTypes.BIGINT());
+
+ @Test
+ void testKeepsTheLastN() {
+ FieldLastNAgg agg = new FieldLastNAgg(BIGINTS, 3);
+ Object acc = null;
+ for (long i = 0; i < 10; i++) {
+ acc = agg.agg(acc, longs(i));
+ }
+ assertThat(toList(acc)).containsExactly(7L, 8L, 9L);
+ }
+
+ @Test
+ void testBelowSizeKeepsEverythingInOrder() {
+ FieldLastNAgg agg = new FieldLastNAgg(BIGINTS, 5);
+ Object acc = agg.agg(null, longs(1));
+ acc = agg.agg(acc, longs(2, 3));
+ assertThat(toList(acc)).containsExactly(1L, 2L, 3L);
+ }
+
+ @Test
+ void testInputLongerThanSize() {
+ FieldLastNAgg agg = new FieldLastNAgg(BIGINTS, 2);
+ assertThat(toList(agg.agg(longs(1), longs(2, 3, 4)))).containsExactly(3L, 4L);
+ }
+
+ @Test
+ void testNullAndEmptyInputs() {
+ FieldLastNAgg agg = new FieldLastNAgg(BIGINTS, 2);
+ GenericArray acc = longs(1, 2);
+ assertThat(agg.agg(acc, null)).isSameAs(acc);
+ assertThat(toList(agg.agg(acc, longs()))).containsExactly(1L, 2L);
+ }
+
+ @Test
+ void testInputExactlySize() {
+ FieldLastNAgg agg = new FieldLastNAgg(BIGINTS, 3);
+ GenericArray input = longs(1, 2, 3);
+ assertThat(agg.agg(null, input)).isSameAs(input);
+ assertThat(toList(agg.agg(longs(0), input))).containsExactly(1L, 2L, 3L);
+ }
+
+ @Test
+ void testEmptyInputWithoutAccumulator() {
+ FieldLastNAgg agg = new FieldLastNAgg(BIGINTS, 2);
+ assertThat(((InternalArray) agg.agg(null, longs())).size()).isZero();
+ }
+
+ @Test
+ void testNullElements() {
+ FieldLastNAgg agg = new FieldLastNAgg(BIGINTS, 3);
+ Object acc =
+ agg.agg(
+ new GenericArray(new Object[] {1L, null}),
+ new GenericArray(new Object[] {null, 4L}));
+ InternalArray result = (InternalArray) acc;
+ assertThat(result.size()).isEqualTo(3);
+ assertThat(result.isNullAt(0)).isTrue();
+ assertThat(result.isNullAt(1)).isTrue();
+ assertThat(result.getLong(2)).isEqualTo(4L);
+ }
+
+ @Test
+ void testReversedOrderTreatsInputAsOlder() {
+ FieldLastNAgg agg = new FieldLastNAgg(BIGINTS, 3);
+ assertThat(toList(agg.aggReversed(longs(3, 4), longs(1, 2)))).containsExactly(2L, 3L, 4L);
+ }
+
+ @Test
+ void testNonPrimitiveElements() {
+ FieldLastNAgg agg = new FieldLastNAgg(DataTypes.ARRAY(DataTypes.STRING()), 2);
+ Object acc = null;
+ for (String s : new String[] {"a", "b", "c"}) {
+ acc = agg.agg(acc, new GenericArray(new Object[] {BinaryString.fromString(s)}));
+ }
+ InternalArray result = (InternalArray) acc;
+ assertThat(result.size()).isEqualTo(2);
+ assertThat(result.getString(0).toString()).isEqualTo("b");
+ assertThat(result.getString(1).toString()).isEqualTo("c");
+ }
+
+ @Test
+ void testRequiresAggOnFirstWrite() {
+ assertThat(new FieldLastNAgg(BIGINTS, 2).requiresAggOnFirstWrite()).isTrue();
+ assertThat(toList(new FieldLastNAgg(BIGINTS, 2).agg(null, longs(1, 2, 3))))
+ .containsExactly(2L, 3L);
+ }
+
+ @Test
+ void testFactoryAndValidation() {
+ FieldAggregatorFactory factory = FieldAggregatorFactory.getFactory(AggFunctionType.LAST_N);
+ assertThat(factory.create(BIGINTS, AggFunctions.LAST_N(4)))
+ .isInstanceOf(FieldLastNAgg.class);
+ assertThatThrownBy(() -> factory.create(DataTypes.BIGINT(), AggFunctions.LAST_N(4)))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> AggFunctionType.LAST_N.validateParameter("size", "0"))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> AggFunctionType.LAST_N.validateParameter("size", "x"))
+ .isInstanceOf(IllegalArgumentException.class);
+ AggFunctionType.LAST_N.validateParameter("size", "1000");
+ AggFunctionType.LAST_N.validateDataType(BIGINTS);
+ assertThatThrownBy(() -> new FieldLastNAgg(BIGINTS, 0))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ private static GenericArray longs(long... values) {
+ return new GenericArray(values);
+ }
+
+ private static List<Long> toList(Object array) {
+ InternalArray a = (InternalArray) array;
+ List<Long> out = new ArrayList<>();
+ for (int i = 0; i < a.size(); i++) {
+ out.add(a.getLong(i));
+ }
+ return out;
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java
index 86853e6af..67c70e719 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/TableDescriptorValidationTest.java
@@ -23,7 +23,11 @@ import org.apache.fluss.exception.InvalidAlterTableException;
import org.apache.fluss.exception.InvalidConfigException;
import org.apache.fluss.exception.InvalidDatabaseException;
import org.apache.fluss.exception.InvalidTableException;
+import org.apache.fluss.metadata.AggFunction;
+import org.apache.fluss.metadata.AggFunctionType;
+import org.apache.fluss.metadata.AggFunctions;
import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.metadata.MergeEngineType;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
@@ -301,6 +305,32 @@ class TableDescriptorValidationTest {
.hasMessageContaining("Custom lake table path is only supported for Paimon");
}
+ @Test
+ void testCreateTableWithLastNWithoutSizeFails() {
+ assertThatCode(() -> validate(lastNTable(AggFunctions.LAST_N(3)), null))
+ .doesNotThrowAnyException();
+
+ assertThatThrownBy(
+ () -> validate(lastNTable(AggFunctions.of(AggFunctionType.LAST_N)), null))
+ .isInstanceOf(InvalidConfigException.class)
+ .hasMessageContaining("events")
+ .hasMessageContaining("'size' is required");
+ }
+
+ private static TableDescriptor lastNTable(AggFunction lastN) {
+ return TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("events", DataTypes.ARRAY(DataTypes.STRING()), lastN)
+ .primaryKey("id")
+ .build())
+ .distributedBy(1)
+ .property(ConfigOptions.TABLE_REPLICATION_FACTOR, 1)
+ .property(ConfigOptions.TABLE_MERGE_ENGINE, MergeEngineType.AGGREGATION)
+ .build();
+ }
+
private static Stream<Arguments> supportedKvTTLTimeColumnTypes() {
return Stream.of(
Arguments.of(DataTypes.BIGINT()),
diff --git a/website/docs/table-design/merge-engines/aggregation.md b/website/docs/table-design/merge-engines/aggregation.md
index fff9f7180..cc5150e4c 100644
--- a/website/docs/table-design/merge-engines/aggregation.md
+++ b/website/docs/table-design/merge-engines/aggregation.md
@@ -1090,6 +1090,70 @@ TableDescriptor.builder()
</TabItem>
</Tabs>
+### last_n
+
+Appends input arrays to the stored array and keeps the last `size` elements.
+
+- **Supported Data Types**: `ARRAY` of any element type
+- **Behavior**: Appends the input elements, then drops the oldest elements beyond `size`. The first write for a key is trimmed as well
+- **Ordering**: Elements are kept in the order writes reach the server. With the default `'sink.distribution-mode'`, Flink writes for the same key go through a single sink task, so the order follows the input. Independent clients writing to the same key may interleave
+- **Null Handling**: A null input leaves the column unchanged
+- **Size**: Required, a positive integer. Creating a table without it fails
+
+:::note
+Every update rewrites the whole stored array. With the default `'table.changelog.image' = 'full'`, both the old and the new array are also written to the changelog on every update. Keep `size` small.
+:::
+
+**Example:**
+<Tabs>
+<TabItem value="flink-sql" label="Flink SQL" default>
+
+```sql
+CREATE TABLE test_last_n (
+ user_id BIGINT,
+ recent_events ARRAY<STRING>,
+ PRIMARY KEY (user_id) NOT ENFORCED
+) WITH (
+ 'table.merge-engine' = 'aggregation',
+ 'fields.recent_events.agg' = 'last_n',
+ 'fields.recent_events.last_n.size' = '3'
+);
+
+INSERT INTO test_last_n VALUES
+ (1, ARRAY['login']),
+ (1, ARRAY['search']),
+ (1, ARRAY['view', 'add_to_cart']);
+
+SELECT * FROM test_last_n WHERE user_id = 1;
++------------+----------------------------------+
+| user_id | recent_events |
++------------+----------------------------------+
+| 1 | [search, view, add_to_cart] |
++------------+----------------------------------+
+```
+
+</TabItem>
+<TabItem value="java-client" label="Java Client">
+
+```java
+Schema schema = Schema.newBuilder()
+ .column("user_id", DataTypes.BIGINT())
+ .column("recent_events", DataTypes.ARRAY(DataTypes.STRING()), AggFunctions.LAST_N(3))
+ .primaryKey("user_id")
+ .build();
+
+TableDescriptor.builder()
+ .schema(schema)
+ .property("table.merge-engine", "aggregation")
+ .build();
+
+// Input: (1, ['login']), (1, ['search']), (1, ['view', 'add_to_cart'])
+// Result: (1, ['search', 'view', 'add_to_cart'])
+```
+
+</TabItem>
+</Tabs>
+
## Delete Behavior
The aggregation merge engine provides limited support for delete operations. You can configure the behavior using the `'table.delete.behavior'` option: