feat(aggregate): add Java-compatible field aggregators

cmake_modules/DefineOptions.cmake

@@ -223,6 +223,12 @@ if("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_CURRENT_SOURCE_DIR}")
                          AUTO
                          BUNDLED
                          SYSTEM)
+    define_option_string(DataSketches_SOURCE
+                         "Dependency source for DataSketches"
+                         ""
+                         AUTO
+                         BUNDLED
+                         SYSTEM)
     define_option_string(TBB_SOURCE
                          "Dependency source for TBB"
                          ""

cmake_modules/FindDataSketchesAlt.cmake

@@ -0,0 +1,39 @@
+# Copyright 2026-present Alibaba Inc.
+#
+# Licensed 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.
+
+# Headers are located directly: the installed config exports unnamespaced targets like `hll`.
+
+set(_PAIMON_DATASKETCHES_ROOTS ${DataSketches_ROOT} ${DATASKETCHES_ROOT}
+                               ${PAIMON_PACKAGE_PREFIX})
+list(REMOVE_ITEM _PAIMON_DATASKETCHES_ROOTS "")
+if(_PAIMON_DATASKETCHES_ROOTS)
+    set(_PAIMON_DATASKETCHES_FIND_ARGS HINTS ${_PAIMON_DATASKETCHES_ROOTS}
+                                       NO_DEFAULT_PATH)
+endif()
+
+find_path(DATASKETCHES_INCLUDE_DIR
+          NAMES DataSketches/hll.hpp ${_PAIMON_DATASKETCHES_FIND_ARGS}
+          PATH_SUFFIXES include)
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(DataSketchesAlt REQUIRED_VARS DATASKETCHES_INCLUDE_DIR)
+
+if(DataSketchesAlt_FOUND AND NOT TARGET DataSketches)
+    add_library(DataSketches INTERFACE IMPORTED)
+    target_include_directories(DataSketches SYSTEM
+                               INTERFACE "${DATASKETCHES_INCLUDE_DIR}")
+endif()
+
+unset(_PAIMON_DATASKETCHES_FIND_ARGS)
+unset(_PAIMON_DATASKETCHES_ROOTS)

cmake_modules/ThirdpartyToolchain.cmake

@@ -103,6 +103,19 @@ else()
     endif()
 endif()
 
+if(DEFINED ENV{PAIMON_DATASKETCHES_URL})
+    set(DATASKETCHES_SOURCE_URL "$ENV{PAIMON_DATASKETCHES_URL}")
+else()
+    if(EXISTS "${THIRDPARTY_DIR}/${PAIMON_DATASKETCHES_PKG_NAME}")
+        set_urls(DATASKETCHES_SOURCE_URL
+                 "${THIRDPARTY_DIR}/${PAIMON_DATASKETCHES_PKG_NAME}")
+    else()
+        set_urls(DATASKETCHES_SOURCE_URL
+                 "${THIRDPARTY_MIRROR_URL}https://github.com/apache/datasketches-cpp/archive/refs/tags/${PAIMON_DATASKETCHES_BUILD_VERSION}.tar.gz"
+        )
+    endif()
+endif()
+
 if(DEFINED ENV{PAIMON_FMT_URL})
     set(FMT_SOURCE_URL "$ENV{PAIMON_FMT_URL}")
 else()
@@ -600,6 +613,8 @@ macro(paimon_build_dependency DEPENDENCY_NAME)
         build_fmt()
     elseif("${DEPENDENCY_NAME}" STREQUAL "RapidJSON")
         build_rapidjson()
+    elseif("${DEPENDENCY_NAME}" STREQUAL "DataSketches")
+        build_datasketches()
     elseif("${DEPENDENCY_NAME}" STREQUAL "zstd")
         build_zstd()
     elseif("${DEPENDENCY_NAME}" STREQUAL "Snappy")
@@ -883,6 +898,26 @@ macro(build_rapidjson)
     add_dependencies(RapidJSON rapidjson_ep)
 endmacro()
 
+macro(build_datasketches)
+    message(STATUS "Building DataSketches from source")
+    set(DATASKETCHES_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/datasketches_ep-install")
+    set(DATASKETCHES_INCLUDE_DIR "${DATASKETCHES_PREFIX}/include")
+    set(DATASKETCHES_CMAKE_ARGS ${EP_COMMON_CMAKE_ARGS} -DBUILD_TESTS=OFF
+                                "-DCMAKE_INSTALL_PREFIX=${DATASKETCHES_PREFIX}")
+
+    externalproject_add(datasketches_ep
+                        ${EP_COMMON_OPTIONS}
+                        URL ${DATASKETCHES_SOURCE_URL}
+                        URL_HASH "SHA256=${PAIMON_DATASKETCHES_BUILD_SHA256_CHECKSUM}"
+                        CMAKE_ARGS ${DATASKETCHES_CMAKE_ARGS})
+
+    file(MAKE_DIRECTORY "${DATASKETCHES_INCLUDE_DIR}")
+    add_library(DataSketches INTERFACE IMPORTED)
+    target_include_directories(DataSketches SYSTEM
+                               INTERFACE "${DATASKETCHES_INCLUDE_DIR}")
+    add_dependencies(DataSketches datasketches_ep)
+endmacro()
+
 macro(build_fmt)
     message(STATUS "Building fmt from source")
     set(FMT_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/fmt_ep-install")
@@ -1862,6 +1897,7 @@ endmacro()
 
 resolve_dependency(fmt)
 resolve_dependency(RapidJSON)
+resolve_dependency(DataSketches)
 paimon_apply_dependency_source_defaults()
 resolve_dependency(RE2)
 resolve_dependency(Snappy)

include/paimon/defs.h

@@ -80,6 +80,14 @@ struct PAIMON_EXPORT Options {
     static const char DEFAULT_AGG_FUNCTION[];
     /// IGNORE_RETRACT is "ignore-retract"
     static const char IGNORE_RETRACT[];
+    /// NESTED_KEY is "nested-key"
+    static const char NESTED_KEY[];
+    /// NESTED_KEY_NULL_STRATEGY is "nested-key-null-strategy"
+    static const char NESTED_KEY_NULL_STRATEGY[];
+    /// NESTED_SEQUENCE_FIELD is "nested-sequence-field"
+    static const char NESTED_SEQUENCE_FIELD[];
+    /// COUNT_LIMIT is "count-limit"
+    static const char COUNT_LIMIT[];
     /// "distinct" - Distinct option for aggregate functions like listagg. Default value is false.
     /// Example: fields.f.distinct=true to deduplicate values during aggregation.
     static const char DISTINCT[];

src/paimon/CMakeLists.txt

@@ -32,6 +32,8 @@ set(PAIMON_COMMON_SRCS
     common/data/columnar/columnar_row.cpp
     common/data/columnar/columnar_row_ref.cpp
     common/data/decimal.cpp
+    common/data/generic_array.cpp
+    common/data/generic_map.cpp
     common/data/internal_row.cpp
     common/data/record_batch.cpp
     common/data/serializer/binary_row_serializer.cpp
@@ -286,6 +288,12 @@ set(PAIMON_CORE_SRCS
     core/mergetree/compact/universal_compaction.cpp
     core/mergetree/compact/early_full_compaction.cpp
     core/mergetree/compact/aggregate/aggregate_merge_function.cpp
+    core/mergetree/compact/aggregate/field_aggregate_utils.cpp
+    core/mergetree/compact/aggregate/field_collect_agg.cpp
+    core/mergetree/compact/aggregate/field_merge_map_agg.cpp
+    core/mergetree/compact/aggregate/field_nested_update_agg.cpp
+    core/mergetree/compact/aggregate/field_roaring_bitmap64_agg.cpp
+    core/mergetree/compact/aggregate/field_sketch_agg.cpp
     core/mergetree/compact/aggregate/field_sum_agg.cpp
     core/mergetree/compact/interval_partition.cpp
     core/mergetree/compact/loser_tree.cpp
@@ -416,6 +424,7 @@ add_paimon_lib(paimon
                xxhash
                Threads::Threads
                RapidJSON
+               DataSketches
                ${PAIMON_OBJECT_STORE_LINK_LIBS}
                STATIC_LINK_LIBS
                arrow
@@ -426,6 +435,7 @@ add_paimon_lib(paimon
                xxhash
                Threads::Threads
                RapidJSON
+               DataSketches
                ${PAIMON_OBJECT_STORE_LINK_LIBS}
                SHARED_LINK_FLAGS
                ${PAIMON_VERSION_SCRIPT_FLAGS})
@@ -707,14 +717,19 @@ if(PAIMON_BUILD_TESTS)
                     core/mergetree/compact/aggregate/aggregate_merge_function_test.cpp
                     core/mergetree/compact/aggregate/field_aggregator_factory_test.cpp
                     core/mergetree/compact/aggregate/field_bool_agg_test.cpp
+                    core/mergetree/compact/aggregate/field_collect_agg_test.cpp
                     core/mergetree/compact/aggregate/field_first_non_null_value_agg_test.cpp
                     core/mergetree/compact/aggregate/field_first_value_agg_test.cpp
                     core/mergetree/compact/aggregate/field_ignore_retract_agg_test.cpp
                     core/mergetree/compact/aggregate/field_last_non_null_value_agg_test.cpp
                     core/mergetree/compact/aggregate/field_last_value_agg_test.cpp
                     core/mergetree/compact/aggregate/field_listagg_agg_test.cpp
+                    core/mergetree/compact/aggregate/field_merge_map_agg_test.cpp
                     core/mergetree/compact/aggregate/field_min_max_agg_test.cpp
+                    core/mergetree/compact/aggregate/field_nested_update_agg_test.cpp
                     core/mergetree/compact/aggregate/field_primary_key_agg_test.cpp
+                    core/mergetree/compact/aggregate/field_roaring_bitmap64_agg_test.cpp
+                    core/mergetree/compact/aggregate/field_sketch_agg_test.cpp
                     core/mergetree/compact/aggregate/field_sum_agg_test.cpp
                     core/mergetree/compact/deduplicate_merge_function_test.cpp
                     core/mergetree/compact/first_row_merge_function_test.cpp
@@ -827,6 +842,7 @@ if(PAIMON_BUILD_TESTS)
                     STATIC_LINK_LIBS
                     paimon_shared
                     test_utils_static
+                    DataSketches
                     ${TEST_STATIC_LINK_LIBS}
                     ${GTEST_LINK_TOOLCHAIN})
 

src/paimon/common/data/generic_array.cpp

@@ -0,0 +1,164 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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.
+ */
+
+#include "paimon/common/data/generic_array.h"
+
+#include <cassert>
+#include <string>
+
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+int32_t GenericArray::Size() const {
+    return static_cast<int32_t>(values_.size());
+}
+
+const VariantType& GenericArray::ValueAt(int32_t pos) const {
+    assert(pos >= 0 && static_cast<size_t>(pos) < values_.size());
+    return values_[pos];
+}
+
+bool GenericArray::IsNullAt(int32_t pos) const {
+    return DataDefine::IsVariantNull(ValueAt(pos));
+}
+
+bool GenericArray::GetBoolean(int32_t pos) const {
+    return DataDefine::GetVariantValue<bool>(ValueAt(pos));
+}
+
+char GenericArray::GetByte(int32_t pos) const {
+    return DataDefine::GetVariantValue<char>(ValueAt(pos));
+}
+
+int16_t GenericArray::GetShort(int32_t pos) const {
+    return DataDefine::GetVariantValue<int16_t>(ValueAt(pos));
+}
+
+int32_t GenericArray::GetInt(int32_t pos) const {
+    return DataDefine::GetVariantValue<int32_t>(ValueAt(pos));
+}
+
+int32_t GenericArray::GetDate(int32_t pos) const {
+    return GetInt(pos);
+}
+
+int64_t GenericArray::GetLong(int32_t pos) const {
+    return DataDefine::GetVariantValue<int64_t>(ValueAt(pos));
+}
+
+float GenericArray::GetFloat(int32_t pos) const {
+    return DataDefine::GetVariantValue<float>(ValueAt(pos));
+}
+
+double GenericArray::GetDouble(int32_t pos) const {
+    return DataDefine::GetVariantValue<double>(ValueAt(pos));
+}
+
+BinaryString GenericArray::GetString(int32_t pos) const {
+    const VariantType& value = ValueAt(pos);
+    if (const auto* string = DataDefine::GetVariantPtr<BinaryString>(value)) {
+        return *string;
+    }
+    return BinaryString::FromString(std::string(DataDefine::GetStringView(value)),
+                                    GetDefaultPool().get());
+}
+
+std::string_view GenericArray::GetStringView(int32_t pos) const {
+    return DataDefine::GetStringView(ValueAt(pos));
+}
+
+Decimal GenericArray::GetDecimal(int32_t pos, int32_t precision, int32_t scale) const {
+    return DataDefine::GetVariantValue<Decimal>(ValueAt(pos));
+}
+
+Timestamp GenericArray::GetTimestamp(int32_t pos, int32_t precision) const {
+    return DataDefine::GetVariantValue<Timestamp>(ValueAt(pos));
+}
+
+std::shared_ptr<Bytes> GenericArray::GetBinary(int32_t pos) const {
+    const VariantType& value = ValueAt(pos);
+    if (const auto* bytes = DataDefine::GetVariantPtr<std::shared_ptr<Bytes>>(value)) {
+        return *bytes;
+    }
+    return std::shared_ptr<Bytes>(Bytes::AllocateBytes(
+        std::string(DataDefine::GetStringView(value)), GetDefaultPool().get()));
+}
+
+std::shared_ptr<InternalArray> GenericArray::GetArray(int32_t pos) const {
+    return DataDefine::GetVariantValue<std::shared_ptr<InternalArray>>(ValueAt(pos));
+}
+
+std::shared_ptr<InternalMap> GenericArray::GetMap(int32_t pos) const {
+    return DataDefine::GetVariantValue<std::shared_ptr<InternalMap>>(ValueAt(pos));
+}
+
+std::shared_ptr<InternalRow> GenericArray::GetRow(int32_t pos, int32_t num_fields) const {
+    return DataDefine::GetVariantValue<std::shared_ptr<InternalRow>>(ValueAt(pos));
+}
+
+template <typename T>
+Result<std::vector<T>> GenericArray::ToPrimitiveArray() const {
+    std::vector<T> result;
+    result.reserve(values_.size());
+    for (const VariantType& value : values_) {
+        if (DataDefine::IsVariantNull(value)) {
+            return Status::Invalid("Primitive array must not contain a null value.");
+        }
+        result.push_back(DataDefine::GetVariantValue<T>(value));
+    }
+    return result;
+}
+
+Result<std::vector<char>> GenericArray::ToBooleanArray() const {
+    std::vector<char> result;
+    result.reserve(values_.size());
+    for (const VariantType& value : values_) {
+        if (DataDefine::IsVariantNull(value)) {
+            return Status::Invalid("Primitive array must not contain a null value.");
+        }
+        result.push_back(DataDefine::GetVariantValue<bool>(value));
+    }
+    return result;
+}
+
+Result<std::vector<char>> GenericArray::ToByteArray() const {
+    return ToPrimitiveArray<char>();
+}
+
+Result<std::vector<int16_t>> GenericArray::ToShortArray() const {
+    return ToPrimitiveArray<int16_t>();
+}
+
+Result<std::vector<int32_t>> GenericArray::ToIntArray() const {
+    return ToPrimitiveArray<int32_t>();
+}
+
+Result<std::vector<int64_t>> GenericArray::ToLongArray() const {
+    return ToPrimitiveArray<int64_t>();
+}
+
+Result<std::vector<float>> GenericArray::ToFloatArray() const {
+    return ToPrimitiveArray<float>();
+}
+
+Result<std::vector<double>> GenericArray::ToDoubleArray() const {
+    return ToPrimitiveArray<double>();
+}
+
+}  // namespace paimon