mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-11-18 05:32:52 +00:00
132 lines
4.0 KiB
C++
132 lines
4.0 KiB
C++
#pragma once
|
|
|
|
#include <memory>
|
|
#include <shared_mutex>
|
|
|
|
#include <Common/LRUCache.h>
|
|
#include <Common/filesystemHelpers.h>
|
|
#include <Core/Block.h>
|
|
#include <Core/SortDescription.h>
|
|
#include <Interpreters/IJoin.h>
|
|
#include <DataStreams/SizeLimits.h>
|
|
|
|
namespace DB
|
|
{
|
|
|
|
class AnalyzedJoin;
|
|
class MergeJoinCursor;
|
|
struct MergeJoinEqualRange;
|
|
|
|
struct MiniLSM
|
|
{
|
|
using SortedFiles = std::vector<std::unique_ptr<TemporaryFile>>;
|
|
|
|
const String & path;
|
|
const Block & sample_block;
|
|
const SortDescription & sort_description;
|
|
const size_t rows_in_block;
|
|
const size_t max_size;
|
|
std::vector<SortedFiles> sorted_files;
|
|
|
|
MiniLSM(const String & path_, const Block & sample_block_, const SortDescription & description,
|
|
size_t rows_in_block_, size_t max_size_ = 16)
|
|
: path(path_)
|
|
, sample_block(sample_block_)
|
|
, sort_description(description)
|
|
, rows_in_block(rows_in_block_)
|
|
, max_size(max_size_)
|
|
{}
|
|
|
|
void insert(const BlocksList & blocks);
|
|
void merge(std::function<void(const Block &)> callback = [](const Block &){});
|
|
};
|
|
|
|
|
|
class MergeJoin : public IJoin
|
|
{
|
|
public:
|
|
MergeJoin(std::shared_ptr<AnalyzedJoin> table_join_, const Block & right_sample_block);
|
|
|
|
bool addJoinedBlock(const Block & block) override;
|
|
void joinBlock(Block &) override;
|
|
void joinTotals(Block &) const override;
|
|
void setTotals(const Block &) override;
|
|
bool hasTotals() const override { return totals; }
|
|
size_t getTotalRowCount() const override { return right_blocks_row_count; }
|
|
|
|
private:
|
|
/// There're two size limits for right-hand table: max_rows_in_join, max_bytes_in_join.
|
|
/// max_bytes is prefered. If it isn't set we aproximate it as (max_rows * bytes/row).
|
|
struct BlockByteWeight
|
|
{
|
|
size_t operator()(const Block & block) const { return block.bytes(); }
|
|
};
|
|
|
|
using Cache = LRUCache<size_t, Block, std::hash<size_t>, BlockByteWeight>;
|
|
|
|
mutable std::shared_mutex rwlock;
|
|
std::shared_ptr<AnalyzedJoin> table_join;
|
|
SizeLimits size_limits;
|
|
SortDescription left_sort_description;
|
|
SortDescription right_sort_description;
|
|
SortDescription left_merge_description;
|
|
SortDescription right_merge_description;
|
|
Block right_sample_block;
|
|
Block right_table_keys;
|
|
Block right_columns_to_add;
|
|
BlocksList right_blocks;
|
|
Blocks min_max_right_blocks;
|
|
std::unique_ptr<Cache> cached_right_blocks;
|
|
std::vector<std::shared_ptr<Block>> loaded_right_blocks;
|
|
std::unique_ptr<MiniLSM> lsm;
|
|
MiniLSM::SortedFiles flushed_right_blocks;
|
|
Block totals;
|
|
size_t right_blocks_row_count = 0;
|
|
size_t right_blocks_bytes = 0;
|
|
bool is_in_memory = true;
|
|
const bool nullable_right_side;
|
|
const bool is_all;
|
|
const bool is_inner;
|
|
const bool is_left;
|
|
const bool skip_not_intersected;
|
|
const size_t max_rows_in_right_block;
|
|
|
|
void changeLeftColumns(Block & block, MutableColumns && columns);
|
|
void addRightColumns(Block & block, MutableColumns && columns);
|
|
|
|
void mergeRightBlocks();
|
|
|
|
template <bool in_memory>
|
|
size_t rightBlocksCount();
|
|
template <bool in_memory>
|
|
void joinSortedBlock(Block & block);
|
|
template <bool in_memory>
|
|
std::shared_ptr<Block> loadRightBlock(size_t pos);
|
|
|
|
void leftJoin(MergeJoinCursor & left_cursor, const Block & left_block, const Block & right_block,
|
|
MutableColumns & left_columns, MutableColumns & right_columns, size_t & left_key_tail);
|
|
void innerJoin(MergeJoinCursor & left_cursor, const Block & left_block, const Block & right_block,
|
|
MutableColumns & left_columns, MutableColumns & right_columns, size_t & left_key_tail);
|
|
|
|
bool saveRightBlock(Block && block);
|
|
void flushRightBlocks();
|
|
|
|
void mergeInMemoryRightBlocks();
|
|
void mergeFlushedRightBlocks();
|
|
|
|
void clearRightBlocksList()
|
|
{
|
|
right_blocks.clear();
|
|
right_blocks_row_count = 0;
|
|
right_blocks_bytes = 0;
|
|
}
|
|
|
|
void countBlockSize(const Block & block)
|
|
{
|
|
right_blocks_row_count += block.rows();
|
|
right_blocks_bytes += block.bytes();
|
|
}
|
|
};
|
|
|
|
}
|