This is an automated email from the ASF dual-hosted git repository. biyan pushed a commit to branch release-0.6 in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 454fd9e75a76e21dd20b8d0ff8ac3447130860df Author: Jingsong <[email protected]> AuthorDate: Wed Dec 27 19:13:21 2023 +0800 [core] Fix nested-update to nested_update --- docs/content/concepts/primary-key-table.md | 8 +- .../compact/aggregate/FieldNestedUpdateAgg.java | 91 +++++++ .../apache/paimon/flink/PreAggregationITCase.java | 274 +++++++++++++++++++++ 3 files changed, 369 insertions(+), 4 deletions(-) diff --git a/docs/content/concepts/primary-key-table.md b/docs/content/concepts/primary-key-table.md index a79af8a01..541e5fd83 100644 --- a/docs/content/concepts/primary-key-table.md +++ b/docs/content/concepts/primary-key-table.md @@ -326,14 +326,14 @@ Current supported aggregate functions and data types are: The first_not_null_value function selects the first non-null value in a data set. It supports all data types. -* `nested-update`: - The nested-update function collects multiple rows into one array<row> (so-called 'nested table'). It supports ARRAY<ROW> data types. +* `nested_update`: + 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: - {{< tabs "nested-update-example" >}} + {{< tabs "nested_update-example" >}} {{< tab "Flink" >}} @@ -363,7 +363,7 @@ Current supported aggregate functions and data types are: sub_orders ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>> ) WITH ( 'merge-engine' = 'aggregation', - 'fields.sub_orders.aggregate-function' = 'nested-update', + 'fields.sub_orders.aggregate-function' = 'nested_update', 'fields.sub_orders.nested-key' = 'sub_order_id' ); 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 new file mode 100644 index 000000000..69e2736ce --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java @@ -0,0 +1,91 @@ +/* + * 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; + +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.RowType; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Used to update a field which representing a nested table. The data type of nested table field is + * {@code ARRAY<ROW>}. + */ +public class FieldNestedUpdateAgg extends FieldAggregator { + + public static final String NAME = "nested_update"; + + private final int nestedFields; + + @Nullable private final Projection keyProjection; + + 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 + String name() { + return NAME; + } + + @Override + public Object agg(Object accumulator, Object inputField) { + if (accumulator == null || inputField == null) { + return accumulator == null ? inputField : accumulator; + } + + InternalArray acc = (InternalArray) accumulator; + InternalArray input = (InternalArray) inputField; + + 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)); + } + + if (keyProjection != null) { + Map<BinaryRow, InternalRow> map = new HashMap<>(); + for (InternalRow row : rows) { + BinaryRow key = keyProjection.apply(row).copy(); + map.put(key, row); + } + + rows = new ArrayList<>(map.values()); + } + + return new GenericArray(rows.toArray()); + } +} 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 170d5d5fc..63670eb7c 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 @@ -18,18 +18,24 @@ package org.apache.paimon.flink; +import org.apache.paimon.mergetree.compact.aggregate.FieldNestedUpdateAgg; import org.apache.paimon.utils.BlockingIterator; +import org.apache.flink.table.api.config.ExecutionConfigOptions; import org.apache.flink.types.Row; import org.apache.flink.types.RowKind; import org.apache.flink.util.CloseableIterator; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import java.math.BigDecimal; import java.time.LocalDate; import java.time.LocalDateTime; +import java.util.Arrays; import java.util.Collections; +import java.util.Comparator; import java.util.List; +import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -969,4 +975,272 @@ public class PreAggregationITCase { .containsExactlyInAnyOrder(Row.of(1, 3, 1), Row.of(2, 1, 1)); } } + + /** ITCase for {@link FieldNestedUpdateAgg}. */ + public static class NestedUpdateAggregationITCase extends CatalogITCaseBase { + + @Override + protected List<String> ddl() { + String ordersTable = + "CREATE TABLE orders (\n" + + " order_id INT PRIMARY KEY NOT ENFORCED,\n" + + " user_name STRING,\n" + + " address STRING\n" + + ");"; + + String subordersTable = + "CREATE TABLE sub_orders (\n" + + " order_id INT,\n" + + " daily_id INT,\n" + + " today STRING,\n" + + " product_name STRING,\n" + + " price BIGINT,\n" + + " PRIMARY KEY (order_id, daily_id, today) NOT ENFORCED\n" + + ");"; + + String wideTable = + "CREATE TABLE order_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.nested-key' = 'daily_id,today',\n" + + " 'fields.sub_orders.ignore-retract' = 'true'," + + " 'fields.user_name.ignore-retract' = 'true'," + + " 'fields.address.ignore-retract' = 'true'" + + ")"; + + 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 + public void testUseCase() { + 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)," + + "(1, 2, '12-20', 'Tesla', 400000)," + + "(1, 1, '12-21', 'Sangsung', 5000)," + + "(2, 1, '12-20', 'Tea', 40)," + + "(2, 2, '12-20', 'Pot', 60)," + + "(3, 1, '12-25', 'Bat', 15)," + + "(3, 1, '12-26', 'Cup', 30)"); + + sql(widenSql()); + + List<Row> result = + sql("SELECT * FROM order_wide").stream() + .sorted(Comparator.comparingInt(r -> r.getFieldAs(0))) + .collect(Collectors.toList()); + + assertThat( + checkOneRecord( + result.get(0), + 1, + "Wang", + "HangZhou", + Row.of(1, "12-20", "Apple", 8000L), + Row.of(1, "12-21", "Sangsung", 5000L), + Row.of(2, "12-20", "Tesla", 400000L))) + .isTrue(); + assertThat( + checkOneRecord( + result.get(1), + 2, + "Zhao", + "ChengDu", + Row.of(1, "12-20", "Tea", 40L), + Row.of(2, "12-20", "Pot", 60L))) + .isTrue(); + assertThat( + checkOneRecord( + result.get(2), + 3, + "Liu", + "NanJing", + Row.of(1, "12-25", "Bat", 15L), + Row.of(1, "12-26", "Cup", 30L))) + .isTrue(); + + // query using UNNEST + List<Row> unnested = + sql( + "SELECT order_id, user_name, address, daily_id, today, product_name, price " + + "FROM order_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(1, "Wang", "HangZhou", 2, "12-20", "Tesla", 400000L), + Row.of(1, "Wang", "HangZhou", 1, "12-21", "Sangsung", 5000L), + Row.of(2, "Zhao", "ChengDu", 1, "12-20", "Tea", 40L), + Row.of(2, "Zhao", "ChengDu", 2, "12-20", "Pot", 60L), + Row.of(3, "Liu", "NanJing", 1, "12-25", "Bat", 15L), + 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 { + sEnv.getConfig() + .set( + ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE, + ExecutionConfigOptions.UpsertMaterialize.NONE); + + sql("INSERT INTO orders VALUES (1, 'Wang', 'HangZhou')"); + + sql( + "INSERT INTO sub_orders VALUES " + + "(1, 1, '12-20', 'Apple', 8000)," + + "(1, 2, '12-20', 'Tesla', 400000)," + + "(1, 1, '12-21', 'Sangsung', 5000)"); + + sEnv.executeSql(widenSql()); + + boolean checkResult; + List<Row> result; + + do { + Thread.sleep(500); + result = sql("SELECT * FROM order_wide"); + checkResult = + !result.isEmpty() + && checkOneRecord( + result.get(0), + 1, + "Wang", + "HangZhou", + Row.of(1, "12-20", "Apple", 8000L), + Row.of(1, "12-21", "Sangsung", 5000L), + Row.of(2, "12-20", "Tesla", 400000L)); + } while (!checkResult); + + sql("INSERT INTO sub_orders VALUES (1, 2, '12-20', 'Benz', 380000)"); + + do { + Thread.sleep(500); + result = sql("SELECT * FROM order_wide"); + checkResult = + !result.isEmpty() + && checkOneRecord( + result.get(0), + 1, + "Wang", + "HangZhou", + Row.of(1, "12-20", "Apple", 8000L), + Row.of(1, "12-21", "Sangsung", 5000L), + Row.of(2, "12-20", "Benz", 380000L)); + } while (!checkResult); + } + + private String widenSql() { + return "INSERT INTO order_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 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) { + return false; + } + if (!record.getFieldAs(1).equals(userName)) { + return false; + } + if (!record.getFieldAs(2).equals(address)) { + return false; + } + + return checkNestedTable(record.getFieldAs(3), subOrders); + } + + private boolean checkNestedTable(Row[] nestedTable, Row... subOrders) { + if (nestedTable.length != subOrders.length) { + return false; + } + + Comparator<Row> comparator = + (Comparator) + Comparator.comparingInt(r -> ((Row) r).getFieldAs(0)) + .thenComparing(r -> (String) ((Row) r).getField(1)); + + List<Row> sortedActual = + Arrays.stream(nestedTable).sorted(comparator).collect(Collectors.toList()); + List<Row> sortedExpected = + Arrays.stream(subOrders).sorted(comparator).collect(Collectors.toList()); + + for (int i = 0; i < sortedActual.size(); i++) { + if (!sortedActual.get(i).equals(sortedExpected.get(i))) { + return false; + } + } + + return true; + } + } }
