[Enhancement](auto-partition) Re-add deduplication to auto partition …

…rpc (#40580) Issue Number: close #xxx removed in #27817. we need it so re add it.
apache · Sep 11, 2024 · 561ca47 · 561ca47
1 parent b0ab02b
commit 561ca47
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 7 deletions.
diff --git a/be/src/vec/sink/vrow_distribution.cpp b/be/src/vec/sink/vrow_distribution.cpp
@@ -66,8 +66,10 @@ Status VRowDistribution::_save_missing_values(
             }
             cur_row_values.push_back(node);
         }
-        //For duplicate cur_values, they will be filtered in FE
-        _partitions_need_create.emplace_back(cur_row_values);
+        if (!_deduper.contains(cur_row_values)) {
+            _deduper.insert(cur_row_values);
+            _partitions_need_create.emplace_back(cur_row_values);
+        }
     }
 
     // to avoid too large mem use

diff --git a/be/src/vec/sink/vrow_distribution.h b/be/src/vec/sink/vrow_distribution.h
@@ -24,7 +24,9 @@
 #include <gen_cpp/PaloInternalService_types.h>
 
 #include <cstdint>
+#include <functional>
 #include <string>
+#include <unordered_set>
 #include <vector>
 
 #include "common/status.h"
@@ -133,6 +135,10 @@ class VRowDistribution {
     Status automatic_create_partition();
     void clear_batching_stats();
 
+    // for auto partition
+    std::unique_ptr<MutableBlock> _batching_block;
+    bool _deal_batched = false; // If true, send batched block before any block's append.
+
 private:
     std::pair<vectorized::VExprContextSPtrs, vectorized::VExprSPtrs> _get_partition_function();
 
@@ -170,17 +176,29 @@ class VRowDistribution {
                                     int64_t rows);
     void _reset_find_tablets(int64_t rows);
 
+    struct NullableStringListHash {
+        std::size_t _hash(const TNullableStringLiteral& arg) const {
+            if (arg.is_null) {
+                return 0;
+            }
+            return std::hash<std::string>()(arg.value);
+        }
+        std::size_t operator()(const std::vector<TNullableStringLiteral>& arg) const {
+            std::size_t result = 0;
+            for (const auto& v : arg) {
+                result = (result << 1) ^ _hash(v);
+            }
+            return result;
+        }
+    };
+
     RuntimeState* _state = nullptr;
     int _batch_size = 0;
 
     // for auto partitions
     std::vector<std::vector<TNullableStringLiteral>> _partitions_need_create;
-
-public:
-    std::unique_ptr<MutableBlock> _batching_block;
-    bool _deal_batched = false; // If true, send batched block before any block's append.
-private:
     size_t _batching_rows = 0, _batching_bytes = 0;
+    std::unordered_set<std::vector<TNullableStringLiteral>, NullableStringListHash> _deduper;
 
     OlapTableBlockConvertor* _block_convertor = nullptr;
     OlapTabletFinder* _tablet_finder = nullptr;