branch-4.0: [opt](varbinary) support multi_distinct_count about varbinary and add conf mapping iceberg uuid #59406 (#59483)

Cherry-picked from #59406

Co-authored-by: zhangstar333 <zhangsida@selectdb.com>

Author: github-actions[bot]

be/src/vec/aggregate_functions/aggregate_function_uniq.cpp

@@ -40,9 +40,9 @@ AggregateFunctionPtr create_aggregate_function_uniq(const std::string& name,
     return creator_with_type_list<
             TYPE_BOOLEAN, TYPE_TINYINT, TYPE_SMALLINT, TYPE_INT, TYPE_BIGINT, TYPE_LARGEINT,
             TYPE_DECIMAL32, TYPE_DECIMAL64, TYPE_DECIMAL128I, TYPE_DECIMAL256, TYPE_VARCHAR,
-            TYPE_ARRAY, TYPE_FLOAT, TYPE_DOUBLE, TYPE_DATEV2, TYPE_DATETIMEV2,
-            TYPE_TIMESTAMPTZ>::create<AggregateFunctionUniq, Data>(argument_types,
-                                                                   result_is_nullable, attr);
+            TYPE_ARRAY, TYPE_FLOAT, TYPE_DOUBLE, TYPE_DATEV2, TYPE_DATETIMEV2, TYPE_TIMESTAMPTZ,
+            TYPE_VARBINARY>::create<AggregateFunctionUniq, Data>(argument_types, result_is_nullable,
+                                                                 attr);
 }
 
 void register_aggregate_function_uniq(AggregateFunctionSimpleFactory& factory) {

be/src/vec/aggregate_functions/aggregate_function_uniq.h

@@ -28,6 +28,7 @@
 #include <vector>
 
 #include "common/compiler_util.h" // IWYU pragma: keep
+#include "runtime/primitive_type.h"
 #include "vec/aggregate_functions/aggregate_function.h"
 #include "vec/columns/column.h"
 #include "vec/columns/column_vector.h"
@@ -56,7 +57,7 @@ class ColumnDecimal;
 
 template <PrimitiveType T>
 struct AggregateFunctionUniqExactData {
-    static constexpr bool is_string_key = is_string_type(T);
+    static constexpr bool is_string_key = is_string_type(T) || is_varbinary(T);
     using Key = std::conditional_t<
             is_string_key, UInt128,
             std::conditional_t<T == TYPE_ARRAY, UInt64,
@@ -91,7 +92,7 @@ namespace detail {
 template <PrimitiveType T, typename Data>
 struct OneAdder {
     static void ALWAYS_INLINE add(Data& data, const IColumn& column, size_t row_num) {
-        if constexpr (is_string_type(T)) {
+        if constexpr (is_string_type(T) || is_varbinary(T)) {
             StringRef value = column.get_data_at(row_num);
             data.set.insert(Data::get_key(value));
         } else if constexpr (T == TYPE_ARRAY) {
@@ -119,7 +120,7 @@ class AggregateFunctionUniq final
           NotNullableAggregateFunction {
 public:
     using KeyType =
-            std::conditional_t<is_string_type(T), UInt128,
+            std::conditional_t<is_string_type(T) || is_varbinary(T), UInt128,
                                std::conditional_t<T == TYPE_ARRAY, UInt64,
                                                   typename PrimitiveTypeTraits<T>::ColumnItemType>>;
     AggregateFunctionUniq(const DataTypes& argument_types_)
@@ -138,7 +139,7 @@ class AggregateFunctionUniq final
 
     static ALWAYS_INLINE const KeyType* get_keys(std::vector<KeyType>& keys_container,
                                                  const IColumn& column, size_t batch_size) {
-        if constexpr (is_string_type(T)) {
+        if constexpr (is_string_type(T) || is_varbinary(T)) {
             keys_container.resize(batch_size);
             for (size_t i = 0; i != batch_size; ++i) {
                 StringRef value = column.get_data_at(i);

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java

@@ -572,7 +572,7 @@ private static Type icebergPrimitiveTypeToDorisType(org.apache.iceberg.types.Typ
             case STRING:
                 return Type.STRING;
             case UUID:
-                return ScalarType.createVarbinaryType(16);
+                return enableMappingVarbinary ? ScalarType.createVarbinaryType(16) : Type.STRING;
             case BINARY:
                 return enableMappingVarbinary ? ScalarType.createVarbinaryType(VarBinaryType.MAX_VARBINARY_LENGTH)
                         : Type.STRING;

regression-test/data/external_table_p0/iceberg/test_iceberg_varbinary.out

@@ -149,4 +149,12 @@ suite("test_iceberg_varbinary", "p0,external,doris,external_docker,external_dock
     qt_select19 """
         select * from test_ice_uuid_parquet_write_with_mapping order by id;
     """
+
+    qt_select21 """
+        select multi_distinct_count(col2),multi_distinct_count(col1) from test_ice_uuid_orc;
+    """
+
+    qt_select22 """
+        select multi_distinct_count(col2),multi_distinct_count(col1) from test_ice_uuid_parquet;
+    """
 }