You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/12/26 12:52:49 UTC
(incubator-paimon) branch master updated: [core] Improve Nested Update Aggregation Function
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/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e88944812 [core] Improve Nested Update Aggregation Function
e88944812 is described below
commit e889448126e0130ec9a272186af0b23d0380f3fb
Author: Jingsong <ji...@gmail.com>
AuthorDate: Tue Dec 26 20:52:37 2023 +0800
[core] Improve Nested Update Aggregation Function
---
docs/content/concepts/primary-key-table.md | 8 +--
.../main/java/org/apache/paimon/CoreOptions.java | 18 +++---
.../compact/PartialUpdateMergeFunction.java | 26 +++-----
.../compact/aggregate/AggregateMergeFunction.java | 23 +++----
.../compact/aggregate/FieldAggregator.java | 25 +++++---
.../compact/aggregate/FieldNestedUpdateAgg.java | 72 ++++++++--------------
.../compact/aggregate/FieldAggregatorTest.java | 24 ++++++++
.../apache/paimon/flink/PreAggregationITCase.java | 58 ++++++++++++++++-
8 files changed, 157 insertions(+), 97 deletions(-)
diff --git a/docs/content/concepts/primary-key-table.md b/docs/content/concepts/primary-key-table.md
index 61a4e54b4..bf6418387 100644
--- a/docs/content/concepts/primary-key-table.md
+++ b/docs/content/concepts/primary-key-table.md
@@ -327,9 +327,9 @@ Current supported aggregate functions and data types are:
It supports all data types.
* `nested-update`:
- The nested-update function collects multiple rows of a primary-key table into one single row
- (so-called 'nested table'). Use `fields.<field-name>.nested-keys=pk0;pk1;...` to specify the
- primary keys of the nested table. It supports ARRAY<ROW> data types.
+ The nested-update function collects multiple rows into one array<row> (so-called 'nested table'). It supports ARRAY<ROW> data types.
+
+ Use `fields.<field-name>.nested-key=pk0,pk1;...` to specify the primary keys of the nested table. If no keys, row will be appended to array<row>.
An example:
@@ -364,7 +364,7 @@ Current supported aggregate functions and data types are:
) WITH (
'merge-engine' = 'aggregation',
'fields.sub_orders.aggregate-function' = 'nested-update',
- 'fields.sub_orders.nested-keys' = 'sub_order_id'
+ 'fields.sub_orders.nested-key' = 'sub_order_id'
);
-- widen
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index c2009f774..84497806f 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -64,7 +64,7 @@ public class CoreOptions implements Serializable {
public static final String IGNORE_RETRACT = "ignore-retract";
- public static final String NESTED_KEYS = "nested-keys";
+ public static final String NESTED_KEY = "nested-key";
public static final ConfigOption<Integer> BUCKET =
key("bucket")
@@ -1095,12 +1095,16 @@ public class CoreOptions implements Serializable {
.defaultValue(false));
}
- public List<String> fieldNestedUpdateAggNestedKeys(String fieldName) {
- return options.get(
- key(FIELDS_PREFIX + "." + fieldName + "." + NESTED_KEYS)
- .stringType()
- .asList()
- .noDefaultValue());
+ public List<String> fieldNestedUpdateAggNestedKey(String fieldName) {
+ String keyString =
+ options.get(
+ key(FIELDS_PREFIX + "." + fieldName + "." + NESTED_KEY)
+ .stringType()
+ .noDefaultValue());
+ if (keyString == null) {
+ return Collections.emptyList();
+ }
+ return Arrays.asList(keyString.split(","));
}
public String fileCompression() {
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
index 3175ea546..8625530dc 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
@@ -23,7 +23,6 @@ import org.apache.paimon.KeyValue;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
-import org.apache.paimon.mergetree.compact.aggregate.FieldNestedUpdateAgg;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.sink.SequenceGenerator;
import org.apache.paimon.types.DataType;
@@ -45,7 +44,6 @@ import java.util.stream.Stream;
import static org.apache.paimon.CoreOptions.FIELDS_PREFIX;
import static org.apache.paimon.utils.InternalRowUtils.createFieldGetters;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
/**
* A {@link MergeFunction} where key is primary key (unique) and value is the partial record, update
@@ -375,21 +373,15 @@ public class PartialUpdateMergeFunction implements MergeFunction<KeyValue> {
boolean ignoreRetract = options.fieldAggIgnoreRetract(fieldName);
if (strAggFunc != null) {
- if (FieldAggregator.isNestedUpdateAgg(strAggFunc)) {
- List<String> nestedKeys = options.fieldNestedUpdateAggNestedKeys(fieldName);
- checkArgument(
- nestedKeys != null && !nestedKeys.isEmpty(),
- "Must set nested keys when using " + FieldNestedUpdateAgg.NAME);
- fieldAggregators.put(
- i,
- FieldAggregator.createFieldNestedUpdateAgg(
- fieldType, nestedKeys, ignoreRetract));
- } else {
- fieldAggregators.put(
- i,
- FieldAggregator.createFieldAggregator(
- fieldType, strAggFunc, ignoreRetract, isPrimaryKey));
- }
+ fieldAggregators.put(
+ i,
+ FieldAggregator.createFieldAggregator(
+ fieldType,
+ strAggFunc,
+ ignoreRetract,
+ isPrimaryKey,
+ options,
+ fieldName));
}
}
return fieldAggregators;
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
index b2613e9df..fd874e903 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
@@ -35,7 +35,6 @@ import java.util.Arrays;
import java.util.List;
import static org.apache.paimon.utils.InternalRowUtils.createFieldGetters;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
/**
@@ -140,20 +139,14 @@ public class AggregateMergeFunction implements MergeFunction<KeyValue> {
boolean isPrimaryKey = primaryKeys.contains(fieldName);
String strAggFunc = options.fieldAggFunc(fieldName);
boolean ignoreRetract = options.fieldAggIgnoreRetract(fieldName);
-
- if (FieldAggregator.isNestedUpdateAgg(strAggFunc)) {
- List<String> nestedKeys = options.fieldNestedUpdateAggNestedKeys(fieldName);
- checkArgument(
- nestedKeys != null && !nestedKeys.isEmpty(),
- "Must set nested keys when using " + FieldNestedUpdateAgg.NAME);
- fieldAggregators[i] =
- FieldAggregator.createFieldNestedUpdateAgg(
- fieldType, nestedKeys, ignoreRetract);
- } else {
- fieldAggregators[i] =
- FieldAggregator.createFieldAggregator(
- fieldType, strAggFunc, ignoreRetract, isPrimaryKey);
- }
+ fieldAggregators[i] =
+ FieldAggregator.createFieldAggregator(
+ fieldType,
+ strAggFunc,
+ ignoreRetract,
+ isPrimaryKey,
+ options,
+ fieldName);
}
return new AggregateMergeFunction(createFieldGetters(fieldTypes), fieldAggregators);
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 520ce10b0..5178576fe 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
@@ -18,6 +18,7 @@
package org.apache.paimon.mergetree.compact.aggregate;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowType;
@@ -25,6 +26,7 @@ import org.apache.paimon.types.RowType;
import javax.annotation.Nullable;
import java.io.Serializable;
+import java.util.Collections;
import java.util.List;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -41,7 +43,9 @@ public abstract class FieldAggregator implements Serializable {
DataType fieldType,
@Nullable String strAgg,
boolean ignoreRetract,
- boolean isPrimaryKey) {
+ boolean isPrimaryKey,
+ CoreOptions options,
+ String field) {
FieldAggregator fieldAggregator;
if (isPrimaryKey) {
fieldAggregator = new FieldPrimaryKeyAgg(fieldType);
@@ -88,6 +92,13 @@ public abstract class FieldAggregator implements Serializable {
case FieldProductAgg.NAME:
fieldAggregator = new FieldProductAgg(fieldType);
break;
+ case FieldNestedUpdateAgg.NAME:
+ fieldAggregator =
+ createFieldNestedUpdateAgg(
+ fieldType,
+ options.fieldNestedUpdateAggNestedKey(field),
+ ignoreRetract);
+ break;
default:
throw new RuntimeException(
String.format(
@@ -104,18 +115,18 @@ public abstract class FieldAggregator implements Serializable {
return fieldAggregator;
}
- public static boolean isNestedUpdateAgg(@Nullable String strAggFunc) {
- return FieldNestedUpdateAgg.NAME.equals(strAggFunc);
- }
-
public static FieldAggregator createFieldNestedUpdateAgg(
- DataType fieldType, List<String> nestedKeys, boolean ignoreRetract) {
+ DataType fieldType, List<String> nestedKey, boolean ignoreRetract) {
+ 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);
- FieldNestedUpdateAgg agg = new FieldNestedUpdateAgg(arrayType, nestedKeys);
+ FieldNestedUpdateAgg agg = new FieldNestedUpdateAgg(arrayType, nestedKey);
return ignoreRetract ? new FieldIgnoreRetractAgg(agg) : agg;
}
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 8cb47ae1b..9a3c79c79 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
@@ -18,11 +18,13 @@
package org.apache.paimon.mergetree.compact.aggregate;
+import org.apache.paimon.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.Projection;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.types.ArrayType;
-import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import javax.annotation.Nullable;
@@ -31,10 +33,6 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.function.BiFunction;
-import java.util.stream.Collectors;
-
-import static org.apache.paimon.utils.Preconditions.checkArgument;
/**
* Used to update a field which representing a nested table. The data type of nested table field is
@@ -44,25 +42,16 @@ public class FieldNestedUpdateAgg extends FieldAggregator {
public static final String NAME = "nested-update";
- private final BiFunction<InternalArray, Integer, InternalRow> rowGetter;
- private final List<InternalRow.FieldGetter> keyGetters;
-
- public FieldNestedUpdateAgg(ArrayType dataType, List<String> nestedKeys) {
- super(dataType);
- RowType rowType = (RowType) dataType.getElementType();
+ private final int nestedFields;
- InternalArray.ElementGetter objectGetter = InternalArray.createElementGetter(rowType);
- this.rowGetter = (array, pos) -> (InternalRow) objectGetter.getElementOrNull(array, pos);
+ @Nullable private final Projection keyProjection;
- this.keyGetters = new ArrayList<>(nestedKeys.size());
- List<DataField> dataFields = rowType.getFields();
- for (int i = 0; i < dataFields.size(); i++) {
- DataField dataField = dataFields.get(i);
- if (nestedKeys.contains(dataField.name())) {
- keyGetters.add(InternalRow.createFieldGetter(dataField.type(), i));
- }
- }
- checkArgument(keyGetters.size() == nestedKeys.size(), "You have set wrong nested keys.");
+ public FieldNestedUpdateAgg(ArrayType dataType, List<String> nestedKey) {
+ super(dataType);
+ RowType nestedType = (RowType) dataType.getElementType();
+ this.nestedFields = nestedType.getFieldCount();
+ this.keyProjection =
+ nestedKey.isEmpty() ? null : CodeGenUtils.newProjection(nestedType, nestedKey);
}
@Override
@@ -76,34 +65,27 @@ public class FieldNestedUpdateAgg extends FieldAggregator {
return accumulator == null ? inputField : accumulator;
}
- Map<List<Object>, InternalRow> unnestedAcc = unnest(accumulator);
- InternalArray inputs = (InternalArray) inputField;
+ InternalArray acc = (InternalArray) accumulator;
+ InternalArray input = (InternalArray) inputField;
- for (int i = 0; i < inputs.size(); i++) {
- InternalRow row = rowGetter.apply(inputs, i);
- List<Object> keys = getKeys(row);
- // update by nested keys
- unnestedAcc.put(keys, row);
+ List<InternalRow> rows = new ArrayList<>();
+ for (int i = 0; i < acc.size(); i++) {
+ rows.add(acc.getRow(i, nestedFields));
+ }
+ for (int i = 0; i < input.size(); i++) {
+ rows.add(input.getRow(i, nestedFields));
}
- return new GenericArray(unnestedAcc.values().toArray());
- }
-
- private Map<List<Object>, InternalRow> unnest(@Nullable Object accumulator) {
- Map<List<Object>, InternalRow> unnested = new HashMap<>();
- if (accumulator != null) {
- InternalArray array = (InternalArray) accumulator;
- for (int i = 0; i < array.size(); i++) {
- InternalRow row = rowGetter.apply(array, i);
- List<Object> keys = getKeys(row);
- unnested.put(keys, row);
+ if (keyProjection != null) {
+ Map<BinaryRow, InternalRow> map = new HashMap<>();
+ for (InternalRow row : rows) {
+ BinaryRow key = keyProjection.apply(row).copy();
+ map.put(key, row);
}
- }
- return unnested;
- }
+ rows = new ArrayList<>(map.values());
+ }
- private List<Object> getKeys(InternalRow row) {
- return keyGetters.stream().map(g -> g.getFieldOrNull(row)).collect(Collectors.toList());
+ return new GenericArray(rows.toArray());
}
}
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 62f1ff28b..a98c3b3e6 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
@@ -314,6 +314,30 @@ public class FieldAggregatorTest {
.containsExactlyInAnyOrderElementsOf(Arrays.asList(row(0, 0, "A"), row(0, 1, "b")));
}
+ @Test
+ public void testFieldNestedAppendAgg() {
+ FieldNestedUpdateAgg agg =
+ new FieldNestedUpdateAgg(
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "k0", DataTypes.INT()),
+ DataTypes.FIELD(1, "k1", DataTypes.INT()),
+ DataTypes.FIELD(2, "v", DataTypes.STRING()))),
+ Collections.emptyList());
+
+ InternalArray accumulator = null;
+
+ InternalRow current = row(0, 1, "B");
+ accumulator = (InternalArray) agg.agg(accumulator, singletonArray(current));
+ assertThat(unnest(accumulator))
+ .containsExactlyInAnyOrderElementsOf(Collections.singletonList(row(0, 1, "B")));
+
+ current = row(0, 1, "b");
+ accumulator = (InternalArray) agg.agg(accumulator, singletonArray(current));
+ assertThat(unnest(accumulator))
+ .containsExactlyInAnyOrderElementsOf(Arrays.asList(row(0, 1, "B"), row(0, 1, "b")));
+ }
+
private List<InternalRow> unnest(InternalArray array) {
return IntStream.range(0, array.size())
.mapToObj(i -> array.getRow(i, 3))
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
index 98cfff513..9b5fb6dbf 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
@@ -1007,13 +1007,27 @@ public class PreAggregationITCase {
+ ") WITH (\n"
+ " 'merge-engine' = 'aggregation',\n"
+ " 'fields.sub_orders.aggregate-function' = 'nested-update',\n"
- + " 'fields.sub_orders.nested-keys' = 'daily_id;today',\n"
+ + " 'fields.sub_orders.nested-key' = 'daily_id,today',\n"
+ " 'fields.sub_orders.ignore-retract' = 'true',"
+ " 'fields.user_name.ignore-retract' = 'true',"
+ " 'fields.address.ignore-retract' = 'true'"
+ ")";
- return Arrays.asList(ordersTable, subordersTable, wideTable);
+ String wideAppendTable =
+ "CREATE TABLE order_append_wide (\n"
+ + " order_id INT PRIMARY KEY NOT ENFORCED,\n"
+ + " user_name STRING,\n"
+ + " address STRING,\n"
+ + " sub_orders ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>\n"
+ + ") WITH (\n"
+ + " 'merge-engine' = 'aggregation',\n"
+ + " 'fields.sub_orders.aggregate-function' = 'nested-update',\n"
+ + " 'fields.sub_orders.ignore-retract' = 'true',"
+ + " 'fields.user_name.ignore-retract' = 'true',"
+ + " 'fields.address.ignore-retract' = 'true'"
+ + ")";
+
+ return Arrays.asList(ordersTable, subordersTable, wideTable, wideAppendTable);
}
@Test
@@ -1087,6 +1101,37 @@ public class PreAggregationITCase {
Row.of(3, "Liu", "NanJing", 1, "12-26", "Cup", 30L));
}
+ @Test
+ public void testUseCaseAppend() {
+ sql(
+ "INSERT INTO orders VALUES "
+ + "(1, 'Wang', 'HangZhou'),"
+ + "(2, 'Zhao', 'ChengDu'),"
+ + "(3, 'Liu', 'NanJing')");
+
+ sql(
+ "INSERT INTO sub_orders VALUES "
+ + "(1, 1, '12-20', 'Apple', 8000),"
+ + "(2, 1, '12-20', 'Tesla', 400000),"
+ + "(3, 1, '12-25', 'Bat', 15),"
+ + "(3, 1, '12-26', 'Cup', 30)");
+
+ sql(widenAppendSql());
+
+ // query using UNNEST
+ List<Row> unnested =
+ sql(
+ "SELECT order_id, user_name, address, daily_id, today, product_name, price "
+ + "FROM order_append_wide, UNNEST(sub_orders) AS so(daily_id, today, product_name, price)");
+
+ assertThat(unnested)
+ .containsExactlyInAnyOrder(
+ Row.of(1, "Wang", "HangZhou", 1, "12-20", "Apple", 8000L),
+ Row.of(2, "Zhao", "ChengDu", 1, "12-20", "Tesla", 400000L),
+ Row.of(3, "Liu", "NanJing", 1, "12-25", "Bat", 15L),
+ Row.of(3, "Liu", "NanJing", 1, "12-26", "Cup", 30L));
+ }
+
@Test
@Timeout(60)
public void testUpdateWithIgnoreRetract() throws Exception {
@@ -1150,6 +1195,15 @@ public class PreAggregationITCase {
+ "ARRAY[ROW(daily_id, today, product_name, price)] FROM sub_orders";
}
+ private String widenAppendSql() {
+ return "INSERT INTO order_append_wide\n"
+ + "SELECT order_id, user_name, address, "
+ + "CAST (NULL AS ARRAY<ROW<daily_id INT, today STRING, product_name STRING, price BIGINT>>) FROM orders\n"
+ + "UNION ALL\n"
+ + "SELECT order_id, CAST (NULL AS STRING), CAST (NULL AS STRING), "
+ + "ARRAY[ROW(daily_id, today, product_name, price)] FROM sub_orders";
+ }
+
private boolean checkOneRecord(
Row record, int orderId, String userName, String address, Row... subOrders) {
if ((int) record.getField(0) != orderId) {