refine DAGStorageInterpreter

dbms/src/Common/UniformRandomIntGenerator.h

@@ -0,0 +1,26 @@
+#pragma once
+
+#include <random>
+
+namespace DB
+{
+template <typename IntType>
+class UniformRandomIntGenerator
+{
+public:
+    /// [min, max]
+    UniformRandomIntGenerator(IntType min, IntType max)
+        : dis(std::uniform_int_distribution<IntType>(min, max))
+        , gen(std::default_random_engine(std::random_device{}()))
+    {}
+
+    IntType rand()
+    {
+        return dis(gen);
+    }
+
+private:
+    std::uniform_int_distribution<IntType> dis;
+    std::default_random_engine gen;
+};
+} // namespace DB

dbms/src/Flash/Coprocessor/DAGQueryBlockInterpreter.cpp

@@ -13,7 +13,6 @@
 #include <DataStreams/PartialSortingBlockInputStream.h>
 #include <DataStreams/SquashingBlockInputStream.h>
 #include <DataStreams/TiRemoteBlockInputStream.h>
-#include <DataStreams/UnionBlockInputStream.h>
 #include <DataTypes/DataTypeNullable.h>
 #include <DataTypes/DataTypesNumber.h>
 #include <DataTypes/getLeastSupertype.h>
@@ -244,16 +243,22 @@ void DAGQueryBlockInterpreter::handleTableScan(const tipb::TableScan & ts, DAGPi
     if (ts.next_read_engine() != tipb::EngineType::Local)
         throw TiFlashException("Unsupported remote query.", Errors::Coprocessor::BadRequest);
 
+    DAGStorageInterpreter storage_interpreter(context, query_block.source_name, ts, max_streams);
     // construct pushed down filter conditions.
     std::vector<const tipb::Expr *> conditions;
     if (query_block.selection)
     {
         for (const auto & condition : query_block.selection->selection().conditions())
             conditions.push_back(&condition);
-    }
 
-    DAGStorageInterpreter storage_interpreter(context, query_block, ts, conditions, max_streams);
-    storage_interpreter.execute(pipeline);
+        assert(!conditions.empty());
+        assert(!query_block.selection_name.empty());
+        storage_interpreter.execute(pipeline, query_block.selection_name, conditions);
+    }
+    else
+    {
+        storage_interpreter.execute(pipeline);
+    }
 
     analyzer = std::move(storage_interpreter.analyzer);