some refines (pingcap#7)

* save work Signed-off-by: xufei <[email protected]> * Update dbms/src/Interpreters/Join.cpp Co-authored-by: Meng Xin <[email protected]> --------- Signed-off-by: xufei <[email protected]> Co-authored-by: Meng Xin <[email protected]>
windtalker · Mar 6, 2023 · 0911265 · 0911265
1 parent 04744f6
commit 0911265
Show file tree

Hide file tree

Showing 4 changed files with 225 additions and 199 deletions.
diff --git a/dbms/src/Common/HashTable/HashTable.h b/dbms/src/Common/HashTable/HashTable.h
@@ -1497,15 +1497,18 @@ class ConcurrentHashTable : private boost::noncopyable
         return segments[segment_index]->getHashTable();
     }
 
-    void resetSegmentTable(size_t segment_index)
+    size_t resetSegmentTable(size_t segment_index)
     {
+        size_t ret = 0;
         std::unique_ptr<SegmentType> segment_ptr = nullptr;
         {
             /// release the lock before destruct related segment
             std::unique_lock lock(segments[segment_index]->getMutex());
+            ret = segments[segment_index]->getBufferSizeInBytes();
             segment_ptr.swap(segments[segment_index]);
             segments[segment_index].reset();
         }
+        return ret;
     }
 
     std::mutex & getSegmentMutex(size_t segment_index)

diff --git a/dbms/src/DataStreams/NonJoinedBlockInputStream.cpp b/dbms/src/DataStreams/NonJoinedBlockInputStream.cpp
@@ -143,13 +143,14 @@ Block NonJoinedBlockInputStream::readImpl()
         if (std::all_of(
                 std::begin(parent.partitions),
                 std::end(parent.partitions),
-                [](const Join::JoinPartition & partition) { return partition.build_partition.blocks.empty(); }))
+                [](const std::unique_ptr<Join::JoinPartition> & partition) { return partition->build_partition.blocks.empty(); }))
         {
             return Block();
         }
     }
 
 
+    /// todo read data based on JoinPartition
     if (add_not_mapped_rows)
     {
         setNextCurrentNotMappedRow();
@@ -258,7 +259,7 @@ size_t NonJoinedBlockInputStream::fillColumns(const Map & map,
     {
         current_segment = index;
 
-        while (parent.partitions[current_segment].spill)
+        while (parent.partitions[current_segment]->spill)
         {
             current_segment += step;
             if (current_segment >= map.getSegmentSize())
@@ -282,10 +283,10 @@ size_t NonJoinedBlockInputStream::fillColumns(const Map & map,
 
     for (; *it != end || current_segment + step < map.getSegmentSize();)
     {
-        if (*it == end || (parent.max_bytes_before_external_join && parent.partitions[current_segment].spill))
+        if (*it == end || (parent.max_bytes_before_external_join && parent.partitions[current_segment]->spill))
         {
             current_segment += step;
-            while (parent.partitions[current_segment].spill)
+            while (parent.partitions[current_segment]->spill)
             {
                 current_segment += step;
                 if (current_segment >= map.getSegmentSize())