[core] enable self-define FieldAggregator. (#4335)

apache · Oct 16, 2024 · b6da0bb · b6da0bb
1 parent e53e8c5
commit b6da0bb
Show file tree

Hide file tree

Showing 42 changed files with 1,015 additions and 137 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
@@ -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 static FieldAggregator createFieldAggregator(
             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
@@ -32,7 +32,7 @@ public FieldBoolAndAgg(DataType dataType) {
     }
 
     @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
@@ -32,7 +32,7 @@ public FieldBoolOrAgg(DataType dataType) {
     }
 
     @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
@@ -85,7 +85,7 @@ public FieldCollectAgg(ArrayType dataType, boolean distinct) {
     }
 
     @Override
-    String name() {
+    public String name() {
         return NAME;
     }
 

diff --git a/...rc/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java b/...rc/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
@@ -35,7 +35,7 @@ public FieldFirstNonNullValueAgg(DataType dataType) {
     }
 
     @Override
-    String name() {
+    public String name() {
         return NAME;
     }
 

diff --git a/...re/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldIgnoreRetractAgg.java b/...re/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldIgnoreRetractAgg.java
@@ -31,7 +31,7 @@ public FieldIgnoreRetractAgg(FieldAggregator aggregator) {
     }
 
     @Override
-    String name() {
+    public String name() {
         return aggregator.name();
     }
 

diff --git a/...src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastNonNullValueAgg.java b/...src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastNonNullValueAgg.java
@@ -32,7 +32,7 @@ public FieldLastNonNullValueAgg(DataType dataType) {
     }
 
     @Override
-    String name() {
+    public String name() {
         return NAME;
     }
 

diff --git a/...n-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java b/...n-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldLastValueAgg.java
@@ -32,7 +32,7 @@ public FieldLastValueAgg(DataType dataType) {
     }
 
     @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
@@ -38,7 +38,7 @@ public FieldListaggAgg(DataType dataType, CoreOptions options, String field) {
     }
 
     @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
@@ -34,7 +34,7 @@ public FieldMaxAgg(DataType dataType) {
     }
 
     @Override
-    String name() {
+    public String name() {
         return NAME;
     }
 

diff --git a/...on-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMergeMapAgg.java b/...on-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldMergeMapAgg.java
@@ -46,7 +46,7 @@ public FieldMergeMapAgg(MapType dataType) {
     }
 
     @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
@@ -34,7 +34,7 @@ public FieldMinAgg(DataType dataType) {
     }
 
     @Override
-    String name() {
+    public String name() {
         return NAME;
     }
 

diff --git a/...ore/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java b/...ore/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java
@@ -67,7 +67,7 @@ public FieldNestedUpdateAgg(ArrayType dataType, List<String> nestedKey) {
     }
 
     @Override
-    String name() {
+    public String name() {
         return NAME;
     }
 

diff --git a/...-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java b/...-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldPrimaryKeyAgg.java
@@ -32,7 +32,7 @@ public FieldPrimaryKeyAgg(DataType dataType) {
     }
 
     @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
@@ -37,7 +37,7 @@ public FieldProductAgg(DataType dataType) {
     }
 
     @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
@@ -34,7 +34,7 @@ public FieldSumAgg(DataType dataType) {
     }
 
     @Override
-    String name() {
+    public String name() {
         return NAME;
     }
 

diff --git a/...in/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldAggregatorFactory.java b/...in/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldAggregatorFactory.java
@@ -0,0 +1,32 @@
+/*
+ * 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.factories.Factory;
+import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
+import org.apache.paimon.types.DataType;
+
+/** Factory for {@link FieldAggregator}. */
+public interface FieldAggregatorFactory extends Factory {
+
+    FieldAggregator create(DataType fieldType, CoreOptions options, String field);
+
+    String identifier();
+}
diff --git a/...in/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldBoolAndAggFactory.java b/...in/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldBoolAndAggFactory.java
@@ -0,0 +1,37 @@
+/*
+ * 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.FieldBoolAndAgg;
+import org.apache.paimon.types.DataType;
+
+/** Factory for #{@link FieldBoolAndAgg}. */
+public class FieldBoolAndAggFactory implements FieldAggregatorFactory {
+    @Override
+    public FieldAggregator create(DataType fieldType, CoreOptions options, String field) {
+        return new FieldBoolAndAgg(fieldType);
+    }
+
+    @Override
+    public String identifier() {
+        return FieldBoolAndAgg.NAME;
+    }
+}