Refactor Regions Storage

dbms/src/DataStreams/RangesFilterBlockInputStream.cpp

@@ -0,0 +1,83 @@
+#include <Columns/ColumnsNumber.h>
+#include <DataStreams/RangesFilterBlockInputStream.h>
+#include <DataStreams/dedupUtils.h>
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+extern const int LOGICAL_ERROR;
+}
+
+Block RangesFilterBlockInputStream::readImpl()
+{
+    while (true)
+    {
+        Block block = input->read();
+        if (!block)
+            return block;
+
+        if (!block.has(handle_col_name))
+            throw Exception("RangesFilterBlockInputStream: block without _tidb_rowid.", ErrorCodes::LOGICAL_ERROR);
+
+        const ColumnWithTypeAndName & handle_column = block.getByName(handle_col_name);
+        const ColumnInt64 * column = typeid_cast<const ColumnInt64 *>(handle_column.column.get());
+        if (!column)
+        {
+            throw Exception("RangesFilterBlockInputStream: _tidb_rowid column should be type ColumnInt64.", ErrorCodes::LOGICAL_ERROR);
+        }
+
+        size_t rows = block.rows();
+
+        auto handle_begin = column->getElement(0);
+        auto handle_end = column->getElement(rows - 1);
+
+        if (handle_begin >= ranges.second || ranges.first > handle_end)
+            continue;
+
+        if (handle_begin >= ranges.first)
+        {
+            if (handle_end < ranges.second)
+            {
+                return block;
+            }
+            else
+            {
+                size_t pos
+                    = std::lower_bound(column->getData().cbegin(), column->getData().cend(), ranges.second) - column->getData().cbegin();
+                size_t pop_num = rows - pos;
+                for (size_t i = 0; i < block.columns(); i++)
+                {
+                    ColumnWithTypeAndName & ori_column = block.getByPosition(i);
+                    MutableColumnPtr mutable_holder = (*std::move(ori_column.column)).mutate();
+                    mutable_holder->popBack(pop_num);
+                    ori_column.column = std::move(mutable_holder);
+                }
+            }
+        }
+        else
+        {
+            size_t pos_begin
+                = std::lower_bound(column->getData().cbegin(), column->getData().cend(), ranges.first) - column->getData().cbegin();
+            size_t pos_end = rows;
+            if (handle_end >= ranges.second)
+                pos_end = std::lower_bound(column->getData().cbegin(), column->getData().cend(), ranges.second) - column->getData().cbegin();
+
+            size_t len = pos_end - pos_begin;
+            if (!len)
+                continue;
+            for (size_t i = 0; i < block.columns(); i++)
+            {
+                ColumnWithTypeAndName & ori_column = block.getByPosition(i);
+                auto new_column = ori_column.column->cloneEmpty();
+                new_column->insertRangeFrom(*ori_column.column, pos_begin, len);
+                ori_column.column = std::move(new_column);
+            }
+        }
+
+        return block;
+    }
+}
+
+} // namespace DB

dbms/src/DataStreams/RangesFilterBlockInputStream.h

@@ -0,0 +1,35 @@
+#pragma once
+
+#include <DataStreams/IProfilingBlockInputStream.h>
+#include <Storages/Transaction/Region.h>
+#include <common/logger_useful.h>
+
+namespace DB
+{
+
+class RangesFilterBlockInputStream : public IProfilingBlockInputStream
+{
+public:
+    RangesFilterBlockInputStream(const BlockInputStreamPtr & input_, const HandleRange & ranges_, const String & handle_col_name_) : input(input_), ranges(ranges_), handle_col_name(handle_col_name_) {}
+
+protected:
+    Block getHeader() const override { return input->getHeader(); }
+
+    bool isGroupedOutput() const override { return input->isGroupedOutput(); }
+
+    bool isSortedOutput() const override { return input->isSortedOutput(); }
+
+    const SortDescription & getSortDescription() const override { return input->getSortDescription(); }
+
+    String getName() const override { return "RangesFilter"; }
+
+    Block readImpl() override;
+
+private:
+    BlockInputStreamPtr input;
+    const HandleRange ranges;
+    const String handle_col_name;
+    Logger * log = &Logger::get("RangesFilterBlockInputStream");
+};
+
+} // namespace DB

dbms/src/Debug/DBGInvoker.cpp

@@ -51,10 +51,7 @@ DBGInvoker::DBGInvoker()
     regFunc("region_snapshot", dbgFuncRegionSnapshot);
     regFunc("rm_region_data", dbgFuncRegionRmData);
 
-    regFunc("dump_partition", dbgFuncRegionPartition);
-    regFunc("check_partition", dbgFuncCheckPartitionRegionRows);
-    regFunc("scan_partition", dbgFuncScanPartitionExtraRows);
-    regFunc("check_region_correct", dbgFuncCheckRegionCorrect);
+    regFunc("dump_region", dbgFuncDumpRegion);
 }
 
 void replaceSubstr(std::string & str, const std::string & target, const std::string & replacement)

dbms/src/Debug/dbgFuncMockTiDBData.cpp

@@ -26,7 +26,7 @@ void dbgFuncSetFlushThreshold(Context & context, const ASTs & args, DBGInvoker::
     auto seconds = safeGet<UInt64>(typeid_cast<const ASTLiteral &>(*args[1]).value);
 
     TMTContext & tmt = context.getTMTContext();
-    tmt.region_partition.setFlushThresholds({{bytes, Seconds(seconds)}});
+    tmt.region_table.setFlushThresholds({{bytes, Seconds(seconds)}});
 
     std::stringstream ss;
     ss << "set flush threshold to (" << bytes << " bytes, " << seconds << " seconds)";

dbms/src/Debug/dbgFuncMockTiDBTable.cpp

@@ -93,7 +93,7 @@ void dbgFuncDropTiDBTable(Context & context, const ASTs & args, DBGInvoker::Prin
     }
 
     TMTContext & tmt = context.getTMTContext();
-    tmt.region_partition.dropRegionsInTable(table_id);
+    tmt.region_table.dropRegionsInTable(table_id);
 
     MockTiDB::instance().dropTable(database_name, table_name);