(paimon) branch master updated: [core] Add distinct parameter support to FieldListaggAgg operator (#6566)

[lzljs3620320]

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/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new db71b3cbd4 [core] Add distinct parameter support to FieldListaggAgg 
operator (#6566)
db71b3cbd4 is described below

commit db71b3cbd42756a84dc84c9277b32ff5a5ae5c69
Author: HOKNANG_LO <git...@lohoknang.com>
AuthorDate: Sun Nov 9 16:29:05 2025 +0800

    [core] Add distinct parameter support to FieldListaggAgg operator (#6566)
---
 .../compact/aggregate/FieldListaggAgg.java         |  7 ++++++
 .../compact/aggregate/FieldAggregatorTest.java     | 27 ++++++++++++++++++++++
 2 files changed, 34 insertions(+)

diff --git 
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldListaggAgg.java
 
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldListaggAgg.java
index 22843a6315..239e40bfac 100644
--- 
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldListaggAgg.java
+++ 
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldListaggAgg.java
@@ -30,9 +30,12 @@ public class FieldListaggAgg extends FieldAggregator {
 
     private final String delimiter;
 
+    private final boolean distinct;
+
     public FieldListaggAgg(String name, VarCharType dataType, CoreOptions 
options, String field) {
         super(name, dataType);
         this.delimiter = options.fieldListAggDelimiter(field);
+        this.distinct = options.fieldCollectAggDistinct(field);
     }
 
     @Override
@@ -46,6 +49,10 @@ public class FieldListaggAgg extends FieldAggregator {
         BinaryString mergeFieldSD = (BinaryString) accumulator;
         BinaryString inFieldSD = (BinaryString) inputField;
 
+        if (distinct && inFieldSD.getSizeInBytes() > 0 && 
mergeFieldSD.contains(inFieldSD)) {
+            return mergeFieldSD;
+        }
+
         return BinaryStringUtils.concat(
                 mergeFieldSD, BinaryString.fromString(delimiter), inFieldSD);
     }
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 4e09b993d5..1c8506c25c 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
@@ -77,9 +77,12 @@ import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
 
 import static org.apache.paimon.utils.ThetaSketch.sketchOf;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 
 /** test whether {@link FieldAggregator}' subclasses behaviors are expected. */
 public class FieldAggregatorTest {
@@ -160,6 +163,30 @@ public class FieldAggregatorTest {
                 .isEqualTo("user1,user2");
     }
 
+    @Test
+    public void testFieldListAggWithDefaultDelimiterAndDistinct() {
+        FieldListaggAgg fieldListaggAgg =
+                new FieldListaggAggFactory()
+                        .create(
+                                new VarCharType(),
+                                CoreOptions.fromMap(
+                                        
ImmutableMap.of("fields.fieldName.distinct", "true")),
+                                "fieldName");
+
+        BinaryString result =
+                Stream.of(
+                                BinaryString.fromString("user1"),
+                                BinaryString.fromString("user2"),
+                                BinaryString.fromString("user1"),
+                                BinaryString.fromString("user3"))
+                        .sequential()
+                        .reduce((l, r) -> (BinaryString) 
fieldListaggAgg.agg(l, r))
+                        .orElse(null);
+
+        assertNotNull(result);
+        assertEquals("user1,user2,user3", result.toString());
+    }
+
     @Test
     public void testFieldListAggWithCustomDelimiter() {
         FieldListaggAgg fieldListaggAgg =