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 fb5f88613 [core] Rename first_not_null_value agg function to
first_non_null_value (#2719)
fb5f88613 is described below
commit fb5f886135046606af68ea0007cb7ead922f196c
Author: tsreaper <[email protected]>
AuthorDate: Wed Jan 17 12:01:19 2024 +0800
[core] Rename first_not_null_value agg function to first_non_null_value
(#2719)
---
.../concepts/primary-key-table/merge-engine.md | 6 +--
.../compact/aggregate/FieldAggregator.java | 5 ++-
...alueAgg.java => FieldFirstNonNullValueAgg.java} | 7 ++--
.../compact/aggregate/FieldAggregatorTest.java | 18 ++++-----
.../apache/paimon/flink/PreAggregationITCase.java | 43 +++++++++++++---------
5 files changed, 44 insertions(+), 35 deletions(-)
diff --git a/docs/content/concepts/primary-key-table/merge-engine.md
b/docs/content/concepts/primary-key-table/merge-engine.md
index 199ba5018..de7540dc2 100644
--- a/docs/content/concepts/primary-key-table/merge-engine.md
+++ b/docs/content/concepts/primary-key-table/merge-engine.md
@@ -215,8 +215,8 @@ Current supported aggregate functions and data types are:
The first_value function retrieves the first null value from a data set.
It supports all data types.
-* `first_not_null_value`:
- The first_not_null_value function selects the first non-null value in a data
set.
+* `first_non_null_value`:
+ The first_non_null_value function selects the first non-null value in a data
set.
It supports all data types.
* `nested_update`:
@@ -295,7 +295,7 @@ Current supported aggregate functions and data types are:
* `merge_map`:
The merge_map function merge input maps. It only supports MAP type.
-Only `sum` and `product` supports retraction (`UPDATE_BEFORE` and `DELETE`),
others aggregate functions do not support retraction.
+Only `sum`, `product` and `count` supports retraction (`UPDATE_BEFORE` and
`DELETE`), others aggregate functions do not support retraction.
If you allow some functions to ignore retraction messages, you can configure:
`'fields.${field_name}.ignore-retract'='true'`.
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 954d9f63c..eee8fa3f9 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
@@ -84,8 +84,9 @@ public abstract class FieldAggregator implements Serializable
{
case FieldFirstValueAgg.NAME:
fieldAggregator = new FieldFirstValueAgg(fieldType);
break;
- case FieldFirstNotNullValueAgg.NAME:
- fieldAggregator = new
FieldFirstNotNullValueAgg(fieldType);
+ case FieldFirstNonNullValueAgg.NAME:
+ case FieldFirstNonNullValueAgg.LEGACY_NAME:
+ fieldAggregator = new
FieldFirstNonNullValueAgg(fieldType);
break;
case FieldCountAgg.NAME:
fieldAggregator = new FieldCountAgg(fieldType);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNotNullValueAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
similarity index 85%
rename from
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNotNullValueAgg.java
rename to
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
index fd79b468d..f2be36097 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNotNullValueAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
@@ -23,15 +23,16 @@ package org.apache.paimon.mergetree.compact.aggregate;
import org.apache.paimon.types.DataType;
/** first non-null value aggregate a field of a row. */
-public class FieldFirstNotNullValueAgg extends FieldAggregator {
+public class FieldFirstNonNullValueAgg extends FieldAggregator {
- public static final String NAME = "first_not_null_value";
+ public static final String NAME = "first_non_null_value";
+ public static final String LEGACY_NAME = "first_not_null_value";
private static final long serialVersionUID = 1L;
private boolean initialized;
- public FieldFirstNotNullValueAgg(DataType dataType) {
+ public FieldFirstNonNullValueAgg(DataType dataType) {
super(dataType);
}
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 d75813dd7..a13d3fbea 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
@@ -117,15 +117,15 @@ public class FieldAggregatorTest {
}
@Test
- public void testFieldFirstNotNullValueAgg() {
- FieldFirstNotNullValueAgg fieldFirstNotNullValueAgg =
- new FieldFirstNotNullValueAgg(new IntType());
- assertThat(fieldFirstNotNullValueAgg.agg(null, null)).isNull();
- assertThat(fieldFirstNotNullValueAgg.agg(null, 1)).isEqualTo(1);
- assertThat(fieldFirstNotNullValueAgg.agg(1, 2)).isEqualTo(1);
-
- fieldFirstNotNullValueAgg.reset();
- assertThat(fieldFirstNotNullValueAgg.agg(1, 3)).isEqualTo(3);
+ public void testFieldFirstNonNullValueAgg() {
+ FieldFirstNonNullValueAgg fieldFirstNonNullValueAgg =
+ new FieldFirstNonNullValueAgg(new IntType());
+ assertThat(fieldFirstNonNullValueAgg.agg(null, null)).isNull();
+ assertThat(fieldFirstNonNullValueAgg.agg(null, 1)).isEqualTo(1);
+ assertThat(fieldFirstNonNullValueAgg.agg(1, 2)).isEqualTo(1);
+
+ fieldFirstNonNullValueAgg.reset();
+ assertThat(fieldFirstNonNullValueAgg.agg(1, 3)).isEqualTo(3);
}
@Test
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 33e0b4cab..6525454c6 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
@@ -1061,11 +1061,13 @@ public class PreAggregationITCase {
+ "a INT,"
+ "b VARCHAR,"
+ "c VARCHAR,"
+ + "d VARCHAR,"
+ "PRIMARY KEY (k) NOT ENFORCED)"
+ " WITH ('merge-engine'='aggregation', "
+ "'changelog-producer' = 'full-compaction',"
+ "'fields.b.aggregate-function'='first_value',"
- +
"'fields.c.aggregate-function'='first_not_null_value',"
+ +
"'fields.c.aggregate-function'='first_non_null_value',"
+ +
"'fields.d.aggregate-function'='first_not_null_value',"
+ "'sequence.field'='a'"
+ ");",
"CREATE TABLE T2 ("
@@ -1083,41 +1085,46 @@ public class PreAggregationITCase {
public void tesInMemoryMerge() {
batchSql(
"INSERT INTO T VALUES "
- + "(1, 0, CAST(NULL AS VARCHAR), CAST(NULL AS
VARCHAR)),"
- + "(1, 1, '1', '1'), "
- + "(2, 2, '2', '2'),"
- + "(2, 3, '22', '22')");
+ + "(1, 0, CAST(NULL AS VARCHAR), CAST(NULL AS
VARCHAR), CAST(NULL AS VARCHAR)),"
+ + "(1, 1, '1', '1', '1'), "
+ + "(2, 2, '2', '2', '2'),"
+ + "(2, 3, '22', '22', '22')");
List<Row> result = batchSql("SELECT * FROM T");
assertThat(result)
- .containsExactlyInAnyOrder(Row.of(1, 1, null, "1"),
Row.of(2, 3, "2", "2"));
+ .containsExactlyInAnyOrder(
+ Row.of(1, 1, null, "1", "1"), Row.of(2, 3, "2",
"2", "2"));
}
@Test
public void tesUnOrderInput() {
batchSql(
"INSERT INTO T VALUES "
- + "(1, 0, CAST(NULL AS VARCHAR), CAST(NULL AS
VARCHAR)),"
- + "(1, 1, '1', '1'), "
- + "(2, 3, '2', '2'),"
- + "(2, 2, '22', '22')");
+ + "(1, 0, CAST(NULL AS VARCHAR), CAST(NULL AS
VARCHAR), CAST(NULL AS VARCHAR)),"
+ + "(1, 1, '1', '1', '1'), "
+ + "(2, 3, '2', '2', '2'),"
+ + "(2, 2, '22', '22', '22')");
List<Row> result = batchSql("SELECT * FROM T");
assertThat(result)
- .containsExactlyInAnyOrder(Row.of(1, 1, null, "1"),
Row.of(2, 3, "22", "22"));
- batchSql("INSERT INTO T VALUES (2, 1, '1', '1')");
+ .containsExactlyInAnyOrder(
+ Row.of(1, 1, null, "1", "1"), Row.of(2, 3, "22",
"22", "22"));
+ batchSql("INSERT INTO T VALUES (2, 1, '1', '1', '1')");
result = batchSql("SELECT * FROM T");
assertThat(result)
- .containsExactlyInAnyOrder(Row.of(1, 1, null, "1"),
Row.of(2, 3, "1", "1"));
+ .containsExactlyInAnyOrder(
+ Row.of(1, 1, null, "1", "1"), Row.of(2, 3, "1",
"1", "1"));
}
@Test
public void testMergeRead() {
- batchSql("INSERT INTO T VALUES (1, 1, CAST(NULL AS VARCHAR),
CAST(NULL AS VARCHAR))");
- batchSql("INSERT INTO T VALUES (1, 2, '1', '1')");
- batchSql("INSERT INTO T VALUES (2, 1, '2', '2')");
- batchSql("INSERT INTO T VALUES (2, 2, '22', '22')");
+ batchSql(
+ "INSERT INTO T VALUES (1, 1, CAST(NULL AS VARCHAR),
CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR))");
+ batchSql("INSERT INTO T VALUES (1, 2, '1', '1', '1')");
+ batchSql("INSERT INTO T VALUES (2, 1, '2', '2', '2')");
+ batchSql("INSERT INTO T VALUES (2, 2, '22', '22', '22')");
List<Row> result = batchSql("SELECT * FROM T");
assertThat(result)
- .containsExactlyInAnyOrder(Row.of(1, 2, null, "1"),
Row.of(2, 2, "2", "2"));
+ .containsExactlyInAnyOrder(
+ Row.of(1, 2, null, "1", "1"), Row.of(2, 2, "2",
"2", "2"));
}
@Test
