Merge pull request #5267 from 4ertus2/row_ref_list

RowRefList improvements

dbms/src/Interpreters/Join.cpp

@@ -397,17 +397,8 @@ namespace
                 new (&emplace_result.getMapped()) typename Map::mapped_type(stored_block, i);
             else
             {
-                /** The first element of the list is stored in the value of the hash table, the rest in the pool.
-                 * We will insert each time the element into the second place.
-                 * That is, the former second element, if it was, will be the third, and so on.
-                 */
-                auto elem = pool.alloc<typename Map::mapped_type>();
-                auto & mapped = emplace_result.getMapped();
-
-                elem->next = mapped.next;
-                mapped.next = elem;
-                elem->block = stored_block;
-                elem->row_num = i;
+                /// The first element of the list is stored in the value of the hash table, the rest in the pool.
+                emplace_result.getMapped().insert({stored_block, i}, pool);
             }
         }
     };
@@ -659,9 +650,9 @@ void addFoundRow(const typename Map::mapped_type & mapped, AddedColumns & added,
 
     if constexpr (STRICTNESS == ASTTableJoin::Strictness::All)
     {
-        for (auto current = &static_cast<const typename Map::mapped_type::Base &>(mapped); current != nullptr; current = current->next)
+        for (auto it = mapped.begin(); it.ok(); ++it)
         {
-            added.appendFromBlock(*current->block, current->row_num);
+            added.appendFromBlock(*it->block, it->row_num);
             ++current_offset;
         }
     }
@@ -1153,10 +1144,10 @@ struct AdderNonJoined<ASTTableJoin::Strictness::All, Mapped>
 {
     static void add(const Mapped & mapped, size_t & rows_added, MutableColumns & columns_right)
     {
-        for (auto current = &static_cast<const typename Mapped::Base &>(mapped); current != nullptr; current = current->next)
+        for (auto it = mapped.begin(); it.ok(); ++it)
         {
             for (size_t j = 0; j < columns_right.size(); ++j)
-                columns_right[j]->insertFrom(*current->block->getByPosition(j).column.get(), current->row_num);
+                columns_right[j]->insertFrom(*it->block->getByPosition(j).column.get(), it->row_num);
 
             ++rows_added;
         }

dbms/src/Interpreters/RowRefs.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <Common/Arena.h>
 #include <Common/RadixSort.h>
 #include <Columns/IColumn.h>
 
@@ -27,10 +28,99 @@ struct RowRef
 /// Single linked list of references to rows. Used for ALL JOINs (non-unique JOINs)
 struct RowRefList : RowRef
 {
-    RowRefList * next = nullptr;
+    /// Portion of RowRefs, 16 * (MAX_SIZE + 1) bytes sized.
+    struct Batch
+    {
+        static constexpr size_t MAX_SIZE = 7; /// Adequate values are 3, 7, 15, 31.
+
+        size_t size = 0;
+        Batch * next;
+        RowRef row_refs[MAX_SIZE];
+
+        Batch(Batch * parent)
+            : next(parent)
+        {}
+
+        bool full() const { return size == MAX_SIZE; }
+
+        Batch * insert(RowRef && row_ref, Arena & pool)
+        {
+            if (full())
+            {
+                auto batch = pool.alloc<Batch>();
+                *batch = Batch(this);
+                batch->insert(std::move(row_ref), pool);
+                return batch;
+            }
+
+            row_refs[size++] = std::move(row_ref);
+            return this;
+        }
+    };
+
+    class ForwardIterator
+    {
+    public:
+        ForwardIterator(const RowRefList * begin)
+            : root(begin)
+            , first(true)
+            , batch(root->next)
+            , position(0)
+        {}
+
+        const RowRef * operator -> () const
+        {
+            if (first)
+                return root;
+            return &batch->row_refs[position];
+        }
+
+        void operator ++ ()
+        {
+            if (first)
+            {
+                first = false;
+                return;
+            }
+
+            if (batch)
+            {
+                ++position;
+                if (position >= batch->size)
+                {
+                    batch = batch->next;
+                    position = 0;
+                }
+            }
+        }
+
+        bool ok() const { return first || (batch && position < batch->size); }
+
+    private:
+        const RowRefList * root;
+        bool first;
+        Batch * batch;
+        size_t position;
+    };
 
     RowRefList() {}
     RowRefList(const Block * block_, size_t row_num_) : RowRef(block_, row_num_) {}
+
+    ForwardIterator begin() const { return ForwardIterator(this); }
+
+    /// insert element after current one
+    void insert(RowRef && row_ref, Arena & pool)
+    {
+        if (!next)
+        {
+            next = pool.alloc<Batch>();
+            *next = Batch(nullptr);
+        }
+        next = next->insert(std::move(row_ref), pool);
+    }
+
+private:
+    Batch * next = nullptr;
 };
 
 /**

dbms/src/Storages/StorageJoin.cpp

@@ -338,14 +338,13 @@ private:
                 throw Exception("ASOF join storage is not implemented yet", ErrorCodes::NOT_IMPLEMENTED);
             }
             else
-                for (auto current = &static_cast<const typename Map::mapped_type::Base &>(it->getSecond()); current != nullptr;
-                     current = current->next)
+                for (auto ref_it = it->getSecond().begin(); ref_it.ok(); ++ref_it)
                 {
                     for (size_t j = 0; j < columns.size(); ++j)
                         if (j == key_pos)
                             columns[j]->insertData(rawData(it->getFirst()), rawSize(it->getFirst()));
                         else
-                            columns[j]->insertFrom(*current->block->getByPosition(column_indices[j]).column.get(), current->row_num);
+                            columns[j]->insertFrom(*ref_it->block->getByPosition(column_indices[j]).column.get(), ref_it->row_num);
                     ++rows_added;
                 }