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
The following commit(s) were added to refs/heads/release-0.6 by this push:
new 9d5e4fbb3 Revert "[core] Fix nested-update to nested_update"
9d5e4fbb3 is described below
commit 9d5e4fbb31852173537f162c0940ed8b2d79800a
Author: Yann <[email protected]>
AuthorDate: Tue Jan 16 16:45:32 2024 +0800
Revert "[core] Fix nested-update to nested_update"
This reverts commit 454fd9e75a76e21dd20b8d0ff8ac3447130860df.
---
docs/content/concepts/primary-key-table.md | 8 +-
.../compact/aggregate/FieldNestedUpdateAgg.java | 91 -------
.../apache/paimon/flink/PreAggregationITCase.java | 274 ---------------------
3 files changed, 4 insertions(+), 369 deletions(-)
diff --git a/docs/content/concepts/primary-key-table.md
b/docs/content/concepts/primary-key-table.md
index 541e5fd83..a79af8a01 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
deleted file mode 100644
index 69e2736ce..000000000
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * 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 63670eb7c..170d5d5fc 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,24 +18,18 @@
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;
@@ -975,272 +969,4 @@ 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;
- }
- }
}
