This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 b6da0bb3d [core] enable self-define FieldAggregator. (#4335)
b6da0bb3d is described below
commit b6da0bb3d43e6e0750323a840e32754141141dbb
Author: Wenchao Wu <[email protected]>
AuthorDate: Wed Oct 16 13:21:43 2024 +0800
[core] enable self-define FieldAggregator. (#4335)
---
.../compact/aggregate/FieldAggregator.java | 142 +++------------------
.../compact/aggregate/FieldBoolAndAgg.java | 2 +-
.../compact/aggregate/FieldBoolOrAgg.java | 2 +-
.../compact/aggregate/FieldCollectAgg.java | 2 +-
.../aggregate/FieldFirstNonNullValueAgg.java | 2 +-
.../compact/aggregate/FieldIgnoreRetractAgg.java | 2 +-
.../aggregate/FieldLastNonNullValueAgg.java | 2 +-
.../compact/aggregate/FieldLastValueAgg.java | 2 +-
.../compact/aggregate/FieldListaggAgg.java | 2 +-
.../mergetree/compact/aggregate/FieldMaxAgg.java | 2 +-
.../compact/aggregate/FieldMergeMapAgg.java | 2 +-
.../mergetree/compact/aggregate/FieldMinAgg.java | 2 +-
.../compact/aggregate/FieldNestedUpdateAgg.java | 2 +-
.../compact/aggregate/FieldPrimaryKeyAgg.java | 2 +-
.../compact/aggregate/FieldProductAgg.java | 2 +-
.../mergetree/compact/aggregate/FieldSumAgg.java | 2 +-
.../FieldAggregatorFactory.java} | 32 ++---
.../FieldBoolAndAggFactory.java} | 31 ++---
.../FieldBoolOrAggFactory.java} | 31 ++---
.../aggregate/factory/FieldCollectAggFactory.java | 44 +++++++
.../FieldFirstNonNullValueAggFactory.java} | 31 ++---
.../FieldFirstNonNullValueAggLegacyFactory.java} | 31 ++---
.../FieldFirstValueAggFactory.java} | 31 ++---
.../factory/FieldHllSketchAggFactory.java | 44 +++++++
.../FieldLastNonNullValueAggFactory.java} | 31 ++---
.../FieldLastValueAggFactory.java} | 31 ++---
.../FieldListaggAggFactory.java} | 31 ++---
.../FieldMaxAggFactory.java} | 31 ++---
.../FieldMergeMapAggFactory.java} | 44 +++----
.../FieldMinAggFactory.java} | 31 ++---
.../factory/FieldNestedUpdateAggFactory.java | 58 +++++++++
.../FieldPrimaryKeyAggFactory.java} | 31 ++---
.../FieldProductAggFactory.java} | 31 ++---
.../factory/FieldRoaringBitmap32AggFactory.java | 44 +++++++
.../factory/FieldRoaringBitmap64AggFactory.java | 44 +++++++
.../FieldSumAggFactory.java} | 31 ++---
.../factory/FieldThetaSketchAggFactory.java | 44 +++++++
.../services/org.apache.paimon.factories.Factory | 20 +++
.../compact/aggregate/FieldAggregatorTest.java | 15 +++
.../compact/aggregate/TestCostomAgg.java} | 20 +--
.../compact/aggregate/TestCostomAggFactory.java} | 28 ++--
.../services/org.apache.paimon.factories.Factory | 36 ++++++
42 files changed, 551 insertions(+), 497 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregator.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregator.java
index 2737c691d..b776f0a2e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregator.java
@@ -19,19 +19,13 @@
package org.apache.paimon.mergetree.compact.aggregate;
import org.apache.paimon.CoreOptions;
-import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.factories.FactoryUtil;
+import
org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
import org.apache.paimon.types.DataType;
-import org.apache.paimon.types.MapType;
-import org.apache.paimon.types.RowType;
-import org.apache.paimon.types.VarBinaryType;
import javax.annotation.Nullable;
import java.io.Serializable;
-import java.util.Collections;
-import java.util.List;
-
-import static org.apache.paimon.utils.Preconditions.checkArgument;
/** abstract class of aggregating a field of a row. */
public abstract class FieldAggregator implements Serializable {
@@ -52,129 +46,33 @@ public abstract class FieldAggregator implements
Serializable {
String field) {
FieldAggregator fieldAggregator;
if (isPrimaryKey) {
- fieldAggregator = new FieldPrimaryKeyAgg(fieldType);
- } else {
- // If the field has no aggregate function, use last_non_null_value.
- if (strAgg == null) {
- fieldAggregator = new FieldLastNonNullValueAgg(fieldType);
- } else {
- // ordered by type root definition
- switch (strAgg) {
- case FieldSumAgg.NAME:
- fieldAggregator = new FieldSumAgg(fieldType);
- break;
- case FieldMaxAgg.NAME:
- fieldAggregator = new FieldMaxAgg(fieldType);
- break;
- case FieldMinAgg.NAME:
- fieldAggregator = new FieldMinAgg(fieldType);
- break;
- case FieldLastNonNullValueAgg.NAME:
- fieldAggregator = new
FieldLastNonNullValueAgg(fieldType);
- break;
- case FieldLastValueAgg.NAME:
- fieldAggregator = new FieldLastValueAgg(fieldType);
- break;
- case FieldListaggAgg.NAME:
- fieldAggregator = new FieldListaggAgg(fieldType,
options, field);
- break;
- case FieldBoolOrAgg.NAME:
- fieldAggregator = new FieldBoolOrAgg(fieldType);
- break;
- case FieldBoolAndAgg.NAME:
- fieldAggregator = new FieldBoolAndAgg(fieldType);
- break;
- case FieldFirstValueAgg.NAME:
- fieldAggregator = new FieldFirstValueAgg(fieldType);
- break;
- case FieldFirstNonNullValueAgg.NAME:
- case FieldFirstNonNullValueAgg.LEGACY_NAME:
- fieldAggregator = new
FieldFirstNonNullValueAgg(fieldType);
- break;
- case FieldProductAgg.NAME:
- fieldAggregator = new FieldProductAgg(fieldType);
- break;
- case FieldNestedUpdateAgg.NAME:
- fieldAggregator =
- createFieldNestedUpdateAgg(
- fieldType,
options.fieldNestedUpdateAggNestedKey(field));
- break;
- case FieldCollectAgg.NAME:
- checkArgument(
- fieldType instanceof ArrayType,
- "Data type for collect column must be 'Array'
but was '%s'.",
- fieldType);
- fieldAggregator =
- new FieldCollectAgg(
- (ArrayType) fieldType,
-
options.fieldCollectAggDistinct(field));
- break;
- case FieldMergeMapAgg.NAME:
- checkArgument(
- fieldType instanceof MapType,
- "Data type of merge map column must be 'MAP'
but was '%s'",
- fieldType);
- fieldAggregator = new FieldMergeMapAgg((MapType)
fieldType);
- break;
- case FieldThetaSketchAgg.NAME:
- checkArgument(
- fieldType instanceof VarBinaryType,
- "Data type for theta sketch column must be
'VarBinaryType' but was '%s'.",
- fieldType);
- fieldAggregator = new
FieldThetaSketchAgg((VarBinaryType) fieldType);
- break;
- case FieldHllSketchAgg.NAME:
- checkArgument(
- fieldType instanceof VarBinaryType,
- "Data type for hll sketch column must be
'VarBinaryType' but was '%s'.",
- fieldType);
- fieldAggregator = new
FieldHllSketchAgg((VarBinaryType) fieldType);
- break;
- case FieldRoaringBitmap32Agg.NAME:
- checkArgument(
- fieldType instanceof VarBinaryType,
- "Data type for roaring bitmap column must be
'VarBinaryType' but was '%s'.",
- fieldType);
- fieldAggregator = new
FieldRoaringBitmap32Agg((VarBinaryType) fieldType);
- break;
- case FieldRoaringBitmap64Agg.NAME:
- checkArgument(
- fieldType instanceof VarBinaryType,
- "Data type for roaring bitmap column must be
'VarBinaryType' but was '%s'.",
- fieldType);
- fieldAggregator = new
FieldRoaringBitmap64Agg((VarBinaryType) fieldType);
- break;
- default:
- throw new RuntimeException(
- String.format(
- "Use unsupported aggregation: %s or
spell aggregate function incorrectly!",
- strAgg));
- }
- }
+ strAgg = FieldPrimaryKeyAgg.NAME;
+ } else if (strAgg == null) {
+ strAgg = FieldLastNonNullValueAgg.NAME;
}
- if (ignoreRetract) {
- fieldAggregator = new FieldIgnoreRetractAgg(fieldAggregator);
+ FieldAggregatorFactory fieldAggregatorFactory =
+ FactoryUtil.discoverFactory(
+ FieldAggregator.class.getClassLoader(),
+ FieldAggregatorFactory.class,
+ strAgg);
+ if (fieldAggregatorFactory == null) {
+ throw new RuntimeException(
+ String.format(
+ "Use unsupported aggregation: %s or spell
aggregate function incorrectly!",
+ strAgg));
}
- return fieldAggregator;
- }
+ fieldAggregator = fieldAggregatorFactory.create(fieldType, options,
field);
- private static FieldAggregator createFieldNestedUpdateAgg(
- DataType fieldType, List<String> nestedKey) {
- if (nestedKey == null) {
- nestedKey = Collections.emptyList();
+ if (ignoreRetract) {
+ fieldAggregator = new FieldIgnoreRetractAgg(fieldAggregator);
}
- String typeErrorMsg = "Data type of nested table column must be
'Array<Row>' but was '%s'.";
- checkArgument(fieldType instanceof ArrayType, typeErrorMsg, fieldType);
- ArrayType arrayType = (ArrayType) fieldType;
- checkArgument(arrayType.getElementType() instanceof RowType,
typeErrorMsg, fieldType);
-
- return new FieldNestedUpdateAgg(arrayType, nestedKey);
+ return fieldAggregator;
}
- abstract String name();
+ public abstract String name();
public abstract Object agg(Object accumulator, Object inputField);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldBoolAndAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldBoolAndAgg.java
index 0b6371309..5acf2595a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldBoolAndAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldBoolAndAgg.java
@@ -32,7 +32,7 @@ public class FieldBoolAndAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldBoolOrAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldBoolOrAgg.java
index ae385b3cc..03a0c1c3c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldBoolOrAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldBoolOrAgg.java
@@ -32,7 +32,7 @@ public class FieldBoolOrAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
index ddfae4de2..64ef223fa 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
@@ -85,7 +85,7 @@ public class FieldCollectAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
index 3d758d8f5..0bd950bbf 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
@@ -35,7 +35,7 @@ public class FieldFirstNonNullValueAgg extends
FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldIgnoreRetractAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldIgnoreRetractAgg.java
index 83f3f72a2..40772c6d1 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldIgnoreRetractAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldIgnoreRetractAgg.java
@@ -31,7 +31,7 @@ public class FieldIgnoreRetractAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return aggregator.name();
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastNonNullValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastNonNullValueAgg.java
index f069a914d..e189c20b2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastNonNullValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastNonNullValueAgg.java
@@ -32,7 +32,7 @@ public class FieldLastNonNullValueAgg extends FieldAggregator
{
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
index 9a4a5d4de..22c2b3da1 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
@@ -32,7 +32,7 @@ public class FieldLastValueAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldListaggAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldListaggAgg.java
index 25ee8cc24..3bde7e7cc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldListaggAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldListaggAgg.java
@@ -38,7 +38,7 @@ public class FieldListaggAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMaxAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMaxAgg.java
index 292c29510..34e7dd139 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMaxAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMaxAgg.java
@@ -34,7 +34,7 @@ public class FieldMaxAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMergeMapAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMergeMapAgg.java
index 8ba78ad5e..f597e8de5 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMergeMapAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMergeMapAgg.java
@@ -46,7 +46,7 @@ public class FieldMergeMapAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMinAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMinAgg.java
index 403724f25..4002966db 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMinAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMinAgg.java
@@ -34,7 +34,7 @@ public class FieldMinAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java
index 1f725bb7d..3cd29f127 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java
@@ -67,7 +67,7 @@ public class FieldNestedUpdateAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
index 58961bc5a..e8053be1e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
@@ -32,7 +32,7 @@ public class FieldPrimaryKeyAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldProductAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldProductAgg.java
index 1a0200246..c3fb18232 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldProductAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldProductAgg.java
@@ -37,7 +37,7 @@ public class FieldProductAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldSumAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldSumAgg.java
index 38081e20e..18e4bbeaf 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldSumAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldSumAgg.java
@@ -34,7 +34,7 @@ public class FieldSumAgg extends FieldAggregator {
}
@Override
- String name() {
+ public String name() {
return NAME;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldAggregatorFactory.java
similarity index 57%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldAggregatorFactory.java
index 9a4a5d4de..44f2439fe 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldAggregatorFactory.java
@@ -16,33 +16,17 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.factories.Factory;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
+/** Factory for {@link FieldAggregator}. */
+public interface FieldAggregatorFactory extends Factory {
- public static final String NAME = "last_value";
+ FieldAggregator create(DataType fieldType, CoreOptions options, String
field);
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
- @Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
- }
-
- @Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
- }
+ String identifier();
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldBoolAndAggFactory.java
similarity index 60%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldBoolAndAggFactory.java
index 9a4a5d4de..4a3a2dc88 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldBoolAndAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldBoolAndAgg;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldBoolAndAgg}. */
+public class FieldBoolAndAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldBoolAndAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
+ public String identifier() {
+ return FieldBoolAndAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldBoolOrAggFactory.java
similarity index 60%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldBoolOrAggFactory.java
index 9a4a5d4de..488325eae 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldBoolOrAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldBoolOrAgg;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldBoolOrAgg}. */
+public class FieldBoolOrAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldBoolOrAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
+ public String identifier() {
+ return FieldBoolOrAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldCollectAggFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldCollectAggFactory.java
new file mode 100644
index 000000000..b20453e07
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldCollectAggFactory.java
@@ -0,0 +1,44 @@
+/*
+ * 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.paimon.mergetree.compact.aggregate.factory;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldCollectAgg;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Factory for #{@link FieldCollectAgg}. */
+public class FieldCollectAggFactory implements FieldAggregatorFactory {
+ @Override
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ checkArgument(
+ fieldType instanceof ArrayType,
+ "Data type for collect column must be 'Array' but was '%s'.",
+ fieldType);
+ return new FieldCollectAgg((ArrayType) fieldType,
options.fieldCollectAggDistinct(field));
+ }
+
+ @Override
+ public String identifier() {
+ return FieldCollectAgg.NAME;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldFirstNonNullValueAggFactory.java
similarity index 59%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldFirstNonNullValueAggFactory.java
index 58961bc5a..141da6342 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldFirstNonNullValueAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldFirstNonNullValueAgg;
import org.apache.paimon.types.DataType;
-/** primary key aggregate a field of a row. */
-public class FieldPrimaryKeyAgg extends FieldAggregator {
-
- public static final String NAME = "primary-key";
-
- private static final long serialVersionUID = 1L;
-
- public FieldPrimaryKeyAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldFirstNonNullValueAgg}. */
+public class FieldFirstNonNullValueAggFactory implements
FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldFirstNonNullValueAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object inputField) {
- return inputField;
+ public String identifier() {
+ return FieldFirstNonNullValueAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldFirstNonNullValueAggLegacyFactory.java
similarity index 58%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldFirstNonNullValueAggLegacyFactory.java
index 58961bc5a..1d92dd3ec 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldFirstNonNullValueAggLegacyFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldFirstNonNullValueAgg;
import org.apache.paimon.types.DataType;
-/** primary key aggregate a field of a row. */
-public class FieldPrimaryKeyAgg extends FieldAggregator {
-
- public static final String NAME = "primary-key";
-
- private static final long serialVersionUID = 1L;
-
- public FieldPrimaryKeyAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for legacy name of #{@link FieldFirstNonNullValueAgg}. */
+public class FieldFirstNonNullValueAggLegacyFactory implements
FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldFirstNonNullValueAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object inputField) {
- return inputField;
+ public String identifier() {
+ return FieldFirstNonNullValueAgg.LEGACY_NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldFirstValueAggFactory.java
similarity index 60%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldFirstValueAggFactory.java
index 9a4a5d4de..cc36928c6 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldFirstValueAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldFirstValueAgg;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldFirstValueAgg}. */
+public class FieldFirstValueAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldFirstValueAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
+ public String identifier() {
+ return FieldFirstValueAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldHllSketchAggFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldHllSketchAggFactory.java
new file mode 100644
index 000000000..9f57abaee
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldHllSketchAggFactory.java
@@ -0,0 +1,44 @@
+/*
+ * 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.paimon.mergetree.compact.aggregate.factory;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldHllSketchAgg;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.VarBinaryType;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Factory for #{@link FieldHllSketchAgg}. */
+public class FieldHllSketchAggFactory implements FieldAggregatorFactory {
+ @Override
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ checkArgument(
+ fieldType instanceof VarBinaryType,
+ "Data type for hll sketch column must be 'VarBinaryType' but
was '%s'.",
+ fieldType);
+ return new FieldHllSketchAgg((VarBinaryType) fieldType);
+ }
+
+ @Override
+ public String identifier() {
+ return FieldHllSketchAgg.NAME;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldLastNonNullValueAggFactory.java
similarity index 59%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldLastNonNullValueAggFactory.java
index 58961bc5a..e3e2ff079 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldLastNonNullValueAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldLastNonNullValueAgg;
import org.apache.paimon.types.DataType;
-/** primary key aggregate a field of a row. */
-public class FieldPrimaryKeyAgg extends FieldAggregator {
-
- public static final String NAME = "primary-key";
-
- private static final long serialVersionUID = 1L;
-
- public FieldPrimaryKeyAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldLastNonNullValueAgg}. */
+public class FieldLastNonNullValueAggFactory implements FieldAggregatorFactory
{
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldLastNonNullValueAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object inputField) {
- return inputField;
+ public String identifier() {
+ return FieldLastNonNullValueAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldLastValueAggFactory.java
similarity index 60%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldLastValueAggFactory.java
index 9a4a5d4de..b3423a39e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldLastValueAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldLastValueAgg;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldLastValueAgg}. */
+public class FieldLastValueAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldLastValueAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
+ public String identifier() {
+ return FieldLastValueAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldListaggAggFactory.java
similarity index 60%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldListaggAggFactory.java
index 9a4a5d4de..e5e85dbab 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldListaggAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldListaggAgg;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldListaggAgg}. */
+public class FieldListaggAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldListaggAgg(fieldType, options, field);
}
@Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
+ public String identifier() {
+ return FieldListaggAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMaxAggFactory.java
similarity index 60%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMaxAggFactory.java
index 9a4a5d4de..2fda49a76 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMaxAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldMaxAgg;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldMaxAgg}. */
+public class FieldMaxAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldMaxAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
+ public String identifier() {
+ return FieldMaxAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMergeMapAggFactory.java
similarity index 50%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMergeMapAggFactory.java
index 3d758d8f5..ac2409b25 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMergeMapAggFactory.java
@@ -16,41 +16,29 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldMergeMapAgg;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
-/** first non-null value aggregate a field of a row. */
-public class FieldFirstNonNullValueAgg extends FieldAggregator {
-
- public static final String NAME = "first_non_null_value";
- public static final String LEGACY_NAME = "first_not_null_value";
-
- private static final long serialVersionUID = 1L;
-
- private boolean initialized;
-
- public FieldFirstNonNullValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+/** Factory for #{@link FieldMergeMapAgg}. */
+public class FieldMergeMapAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- if (!initialized && inputField != null) {
- initialized = true;
- return inputField;
- } else {
- return accumulator;
- }
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ checkArgument(
+ fieldType instanceof MapType,
+ "Data type of merge map column must be 'MAP' but was '%s'",
+ fieldType);
+ return new FieldMergeMapAgg((MapType) fieldType);
}
@Override
- public void reset() {
- this.initialized = false;
+ public String identifier() {
+ return FieldMergeMapAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMinAggFactory.java
similarity index 60%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMinAggFactory.java
index 9a4a5d4de..e939f8a73 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMinAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldMinAgg;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldMinAgg}. */
+public class FieldMinAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldMinAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
+ public String identifier() {
+ return FieldMinAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedUpdateAggFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedUpdateAggFactory.java
new file mode 100644
index 000000000..41b9f3e41
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedUpdateAggFactory.java
@@ -0,0 +1,58 @@
+/*
+ * 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.paimon.mergetree.compact.aggregate.factory;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldNestedUpdateAgg;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowType;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Factory for #{@link FieldNestedUpdateAgg}. */
+public class FieldNestedUpdateAggFactory implements FieldAggregatorFactory {
+ @Override
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return createFieldNestedUpdateAgg(fieldType,
options.fieldNestedUpdateAggNestedKey(field));
+ }
+
+ @Override
+ public String identifier() {
+ return FieldNestedUpdateAgg.NAME;
+ }
+
+ private static FieldAggregator createFieldNestedUpdateAgg(
+ DataType fieldType, List<String> nestedKey) {
+ if (nestedKey == null) {
+ nestedKey = Collections.emptyList();
+ }
+
+ String typeErrorMsg = "Data type of nested table column must be
'Array<Row>' but was '%s'.";
+ checkArgument(fieldType instanceof ArrayType, typeErrorMsg, fieldType);
+ ArrayType arrayType = (ArrayType) fieldType;
+ checkArgument(arrayType.getElementType() instanceof RowType,
typeErrorMsg, fieldType);
+
+ return new FieldNestedUpdateAgg(arrayType, nestedKey);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldPrimaryKeyAggFactory.java
similarity index 60%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldPrimaryKeyAggFactory.java
index 9a4a5d4de..312d29753 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldPrimaryKeyAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldPrimaryKeyAgg;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldPrimaryKeyAgg}. */
+public class FieldPrimaryKeyAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldPrimaryKeyAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
+ public String identifier() {
+ return FieldPrimaryKeyAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldProductAggFactory.java
similarity index 60%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldProductAggFactory.java
index 9a4a5d4de..88be9c42a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldProductAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldProductAgg;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldProductAgg}. */
+public class FieldProductAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldProductAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
+ public String identifier() {
+ return FieldProductAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldRoaringBitmap32AggFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldRoaringBitmap32AggFactory.java
new file mode 100644
index 000000000..5b2a80b30
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldRoaringBitmap32AggFactory.java
@@ -0,0 +1,44 @@
+/*
+ * 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.paimon.mergetree.compact.aggregate.factory;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldRoaringBitmap32Agg;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.VarBinaryType;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Factory for #{@link FieldRoaringBitmap32Agg}. */
+public class FieldRoaringBitmap32AggFactory implements FieldAggregatorFactory {
+ @Override
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ checkArgument(
+ fieldType instanceof VarBinaryType,
+ "Data type for roaring bitmap column must be 'VarBinaryType'
but was '%s'.",
+ fieldType);
+ return new FieldRoaringBitmap32Agg((VarBinaryType) fieldType);
+ }
+
+ @Override
+ public String identifier() {
+ return FieldRoaringBitmap32Agg.NAME;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldRoaringBitmap64AggFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldRoaringBitmap64AggFactory.java
new file mode 100644
index 000000000..a0e9fe652
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldRoaringBitmap64AggFactory.java
@@ -0,0 +1,44 @@
+/*
+ * 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.paimon.mergetree.compact.aggregate.factory;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldRoaringBitmap64Agg;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.VarBinaryType;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Factory for #{@link FieldRoaringBitmap64Agg}. */
+public class FieldRoaringBitmap64AggFactory implements FieldAggregatorFactory {
+ @Override
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ checkArgument(
+ fieldType instanceof VarBinaryType,
+ "Data type for roaring bitmap column must be 'VarBinaryType'
but was '%s'.",
+ fieldType);
+ return new FieldRoaringBitmap64Agg((VarBinaryType) fieldType);
+ }
+
+ @Override
+ public String identifier() {
+ return FieldRoaringBitmap64Agg.NAME;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldSumAggFactory.java
similarity index 60%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldSumAggFactory.java
index 9a4a5d4de..8470e6d17 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldSumAggFactory.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.mergetree.compact.aggregate;
+package org.apache.paimon.mergetree.compact.aggregate.factory;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldSumAgg;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** Factory for #{@link FieldSumAgg}. */
+public class FieldSumAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new FieldSumAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
+ public String identifier() {
+ return FieldSumAgg.NAME;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldThetaSketchAggFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldThetaSketchAggFactory.java
new file mode 100644
index 000000000..6aa5d7bc4
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldThetaSketchAggFactory.java
@@ -0,0 +1,44 @@
+/*
+ * 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.paimon.mergetree.compact.aggregate.factory;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.mergetree.compact.aggregate.FieldThetaSketchAgg;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.VarBinaryType;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Factory for #{@link FieldThetaSketchAgg}. */
+public class FieldThetaSketchAggFactory implements FieldAggregatorFactory {
+ @Override
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ checkArgument(
+ fieldType instanceof VarBinaryType,
+ "Data type for theta sketch column must be 'VarBinaryType' but
was '%s'.",
+ fieldType);
+ return new FieldThetaSketchAgg((VarBinaryType) fieldType);
+ }
+
+ @Override
+ public String identifier() {
+ return FieldThetaSketchAgg.NAME;
+ }
+}
diff --git
a/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index 0f87c96b0..ac6cc98fe 100644
---
a/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++
b/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -16,3 +16,23 @@
org.apache.paimon.catalog.FileSystemCatalogFactory
org.apache.paimon.jdbc.JdbcCatalogFactory
org.apache.paimon.jdbc.JdbcCatalogLockFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldBoolAndAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldBoolOrAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldCollectAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldFirstNonNullValueAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldFirstNonNullValueAggLegacyFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldFirstValueAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldHllSketchAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldLastNonNullValueAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldLastValueAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldListaggAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldMaxAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldMergeMapAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldMinAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldNestedUpdateAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldPrimaryKeyAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldProductAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldRoaringBitmap32AggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldRoaringBitmap64AggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldSumAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldThetaSketchAggFactory
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java
index 2d9a03b74..0fa1433e7 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java
@@ -810,6 +810,21 @@ public class FieldAggregatorTest {
assertThat(result4).isEqualTo(acc2);
}
+ @Test
+ public void testCustomAgg() throws IOException {
+ FieldAggregator fieldAggregator =
+ FieldAggregator.createFieldAggregator(
+ DataTypes.STRING(),
+ "custom",
+ false,
+ false,
+ CoreOptions.fromMap(new HashMap<>()),
+ "custom");
+
+ Object agg = fieldAggregator.agg("test", "test");
+ assertThat(agg).isEqualTo("test");
+ }
+
private Map<Object, Object> toMap(Object... kvs) {
Map<Object, Object> result = new HashMap<>();
for (int i = 0; i < kvs.length; i += 2) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/TestCostomAgg.java
similarity index 72%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/TestCostomAgg.java
index 9a4a5d4de..aedf6a373 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/TestCostomAgg.java
@@ -20,29 +20,19 @@ package org.apache.paimon.mergetree.compact.aggregate;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
+/** Custom FieldAggregator for Test. */
+public class TestCostomAgg extends FieldAggregator {
+ public TestCostomAgg(DataType dataType) {
super(dataType);
}
@Override
- String name() {
- return NAME;
+ public String name() {
+ return "custom";
}
@Override
public Object agg(Object accumulator, Object inputField) {
return inputField;
}
-
- @Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/TestCostomAggFactory.java
similarity index 63%
copy from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
copy to
paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/TestCostomAggFactory.java
index 9a4a5d4de..e8884bfb5 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/TestCostomAggFactory.java
@@ -18,31 +18,19 @@
package org.apache.paimon.mergetree.compact.aggregate;
+import org.apache.paimon.CoreOptions;
+import
org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
import org.apache.paimon.types.DataType;
-/** last value aggregate a field of a row. */
-public class FieldLastValueAgg extends FieldAggregator {
-
- public static final String NAME = "last_value";
-
- private static final long serialVersionUID = 1L;
-
- public FieldLastValueAgg(DataType dataType) {
- super(dataType);
- }
-
- @Override
- String name() {
- return NAME;
- }
-
+/** FieldAggregatorFactory for test. */
+public class TestCostomAggFactory implements FieldAggregatorFactory {
@Override
- public Object agg(Object accumulator, Object inputField) {
- return inputField;
+ public FieldAggregator create(DataType fieldType, CoreOptions options,
String field) {
+ return new TestCostomAgg(fieldType);
}
@Override
- public Object retract(Object accumulator, Object retractField) {
- return null;
+ public String identifier() {
+ return "custom";
}
}
diff --git
a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
new file mode 100644
index 000000000..3b85790c7
--- /dev/null
+++
b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -0,0 +1,36 @@
+# 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.
+
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldBoolAndAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldBoolOrAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldCollectAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldFirstNonNullValueAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldFirstNonNullValueAggLegacyFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldFirstValueAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldHllSketchAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldLastNonNullValueAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldLastValueAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldListaggAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldMaxAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldMergeMapAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldMinAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldNestedUpdateAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldPrimaryKeyAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldProductAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldRoaringBitmap32AggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldRoaringBitmap64AggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldSumAggFactory
+org.apache.paimon.mergetree.compact.aggregate.factory.FieldThetaSketchAggFactory
+org.apache.paimon.mergetree.compact.aggregate.TestCostomAggFactory
\ No newline at end of file
