tsreaper commented on code in PR #121: URL: https://github.com/apache/flink-table-store/pull/121#discussion_r883447651
########## docs/content/docs/development/create-table.md: ########## @@ -268,3 +268,39 @@ For example, the inputs: Output: - <1, 25.2, 20, 'This is a book'> + +## Aggregation Update + +You can configure partial update from options: + +```sql +CREATE TABLE MyTable ( + a STRING, + b INT, + c INT, + PRIMARY KEY (a) NOT ENFORCED +) WITH ( + 'merge-engine'='aggregation' Review Comment: We also need to specify what aggregate functions are used for each field. ########## flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/SumAggregateFunction.java: ########## @@ -0,0 +1,151 @@ +/* + * 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.flink.table.store.file.mergetree.compact; + +import java.io.Serializable; + +/** Custom column aggregation abstract class. */ +public interface SumAggregateFunction<T> extends Serializable { Review Comment: This interface can also be used by other aggregate functions, not only sum. Make this a more generic aggregate function interface. ########## flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/AggregateFunctionFactory.java: ########## @@ -0,0 +1,71 @@ +/* + * 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.flink.table.store.file.mergetree.compact; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; + +/** + * Aggregate Function Factory is used to get the aggregate type based on the configuration Each + * aggregate type has its own aggregate function factory Different implementation classes are given + * for different data types. + */ +public class AggregateFunctionFactory { + + /** SumFactory. */ + public static class SumAggregateFunctionFactory { + static SumAggregateFunction<?> choiceRightAggregateFunction(Class<?> c) { Review Comment: Use `LogicalType` instead of `Class<?>`. See `TypeUtils` class in `flink-table-store-common` module for an example. ########## flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/AggregateFunctionFactory.java: ########## @@ -0,0 +1,71 @@ +/* + * 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.flink.table.store.file.mergetree.compact; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; + +/** + * Aggregate Function Factory is used to get the aggregate type based on the configuration Each + * aggregate type has its own aggregate function factory Different implementation classes are given + * for different data types. + */ +public class AggregateFunctionFactory { + + /** SumFactory. */ + public static class SumAggregateFunctionFactory { + static SumAggregateFunction<?> choiceRightAggregateFunction(Class<?> c) { + SumAggregateFunction<?> f = null; + if (Double.class.equals(c)) { + f = new DoubleSumAggregateFunction(); + } else if (Long.class.equals(c)) { + f = new LongSumAggregateFunction(); + } else if (Integer.class.equals(c)) { + f = new IntegerSumAggregateFunction(); + } else if (Float.class.equals(c)) { + f = new FloatSumAggregateFunction(); + } + return f; + } + } + + public static AggregationKind getAggregationKind(Collection<String> values) { Review Comment: `Enum`s in Java has a `valueOf` method, which transform a `String` into the corresponding `Enum`. ########## flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/SumAggregateMergeFunction.java: ########## @@ -0,0 +1,142 @@ +/* + * 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.flink.table.store.file.mergetree.compact; + +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +/** + * A {@link MergeFunction} where key is primary key (unique) and value is the partial record, update + * non-null fields on merge. + */ +@SuppressWarnings("checkstyle:RegexpSingleline") +public class SumAggregateMergeFunction implements MergeFunction { + + private static final long serialVersionUID = 1L; + + private final RowData.FieldGetter[] getters; + + private final RowType rowType; + private final ArrayList<SumAggregateFunction<?>> aggregateFunctions; + private final boolean[] isPrimaryKey; + private final RowType primaryKeyType; + private transient GenericRowData row; + + private final Set<String> aggregateColumnNames; + + public SumAggregateMergeFunction( + RowType primaryKeyType, RowType rowType, Set<String> aggregateColumnNames) { + this.primaryKeyType = primaryKeyType; + this.rowType = rowType; + this.aggregateColumnNames = aggregateColumnNames; + + List<LogicalType> fieldTypes = rowType.getChildren(); + this.getters = new RowData.FieldGetter[fieldTypes.size()]; + for (int i = 0; i < fieldTypes.size(); i++) { + getters[i] = RowData.createFieldGetter(fieldTypes.get(i), i); + } + + this.isPrimaryKey = new boolean[this.getters.length]; + List<String> rowNames = rowType.getFieldNames(); + for (String primaryKeyName : primaryKeyType.getFieldNames()) { + isPrimaryKey[rowNames.indexOf(primaryKeyName)] = true; + } + + this.aggregateFunctions = new ArrayList<>(rowType.getFieldCount()); + for (int i = 0; i < rowType.getFieldCount(); i++) { + SumAggregateFunction<?> f = null; + if (aggregateColumnNames.contains(rowNames.get(i))) { + f = + AggregateFunctionFactory.SumAggregateFunctionFactory + .choiceRightAggregateFunction( + rowType.getTypeAt(i).getDefaultConversion()); + } else { + if (!isPrimaryKey[i]) { + throw new IllegalArgumentException( + "should set aggregate function for every column not part of primary key"); + } + } + aggregateFunctions.add(f); + } + } + + @Override + public void reset() { + this.row = new GenericRowData(getters.length); + } + + @Override + public void add(RowData value) { + for (int i = 0; i < getters.length; i++) { + Object currentField = getters[i].getFieldOrNull(value); + SumAggregateFunction<?> f = aggregateFunctions.get(i); + if (isPrimaryKey[i]) { + // primary key + if (currentField != null) { + row.setField(i, currentField); + } + } else { + if (f != null) { + f.reset(); + Object oldValue = row.getField(i); + if (oldValue != null) { + f.aggregate(oldValue); + } + switch (value.getRowKind()) { + case INSERT: + f.aggregate(currentField); + break; + case DELETE: + f.retract(currentField); Review Comment: See [my comments](https://github.com/apache/flink-table-store/pull/121#issuecomment-1132579400). I'm afraid currently we have no way to support retraction. ########## flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreImpl.java: ########## @@ -232,12 +235,29 @@ public static FileStoreImpl createWithPrimaryKey( mergeFunction = new DeduplicateMergeFunction(); break; case PARTIAL_UPDATE: - List<LogicalType> fieldTypes = rowType.getChildren(); - RowData.FieldGetter[] fieldGetters = new RowData.FieldGetter[fieldTypes.size()]; - for (int i = 0; i < fieldTypes.size(); i++) { - fieldGetters[i] = RowData.createFieldGetter(fieldTypes.get(i), i); + mergeFunction = new PartialUpdateMergeFunction(rowType); + break; + case AGGREGATION: + Map<String, String> rightConfMap = + options.getFilterConf(e -> e.getKey().endsWith(".aggregate-function")); + Set<String> aggregateColumnNames = + rightConfMap.keySet().stream() + .distinct() + .flatMap(s -> Stream.of(s.split(".aggregate-function")[0])) + .collect(Collectors.toSet()); + switch (AggregateFunctionFactory.getAggregationKind(rightConfMap.values())) { + case Sum: + mergeFunction = + new SumAggregateMergeFunction( + primaryKeyType, rowType, aggregateColumnNames); + break; + case Avg: + case Max: + case Min: + default: + throw new UnsupportedOperationException( + "merge-function values un supposed"); Review Comment: Extract this to a separated factory class. We'd like each method to be simple. Maybe the outer `switch` statement should also be extracted? If you're extracting both `switch` statements I would suggest creating 2 factory classes. Each class should complete one specific task. ########## flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/SumAggregateMergeFunction.java: ########## @@ -0,0 +1,142 @@ +/* + * 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.flink.table.store.file.mergetree.compact; + +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +/** + * A {@link MergeFunction} where key is primary key (unique) and value is the partial record, update + * non-null fields on merge. + */ +@SuppressWarnings("checkstyle:RegexpSingleline") +public class SumAggregateMergeFunction implements MergeFunction { Review Comment: Ditto. Make this into a more generic `MergeFunction`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
