diff --git a/dbms/src/DataStreams/HashJoinProbeBlockInputStream.cpp b/dbms/src/DataStreams/HashJoinProbeBlockInputStream.cpp index 283883fee13..2d40d1295c5 100644 --- 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 index 23c64dfd8ba..fb4213cff54 100644 --- 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 index 30d8e9e9bd7..86d61ec517f 100644 --- 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("a", {1, 1, 1, 1, 1, 1, 1, 1, 1, 1}), toVec("b", {1, 1, 3, 3, 1, 1, 3, 3, 1, 3})}); + context.addMockTable("split_test", "t1", {{"a", TiDB::TP::TypeLong}}, {toVec("a", {1, 1, 1, 1, 1, 1, 1, 1, 1, 1})}); context.addMockTable("split_test", "t2", {{"a", TiDB::TP::TypeLong}}, {toVec("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(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(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("a", {2, 2, 2, 2, 2})}); - context.addMockTable("split_test", "t4", {{"a", TiDB::TP::TypeLong}}, {toVec("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(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