Remove usage about SquashingHashJoinBlockTransform in `HashJoinProb…

…eBlockInputStream` (#6714) ref #6713
pingcap · Feb 2, 2023 · 632fdaa · 632fdaa
1 parent 28d3c41
commit 632fdaa
Show file tree

Hide file tree

Showing 3 changed files with 2 additions and 60 deletions.
diff --git a/dbms/src/DataStreams/HashJoinProbeBlockInputStream.cpp b/dbms/src/DataStreams/HashJoinProbeBlockInputStream.cpp
@@ -26,7 +26,6 @@ HashJoinProbeBlockInputStream::HashJoinProbeBlockInputStream(
     , join(join_)
     , probe_index(probe_index_)
     , probe_process_info(max_block_size)
-    , squashing_transform(max_block_size)
 {
     children.push_back(input);
 
@@ -85,18 +84,7 @@ Block HashJoinProbeBlockInputStream::readImpl()
 {
     try
     {
-        // if join finished, return {} directly.
-        if (squashing_transform.isJoinFinished())
-        {
-            return Block{};
-        }
-
-        while (squashing_transform.needAppendBlock())
-        {
-            Block result_block = getOutputBlock();
-            squashing_transform.appendBlock(result_block);
-        }
-        auto ret = squashing_transform.getFinalOutputBlock();
+        Block ret = getOutputBlock();
         return ret;
     }
     catch (...)

diff --git a/dbms/src/DataStreams/HashJoinProbeBlockInputStream.h b/dbms/src/DataStreams/HashJoinProbeBlockInputStream.h
@@ -66,7 +66,6 @@ class HashJoinProbeBlockInputStream : public IProfilingBlockInputStream
     size_t probe_index;
     ProbeProcessInfo probe_process_info;
     BlockInputStreamPtr non_joined_stream;
-    SquashingHashJoinBlockTransform squashing_transform;
     ProbeStatus status{ProbeStatus::PROBE};
     size_t joined_rows = 0;
     size_t non_joined_rows = 0;

diff --git a/dbms/src/Flash/tests/gtest_join_executor.cpp b/dbms/src/Flash/tests/gtest_join_executor.cpp
@@ -706,7 +706,7 @@ CATCH
 TEST_F(JoinExecutorTestRunner, SplitJoinResult)
 try
 {
-    context.addMockTable("split_test", "t1", {{"a", TiDB::TP::TypeLong}, {"b", TiDB::TP::TypeLong}}, {toVec<Int32>("a", {1, 1, 1, 1, 1, 1, 1, 1, 1, 1}), toVec<Int32>("b", {1, 1, 3, 3, 1, 1, 3, 3, 1, 3})});
+    context.addMockTable("split_test", "t1", {{"a", TiDB::TP::TypeLong}}, {toVec<Int32>("a", {1, 1, 1, 1, 1, 1, 1, 1, 1, 1})});
     context.addMockTable("split_test", "t2", {{"a", TiDB::TP::TypeLong}}, {toVec<Int32>("a", {1, 1, 1, 1, 1})});
 
     auto request = context
@@ -726,51 +726,6 @@ try
             ASSERT_EQ(expect[i][j], blocks[j].rows());
         }
     }
-
-    // with other condition
-    const auto cond = gt(col("b"), lit(Field(static_cast<Int64>(2))));
-    request = context
-                  .scan("split_test", "t1")
-                  .join(context.scan("split_test", "t2"), tipb::JoinType::TypeInnerJoin, {col("a")}, {}, {}, {cond}, {})
-
-                  .build(context);
-    expect = {{5, 5, 5, 5, 5}, {5, 5, 5, 5, 5}, {5, 5, 5, 5, 5}, {25}, {25}, {25}, {25}, {25}};
-    for (size_t i = 0; i < block_sizes.size(); ++i)
-    {
-        context.context.setSetting("max_block_size", Field(static_cast<UInt64>(block_sizes[i])));
-        auto blocks = getExecuteStreamsReturnBlocks(request);
-        ASSERT_EQ(expect[i].size(), blocks.size());
-        for (size_t j = 0; j < blocks.size(); ++j)
-        {
-            ASSERT_EQ(expect[i][j], blocks[j].rows());
-        }
-    }
-    // test non joined data
-    context.addMockTable("split_test", "t3", {{"a", TiDB::TP::TypeLong}}, {toVec<Int32>("a", {2, 2, 2, 2, 2})});
-    context.addMockTable("split_test", "t4", {{"a", TiDB::TP::TypeLong}}, {toVec<Int32>("a", {1, 3, 1, 3, 1, 3, 1, 3, 1, 3, 1, 3, 1, 3, 1, 3, 1, 3, 1, 3})});
-    request = context
-                  .scan("split_test", "t3")
-                  .join(context.scan("split_test", "t4"), tipb::JoinType::TypeRightOuterJoin, {col("a")})
-                  .build(context);
-
-    expect = {{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1},
-              {2, 2, 2, 2, 2, 2, 2, 2, 2, 2},
-              {7, 7, 6},
-              {20},
-              {20},
-              {20},
-              {20},
-              {20}};
-    for (size_t i = 0; i < block_sizes.size(); ++i)
-    {
-        context.context.setSetting("max_block_size", Field(static_cast<UInt64>(block_sizes[i])));
-        auto blocks = getExecuteStreamsReturnBlocks(request);
-        ASSERT_EQ(expect[i].size(), blocks.size());
-        for (size_t j = 0; j < blocks.size(); ++j)
-        {
-            ASSERT_EQ(expect[i][j], blocks[j].rows());
-        }
-    }
 }
 CATCH