2017-04-01 09:19:00 +00:00
|
|
|
#include <Storages/StorageJoin.h>
|
2017-12-30 00:36:06 +00:00
|
|
|
#include <Storages/StorageFactory.h>
|
2021-04-18 09:38:50 +00:00
|
|
|
#include <Storages/StorageSet.h>
|
2021-09-29 17:30:07 +00:00
|
|
|
#include <Storages/TableLockHolder.h>
|
2020-04-07 09:48:47 +00:00
|
|
|
#include <Interpreters/HashJoin.h>
|
2020-02-10 15:50:12 +00:00
|
|
|
#include <Interpreters/Context.h>
|
2018-11-30 14:49:35 +00:00
|
|
|
#include <Parsers/ASTCreateQuery.h>
|
2021-11-26 15:49:40 +00:00
|
|
|
#include <Parsers/ASTIdentifier_fwd.h>
|
2018-11-30 14:49:35 +00:00
|
|
|
#include <Core/ColumnNumbers.h>
|
|
|
|
#include <DataTypes/NestedUtils.h>
|
2019-07-02 14:38:31 +00:00
|
|
|
#include <Interpreters/joinDispatch.h>
|
2021-04-18 09:38:50 +00:00
|
|
|
#include <Interpreters/MutationsInterpreter.h>
|
2020-04-07 09:48:47 +00:00
|
|
|
#include <Interpreters/TableJoin.h>
|
2020-11-25 12:41:12 +00:00
|
|
|
#include <Interpreters/castColumn.h>
|
2019-12-30 21:30:25 +00:00
|
|
|
#include <Common/quoteString.h>
|
2021-04-18 09:38:50 +00:00
|
|
|
#include <Common/Exception.h>
|
2022-08-04 15:20:19 +00:00
|
|
|
#include <Interpreters/JoinUtils.h>
|
2017-12-30 00:36:06 +00:00
|
|
|
|
2021-04-18 09:38:50 +00:00
|
|
|
#include <Compression/CompressedWriteBuffer.h>
|
2022-05-20 19:49:31 +00:00
|
|
|
#include <Processors/ISource.h>
|
2021-10-16 14:03:50 +00:00
|
|
|
#include <QueryPipeline/Pipe.h>
|
2022-05-24 20:06:08 +00:00
|
|
|
#include <QueryPipeline/QueryPipelineBuilder.h>
|
2021-10-06 17:59:27 +00:00
|
|
|
#include <Processors/Executors/PullingPipelineExecutor.h>
|
2023-03-06 10:54:43 +00:00
|
|
|
#include <Poco/String.h>
|
|
|
|
#include <filesystem>
|
2015-01-27 21:24:24 +00:00
|
|
|
|
2023-03-06 10:54:43 +00:00
|
|
|
namespace fs = std::filesystem;
|
2015-01-27 21:24:24 +00:00
|
|
|
|
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
|
2016-01-11 21:46:36 +00:00
|
|
|
namespace ErrorCodes
|
|
|
|
{
|
2023-03-06 10:54:43 +00:00
|
|
|
extern const int BAD_ARGUMENTS;
|
|
|
|
extern const int DEADLOCK_AVOIDED;
|
|
|
|
extern const int INCOMPATIBLE_TYPE_OF_JOIN;
|
2020-02-25 18:02:41 +00:00
|
|
|
extern const int LOGICAL_ERROR;
|
2017-04-01 07:20:54 +00:00
|
|
|
extern const int NO_SUCH_COLUMN_IN_TABLE;
|
2023-03-06 10:54:43 +00:00
|
|
|
extern const int NOT_IMPLEMENTED;
|
2017-12-30 00:36:06 +00:00
|
|
|
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
|
2023-03-06 10:54:43 +00:00
|
|
|
extern const int UNSUPPORTED_JOIN_KEYS;
|
2016-01-11 21:46:36 +00:00
|
|
|
}
|
|
|
|
|
2015-01-27 21:24:24 +00:00
|
|
|
StorageJoin::StorageJoin(
|
2020-12-15 16:45:13 +00:00
|
|
|
DiskPtr disk_,
|
2019-10-25 19:07:47 +00:00
|
|
|
const String & relative_path_,
|
2019-12-04 16:06:55 +00:00
|
|
|
const StorageID & table_id_,
|
2017-04-01 07:20:54 +00:00
|
|
|
const Names & key_names_,
|
2018-11-30 14:49:35 +00:00
|
|
|
bool use_nulls_,
|
|
|
|
SizeLimits limits_,
|
2022-07-29 16:30:50 +00:00
|
|
|
JoinKind kind_,
|
|
|
|
JoinStrictness strictness_,
|
2018-12-30 15:54:45 +00:00
|
|
|
const ColumnsDescription & columns_,
|
2019-08-24 21:20:20 +00:00
|
|
|
const ConstraintsDescription & constraints_,
|
2021-04-23 12:18:23 +00:00
|
|
|
const String & comment,
|
2020-03-29 10:07:51 +00:00
|
|
|
bool overwrite_,
|
2020-09-18 12:58:27 +00:00
|
|
|
bool persistent_)
|
2021-04-23 12:18:23 +00:00
|
|
|
: StorageSetOrJoinBase{disk_, relative_path_, table_id_, columns_, constraints_, comment, persistent_}
|
2018-11-30 14:49:35 +00:00
|
|
|
, key_names(key_names_)
|
|
|
|
, use_nulls(use_nulls_)
|
|
|
|
, limits(limits_)
|
|
|
|
, kind(kind_)
|
|
|
|
, strictness(strictness_)
|
2020-03-29 10:07:51 +00:00
|
|
|
, overwrite(overwrite_)
|
2015-01-27 21:24:24 +00:00
|
|
|
{
|
2020-06-16 15:51:29 +00:00
|
|
|
auto metadata_snapshot = getInMemoryMetadataPtr();
|
2017-04-01 07:20:54 +00:00
|
|
|
for (const auto & key : key_names)
|
2020-06-16 15:51:29 +00:00
|
|
|
if (!metadata_snapshot->getColumns().hasPhysical(key))
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::NO_SUCH_COLUMN_IN_TABLE, "Key column ({}) does not exist in table declaration.", key);
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2020-04-07 09:48:47 +00:00
|
|
|
table_join = std::make_shared<TableJoin>(limits, use_nulls, kind, strictness, key_names);
|
2022-02-11 20:26:46 +00:00
|
|
|
join = std::make_shared<HashJoin>(table_join, getRightSampleBlock(), overwrite);
|
2017-04-01 07:20:54 +00:00
|
|
|
restore();
|
2015-01-27 21:24:24 +00:00
|
|
|
}
|
|
|
|
|
2021-09-30 08:47:15 +00:00
|
|
|
RWLockImpl::LockHolder StorageJoin::tryLockTimedWithContext(const RWLock & lock, RWLockImpl::Type type, ContextPtr context) const
|
2021-09-29 17:30:07 +00:00
|
|
|
{
|
2021-09-30 08:47:15 +00:00
|
|
|
const String query_id = context ? context->getInitialQueryId() : RWLockImpl::NO_QUERY;
|
2021-09-29 17:30:07 +00:00
|
|
|
const std::chrono::milliseconds acquire_timeout
|
2021-09-30 08:47:15 +00:00
|
|
|
= context ? context->getSettingsRef().lock_acquire_timeout : std::chrono::seconds(DBMS_DEFAULT_LOCK_ACQUIRE_TIMEOUT_SEC);
|
2021-09-29 17:30:07 +00:00
|
|
|
return tryLockTimed(lock, type, query_id, acquire_timeout);
|
|
|
|
}
|
|
|
|
|
2023-03-13 10:49:51 +00:00
|
|
|
RWLockImpl::LockHolder StorageJoin::tryLockForCurrentQueryTimedWithContext(const RWLock & lock, RWLockImpl::Type type, ContextPtr context)
|
2023-03-06 10:54:43 +00:00
|
|
|
{
|
|
|
|
const String query_id = context ? context->getInitialQueryId() : RWLockImpl::NO_QUERY;
|
|
|
|
const std::chrono::milliseconds acquire_timeout
|
|
|
|
= context ? context->getSettingsRef().lock_acquire_timeout : std::chrono::seconds(DBMS_DEFAULT_LOCK_ACQUIRE_TIMEOUT_SEC);
|
|
|
|
return lock->getLock(type, query_id, acquire_timeout, false);
|
|
|
|
}
|
|
|
|
|
2021-07-23 19:33:59 +00:00
|
|
|
SinkToStoragePtr StorageJoin::write(const ASTPtr & query, const StorageMetadataPtr & metadata_snapshot, ContextPtr context)
|
2021-05-31 23:22:05 +00:00
|
|
|
{
|
|
|
|
std::lock_guard mutate_lock(mutate_mutex);
|
|
|
|
return StorageSetOrJoinBase::write(query, metadata_snapshot, context);
|
|
|
|
}
|
2015-01-27 21:24:24 +00:00
|
|
|
|
2022-02-11 20:26:46 +00:00
|
|
|
void StorageJoin::truncate(const ASTPtr &, const StorageMetadataPtr &, ContextPtr context, TableExclusiveLockHolder &)
|
2018-06-09 16:09:37 +00:00
|
|
|
{
|
2021-05-31 23:22:05 +00:00
|
|
|
std::lock_guard mutate_lock(mutate_mutex);
|
2021-09-30 08:47:15 +00:00
|
|
|
TableLockHolder holder = tryLockTimedWithContext(rwlock, RWLockImpl::Write, context);
|
2021-05-31 23:22:05 +00:00
|
|
|
|
2023-01-30 16:00:28 +00:00
|
|
|
if (disk->exists(path))
|
|
|
|
disk->removeRecursive(path);
|
|
|
|
else
|
|
|
|
LOG_INFO(&Poco::Logger::get("StorageJoin"), "Path {} is already removed from disk {}", path, disk->getName());
|
|
|
|
|
2020-12-15 16:45:13 +00:00
|
|
|
disk->createDirectories(path);
|
2023-03-06 10:54:43 +00:00
|
|
|
disk->createDirectories(fs::path(path) / "tmp/");
|
2018-06-09 16:09:37 +00:00
|
|
|
|
|
|
|
increment = 0;
|
2022-02-11 20:26:46 +00:00
|
|
|
join = std::make_shared<HashJoin>(table_join, getRightSampleBlock(), overwrite);
|
2018-08-10 04:02:56 +00:00
|
|
|
}
|
2018-06-09 16:09:37 +00:00
|
|
|
|
2021-04-18 09:38:50 +00:00
|
|
|
void StorageJoin::checkMutationIsPossible(const MutationCommands & commands, const Settings & /* settings */) const
|
|
|
|
{
|
2021-05-31 23:22:05 +00:00
|
|
|
for (const auto & command : commands)
|
|
|
|
if (command.type != MutationCommand::DELETE)
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Table engine Join supports only DELETE mutations");
|
2021-04-18 09:38:50 +00:00
|
|
|
}
|
|
|
|
|
2023-01-30 17:38:28 +00:00
|
|
|
void StorageJoin::mutate(const MutationCommands & commands, ContextPtr context)
|
2021-04-18 09:38:50 +00:00
|
|
|
{
|
2021-06-28 17:02:22 +00:00
|
|
|
/// Firstly acquire lock for mutation, that locks changes of data.
|
|
|
|
/// We cannot acquire rwlock here, because read lock is needed
|
2021-05-31 23:22:05 +00:00
|
|
|
/// for execution of mutation interpreter.
|
|
|
|
std::lock_guard mutate_lock(mutate_mutex);
|
2021-04-18 09:38:50 +00:00
|
|
|
|
2021-05-31 23:22:05 +00:00
|
|
|
constexpr auto tmp_backup_file_name = "tmp/mut.bin";
|
|
|
|
auto metadata_snapshot = getInMemoryMetadataPtr();
|
2021-04-18 09:38:50 +00:00
|
|
|
|
2021-05-31 23:22:05 +00:00
|
|
|
auto backup_buf = disk->writeFile(path + tmp_backup_file_name);
|
2021-04-18 09:38:50 +00:00
|
|
|
auto compressed_backup_buf = CompressedWriteBuffer(*backup_buf);
|
2021-10-08 17:21:19 +00:00
|
|
|
auto backup_stream = NativeWriter(compressed_backup_buf, 0, metadata_snapshot->getSampleBlock());
|
2021-04-18 09:38:50 +00:00
|
|
|
|
2022-02-11 20:26:46 +00:00
|
|
|
auto new_data = std::make_shared<HashJoin>(table_join, getRightSampleBlock(), overwrite);
|
2021-05-31 23:22:05 +00:00
|
|
|
|
2022-05-09 19:13:02 +00:00
|
|
|
// New scope controls lifetime of pipeline.
|
2021-04-19 03:22:35 +00:00
|
|
|
{
|
2021-05-31 23:22:05 +00:00
|
|
|
auto storage_ptr = DatabaseCatalog::instance().getTable(getStorageID(), context);
|
2023-05-25 22:54:54 +00:00
|
|
|
MutationsInterpreter::Settings settings(true);
|
|
|
|
auto interpreter = std::make_unique<MutationsInterpreter>(storage_ptr, metadata_snapshot, commands, context, settings);
|
2022-05-24 20:06:08 +00:00
|
|
|
auto pipeline = QueryPipelineBuilder::getPipeline(interpreter->execute());
|
2021-10-06 17:59:27 +00:00
|
|
|
PullingPipelineExecutor executor(pipeline);
|
2021-05-31 23:22:05 +00:00
|
|
|
|
2021-10-06 17:59:27 +00:00
|
|
|
Block block;
|
|
|
|
while (executor.pull(block))
|
2021-05-31 23:22:05 +00:00
|
|
|
{
|
|
|
|
new_data->addJoinedBlock(block, true);
|
|
|
|
if (persistent)
|
|
|
|
backup_stream.write(block);
|
|
|
|
}
|
2021-04-18 09:38:50 +00:00
|
|
|
}
|
|
|
|
|
2021-06-28 17:02:22 +00:00
|
|
|
/// Now acquire exclusive lock and modify storage.
|
2021-09-29 17:30:07 +00:00
|
|
|
TableLockHolder holder = tryLockTimedWithContext(rwlock, RWLockImpl::Write, context);
|
2021-05-31 23:22:05 +00:00
|
|
|
|
2021-04-18 09:38:50 +00:00
|
|
|
join = std::move(new_data);
|
|
|
|
increment = 1;
|
|
|
|
|
2021-04-19 03:22:35 +00:00
|
|
|
if (persistent)
|
|
|
|
{
|
2021-04-18 09:38:50 +00:00
|
|
|
backup_stream.flush();
|
|
|
|
compressed_backup_buf.next();
|
|
|
|
backup_buf->next();
|
|
|
|
backup_buf->finalize();
|
|
|
|
|
|
|
|
std::vector<std::string> files;
|
|
|
|
disk->listFiles(path, files);
|
2021-04-19 03:22:35 +00:00
|
|
|
for (const auto & file_name: files)
|
|
|
|
{
|
|
|
|
if (file_name.ends_with(".bin"))
|
2021-04-18 09:38:50 +00:00
|
|
|
disk->removeFileIfExists(path + file_name);
|
|
|
|
}
|
|
|
|
|
2021-05-31 23:22:05 +00:00
|
|
|
disk->replaceFile(path + tmp_backup_file_name, path + std::to_string(increment) + ".bin");
|
2021-04-18 09:38:50 +00:00
|
|
|
}
|
|
|
|
}
|
2018-06-09 16:09:37 +00:00
|
|
|
|
2021-09-30 08:47:15 +00:00
|
|
|
HashJoinPtr StorageJoin::getJoinLocked(std::shared_ptr<TableJoin> analyzed_join, ContextPtr context) const
|
2015-01-28 02:37:05 +00:00
|
|
|
{
|
2020-06-16 15:51:29 +00:00
|
|
|
auto metadata_snapshot = getInMemoryMetadataPtr();
|
2020-02-02 16:12:48 +00:00
|
|
|
if (!analyzed_join->sameStrictnessAndKind(strictness, kind))
|
2022-12-08 10:15:31 +00:00
|
|
|
throw Exception(ErrorCodes::INCOMPATIBLE_TYPE_OF_JOIN, "Table '{}' has incompatible type of JOIN", getStorageID().getNameForLogs());
|
2019-12-19 15:50:28 +00:00
|
|
|
|
2019-12-30 20:08:03 +00:00
|
|
|
if ((analyzed_join->forceNullableRight() && !use_nulls) ||
|
|
|
|
(!analyzed_join->forceNullableRight() && isLeftOrFull(analyzed_join->kind()) && use_nulls))
|
2022-02-11 20:26:46 +00:00
|
|
|
throw Exception(
|
|
|
|
ErrorCodes::INCOMPATIBLE_TYPE_OF_JOIN,
|
|
|
|
"Table {} needs the same join_use_nulls setting as present in LEFT or FULL JOIN",
|
|
|
|
getStorageID().getNameForLogs());
|
2019-12-30 19:43:40 +00:00
|
|
|
|
2022-12-06 11:45:41 +00:00
|
|
|
if (analyzed_join->getClauses().size() != 1)
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::INCOMPATIBLE_TYPE_OF_JOIN, "JOIN keys should match to the Join engine keys [{}]",
|
|
|
|
fmt::join(getKeyNames(), ", "));
|
2022-12-06 11:45:41 +00:00
|
|
|
|
2022-12-07 16:27:01 +00:00
|
|
|
const auto & join_on = analyzed_join->getOnlyClause();
|
|
|
|
if (join_on.on_filter_condition_left || join_on.on_filter_condition_right)
|
|
|
|
throw Exception(ErrorCodes::INCOMPATIBLE_TYPE_OF_JOIN, "ON section of JOIN with filter conditions is not implemented");
|
|
|
|
|
|
|
|
const auto & key_names_right = join_on.key_names_right;
|
|
|
|
const auto & key_names_left = join_on.key_names_left;
|
2022-12-07 15:02:37 +00:00
|
|
|
if (key_names.size() != key_names_right.size() || key_names.size() != key_names_left.size())
|
|
|
|
throw Exception(ErrorCodes::INCOMPATIBLE_TYPE_OF_JOIN,
|
|
|
|
"Number of keys in JOIN ON section ({}) doesn't match number of keys in Join engine ({})",
|
|
|
|
key_names_right.size(), key_names.size());
|
|
|
|
|
|
|
|
/* Resort left keys according to right keys order in StorageJoin
|
2022-12-08 10:15:31 +00:00
|
|
|
* We can't change the order of keys in StorageJoin
|
|
|
|
* because the hash table was already built with tuples serialized in the order of key_names.
|
|
|
|
* If we try to use the same hash table with different order of keys,
|
|
|
|
* then calculated hashes and the result of the comparison will be wrong.
|
2022-12-07 15:02:37 +00:00
|
|
|
*
|
|
|
|
* Example:
|
|
|
|
* ```
|
|
|
|
* CREATE TABLE t_right (a UInt32, b UInt32) ENGINE = Join(ALL, INNER, a, b);
|
|
|
|
* SELECT * FROM t_left JOIN t_right ON t_left.y = t_right.b AND t_left.x = t_right.a;
|
|
|
|
* ```
|
|
|
|
* In that case right keys should still be (a, b), need to change the order of the left keys to (x, y).
|
|
|
|
*/
|
|
|
|
Names left_key_names_resorted;
|
2022-12-08 10:15:31 +00:00
|
|
|
for (const auto & key_name : key_names)
|
2022-12-07 15:02:37 +00:00
|
|
|
{
|
2023-04-25 11:28:54 +00:00
|
|
|
const auto & renamed_key = analyzed_join->renamedRightColumnNameWithAlias(key_name);
|
2022-12-07 15:02:37 +00:00
|
|
|
/// find position of renamed_key in key_names_right
|
|
|
|
auto it = std::find(key_names_right.begin(), key_names_right.end(), renamed_key);
|
|
|
|
if (it == key_names_right.end())
|
|
|
|
throw Exception(ErrorCodes::INCOMPATIBLE_TYPE_OF_JOIN,
|
2023-04-25 11:28:54 +00:00
|
|
|
"Key '{}' not found in JOIN ON section. Join engine key{} '{}' have to be used",
|
|
|
|
key_name, key_names.size() > 1 ? "s" : "", fmt::join(key_names, ", "));
|
2022-12-08 10:15:31 +00:00
|
|
|
const size_t key_position = std::distance(key_names_right.begin(), it);
|
2022-12-07 15:02:37 +00:00
|
|
|
left_key_names_resorted.push_back(key_names_left[key_position]);
|
|
|
|
}
|
2019-12-19 15:50:28 +00:00
|
|
|
|
2022-12-06 11:45:41 +00:00
|
|
|
/// Set qualified identifiers to original names (table.column -> column).
|
|
|
|
/// It's required because storage join stores non-qualified names.
|
|
|
|
/// Qualifies will be added by join implementation (TableJoin contains a rename mapping).
|
2019-12-19 20:37:10 +00:00
|
|
|
analyzed_join->setRightKeys(key_names);
|
2022-12-07 15:02:37 +00:00
|
|
|
analyzed_join->setLeftKeys(left_key_names_resorted);
|
2019-12-19 20:37:10 +00:00
|
|
|
|
2022-02-11 20:26:46 +00:00
|
|
|
HashJoinPtr join_clone = std::make_shared<HashJoin>(analyzed_join, getRightSampleBlock());
|
2021-09-29 17:30:07 +00:00
|
|
|
|
2021-09-30 08:47:15 +00:00
|
|
|
RWLockImpl::LockHolder holder = tryLockTimedWithContext(rwlock, RWLockImpl::Read, context);
|
2021-09-29 17:30:07 +00:00
|
|
|
join_clone->setLock(holder);
|
2021-02-24 06:48:12 +00:00
|
|
|
join_clone->reuseJoinedData(*join);
|
2021-02-20 15:00:59 +00:00
|
|
|
|
2019-12-19 15:50:28 +00:00
|
|
|
return join_clone;
|
2015-01-28 02:37:05 +00:00
|
|
|
}
|
|
|
|
|
2021-09-30 08:47:15 +00:00
|
|
|
void StorageJoin::insertBlock(const Block & block, ContextPtr context)
|
2021-02-20 15:00:59 +00:00
|
|
|
{
|
2023-01-11 16:17:21 +00:00
|
|
|
Block block_to_insert = block;
|
|
|
|
convertRightBlock(block_to_insert);
|
2023-03-06 10:54:43 +00:00
|
|
|
TableLockHolder holder = tryLockForCurrentQueryTimedWithContext(rwlock, RWLockImpl::Write, context);
|
|
|
|
|
|
|
|
/// Protection from `INSERT INTO test_table_join SELECT * FROM test_table_join`
|
|
|
|
if (!holder)
|
|
|
|
throw Exception(ErrorCodes::DEADLOCK_AVOIDED, "StorageJoin: cannot insert data because current query tries to read from this storage");
|
2023-01-11 16:17:21 +00:00
|
|
|
|
|
|
|
join->addJoinedBlock(block_to_insert, true);
|
2021-02-20 15:00:59 +00:00
|
|
|
}
|
2020-10-23 18:11:55 +00:00
|
|
|
|
2021-09-30 08:47:15 +00:00
|
|
|
size_t StorageJoin::getSize(ContextPtr context) const
|
2021-02-24 16:19:04 +00:00
|
|
|
{
|
2021-09-30 08:47:15 +00:00
|
|
|
TableLockHolder holder = tryLockTimedWithContext(rwlock, RWLockImpl::Read, context);
|
2021-02-24 16:19:04 +00:00
|
|
|
return join->getTotalRowCount();
|
|
|
|
}
|
|
|
|
|
2021-09-29 17:30:07 +00:00
|
|
|
std::optional<UInt64> StorageJoin::totalRows(const Settings &settings) const
|
2021-02-24 16:19:04 +00:00
|
|
|
{
|
2021-09-29 17:30:07 +00:00
|
|
|
TableLockHolder holder = tryLockTimed(rwlock, RWLockImpl::Read, RWLockImpl::NO_QUERY, settings.lock_acquire_timeout);
|
2021-02-24 16:19:04 +00:00
|
|
|
return join->getTotalRowCount();
|
|
|
|
}
|
|
|
|
|
2021-09-29 17:30:07 +00:00
|
|
|
std::optional<UInt64> StorageJoin::totalBytes(const Settings &settings) const
|
2021-02-24 16:19:04 +00:00
|
|
|
{
|
2021-09-29 17:30:07 +00:00
|
|
|
TableLockHolder holder = tryLockTimed(rwlock, RWLockImpl::Read, RWLockImpl::NO_QUERY, settings.lock_acquire_timeout);
|
2021-02-24 16:19:04 +00:00
|
|
|
return join->getTotalByteCount();
|
|
|
|
}
|
2017-01-14 09:00:19 +00:00
|
|
|
|
2021-02-25 09:31:22 +00:00
|
|
|
DataTypePtr StorageJoin::joinGetCheckAndGetReturnType(const DataTypes & data_types, const String & column_name, bool or_null) const
|
|
|
|
{
|
|
|
|
return join->joinGetCheckAndGetReturnType(data_types, column_name, or_null);
|
|
|
|
}
|
|
|
|
|
2021-09-30 08:47:15 +00:00
|
|
|
ColumnWithTypeAndName StorageJoin::joinGet(const Block & block, const Block & block_with_columns_to_add, ContextPtr context) const
|
2021-02-25 09:31:22 +00:00
|
|
|
{
|
2021-09-30 08:47:15 +00:00
|
|
|
TableLockHolder holder = tryLockTimedWithContext(rwlock, RWLockImpl::Read, context);
|
2021-02-25 09:31:22 +00:00
|
|
|
return join->joinGet(block, block_with_columns_to_add);
|
|
|
|
}
|
2017-12-30 00:36:06 +00:00
|
|
|
|
2023-01-11 16:17:21 +00:00
|
|
|
void StorageJoin::convertRightBlock(Block & block) const
|
|
|
|
{
|
|
|
|
bool need_covert = use_nulls && isLeftOrFull(kind);
|
|
|
|
if (!need_covert)
|
|
|
|
return;
|
|
|
|
|
|
|
|
for (auto & col : block)
|
|
|
|
JoinCommon::convertColumnToNullable(col);
|
|
|
|
}
|
|
|
|
|
2017-12-30 00:36:06 +00:00
|
|
|
void registerStorageJoin(StorageFactory & factory)
|
|
|
|
{
|
2020-01-25 19:14:33 +00:00
|
|
|
auto creator_fn = [](const StorageFactory::Arguments & args)
|
2017-12-30 00:36:06 +00:00
|
|
|
{
|
|
|
|
/// Join(ANY, LEFT, k1, k2, ...)
|
|
|
|
|
|
|
|
ASTs & engine_args = args.engine_args;
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
const auto & settings = args.getContext()->getSettingsRef();
|
2019-12-25 17:32:55 +00:00
|
|
|
|
2018-11-30 14:49:35 +00:00
|
|
|
auto join_use_nulls = settings.join_use_nulls;
|
|
|
|
auto max_rows_in_join = settings.max_rows_in_join;
|
|
|
|
auto max_bytes_in_join = settings.max_bytes_in_join;
|
|
|
|
auto join_overflow_mode = settings.join_overflow_mode;
|
2019-01-24 17:12:05 +00:00
|
|
|
auto join_any_take_last_row = settings.join_any_take_last_row;
|
2019-12-25 17:32:55 +00:00
|
|
|
auto old_any_join = settings.any_join_distinct_right_table_keys;
|
2020-09-19 16:38:36 +00:00
|
|
|
bool persistent = true;
|
2020-12-15 16:45:13 +00:00
|
|
|
String disk_name = "default";
|
2018-11-30 14:49:35 +00:00
|
|
|
|
|
|
|
if (args.storage_def && args.storage_def->settings)
|
|
|
|
{
|
2019-04-18 23:29:32 +00:00
|
|
|
for (const auto & setting : args.storage_def->settings->changes)
|
2018-11-30 14:49:35 +00:00
|
|
|
{
|
2018-12-18 01:26:12 +00:00
|
|
|
if (setting.name == "join_use_nulls")
|
2020-07-27 14:26:18 +00:00
|
|
|
join_use_nulls = setting.value;
|
2018-12-18 01:26:12 +00:00
|
|
|
else if (setting.name == "max_rows_in_join")
|
2020-07-27 14:26:18 +00:00
|
|
|
max_rows_in_join = setting.value;
|
2018-12-18 01:26:12 +00:00
|
|
|
else if (setting.name == "max_bytes_in_join")
|
2020-07-27 14:26:18 +00:00
|
|
|
max_bytes_in_join = setting.value;
|
2018-12-18 01:26:12 +00:00
|
|
|
else if (setting.name == "join_overflow_mode")
|
2020-07-27 14:26:18 +00:00
|
|
|
join_overflow_mode = setting.value;
|
2019-01-24 17:12:05 +00:00
|
|
|
else if (setting.name == "join_any_take_last_row")
|
2020-07-27 14:26:18 +00:00
|
|
|
join_any_take_last_row = setting.value;
|
2019-12-25 17:32:55 +00:00
|
|
|
else if (setting.name == "any_join_distinct_right_table_keys")
|
2020-07-27 14:26:18 +00:00
|
|
|
old_any_join = setting.value;
|
2020-12-15 16:45:13 +00:00
|
|
|
else if (setting.name == "disk")
|
|
|
|
disk_name = setting.value.get<String>();
|
2020-09-18 12:58:27 +00:00
|
|
|
else if (setting.name == "persistent")
|
2020-09-15 09:16:10 +00:00
|
|
|
{
|
2021-12-03 09:06:58 +00:00
|
|
|
persistent = setting.value.get<bool>();
|
2020-09-15 09:16:10 +00:00
|
|
|
}
|
2018-11-30 14:49:35 +00:00
|
|
|
else
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unknown setting {} for storage {}", setting.name, args.engine_name);
|
2018-11-30 14:49:35 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
DiskPtr disk = args.getContext()->getDisk(disk_name);
|
2020-12-15 16:45:13 +00:00
|
|
|
|
2019-12-25 17:32:55 +00:00
|
|
|
if (engine_args.size() < 3)
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH,
|
|
|
|
"Storage Join requires at least 3 parameters: "
|
|
|
|
"Join(ANY|ALL|SEMI|ANTI, LEFT|INNER|RIGHT, keys...).");
|
2019-12-25 17:32:55 +00:00
|
|
|
|
2022-07-29 16:30:50 +00:00
|
|
|
JoinStrictness strictness = JoinStrictness::Unspecified;
|
|
|
|
JoinKind kind = JoinKind::Comma;
|
2019-12-25 17:32:55 +00:00
|
|
|
|
|
|
|
if (auto opt_strictness_id = tryGetIdentifierName(engine_args[0]))
|
|
|
|
{
|
|
|
|
const String strictness_str = Poco::toLower(*opt_strictness_id);
|
|
|
|
|
2019-12-27 12:41:55 +00:00
|
|
|
if (strictness_str == "any")
|
2019-12-25 17:32:55 +00:00
|
|
|
{
|
|
|
|
if (old_any_join)
|
2022-07-29 16:30:50 +00:00
|
|
|
strictness = JoinStrictness::RightAny;
|
2019-12-25 17:32:55 +00:00
|
|
|
else
|
2022-07-29 16:30:50 +00:00
|
|
|
strictness = JoinStrictness::Any;
|
2019-12-25 17:32:55 +00:00
|
|
|
}
|
2019-12-27 12:41:55 +00:00
|
|
|
else if (strictness_str == "all")
|
2022-07-29 16:30:50 +00:00
|
|
|
strictness = JoinStrictness::All;
|
2019-12-27 12:41:55 +00:00
|
|
|
else if (strictness_str == "semi")
|
2022-07-29 16:30:50 +00:00
|
|
|
strictness = JoinStrictness::Semi;
|
2019-12-27 12:41:55 +00:00
|
|
|
else if (strictness_str == "anti")
|
2022-07-29 16:30:50 +00:00
|
|
|
strictness = JoinStrictness::Anti;
|
2019-12-25 17:32:55 +00:00
|
|
|
}
|
|
|
|
|
2022-07-29 16:30:50 +00:00
|
|
|
if (strictness == JoinStrictness::Unspecified)
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::BAD_ARGUMENTS, "First parameter of storage Join must be ANY or ALL or SEMI or ANTI (without quotes).");
|
2019-12-25 17:32:55 +00:00
|
|
|
|
|
|
|
if (auto opt_kind_id = tryGetIdentifierName(engine_args[1]))
|
|
|
|
{
|
|
|
|
const String kind_str = Poco::toLower(*opt_kind_id);
|
|
|
|
|
2019-12-27 12:41:55 +00:00
|
|
|
if (kind_str == "left")
|
2022-07-29 16:30:50 +00:00
|
|
|
kind = JoinKind::Left;
|
2019-12-27 12:41:55 +00:00
|
|
|
else if (kind_str == "inner")
|
2022-07-29 16:30:50 +00:00
|
|
|
kind = JoinKind::Inner;
|
2019-12-27 12:41:55 +00:00
|
|
|
else if (kind_str == "right")
|
2022-07-29 16:30:50 +00:00
|
|
|
kind = JoinKind::Right;
|
2019-12-27 12:41:55 +00:00
|
|
|
else if (kind_str == "full")
|
2019-12-25 17:32:55 +00:00
|
|
|
{
|
2022-07-29 16:30:50 +00:00
|
|
|
if (strictness == JoinStrictness::Any)
|
|
|
|
strictness = JoinStrictness::RightAny;
|
|
|
|
kind = JoinKind::Full;
|
2019-12-25 17:32:55 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-07-29 16:30:50 +00:00
|
|
|
if (kind == JoinKind::Comma)
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Second parameter of storage Join must be LEFT or INNER or RIGHT or FULL (without quotes).");
|
2019-12-25 17:32:55 +00:00
|
|
|
|
|
|
|
Names key_names;
|
|
|
|
key_names.reserve(engine_args.size() - 2);
|
|
|
|
for (size_t i = 2, size = engine_args.size(); i < size; ++i)
|
|
|
|
{
|
|
|
|
auto opt_key = tryGetIdentifierName(engine_args[i]);
|
|
|
|
if (!opt_key)
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Parameter №{} of storage Join don't look like column name.", i + 1);
|
2019-12-25 17:32:55 +00:00
|
|
|
|
|
|
|
key_names.push_back(*opt_key);
|
|
|
|
}
|
|
|
|
|
2022-04-19 20:47:29 +00:00
|
|
|
return std::make_shared<StorageJoin>(
|
2020-12-15 16:45:13 +00:00
|
|
|
disk,
|
2019-10-25 19:07:47 +00:00
|
|
|
args.relative_data_path,
|
2019-12-04 16:06:55 +00:00
|
|
|
args.table_id,
|
2018-11-30 14:49:35 +00:00
|
|
|
key_names,
|
2019-08-09 13:02:01 +00:00
|
|
|
join_use_nulls,
|
|
|
|
SizeLimits{max_rows_in_join, max_bytes_in_join, join_overflow_mode},
|
2018-11-30 14:49:35 +00:00
|
|
|
kind,
|
|
|
|
strictness,
|
2018-12-30 15:54:45 +00:00
|
|
|
args.columns,
|
2019-08-24 21:20:20 +00:00
|
|
|
args.constraints,
|
2021-04-23 12:18:23 +00:00
|
|
|
args.comment,
|
2019-10-25 19:07:47 +00:00
|
|
|
join_any_take_last_row,
|
2020-09-19 16:38:36 +00:00
|
|
|
persistent);
|
2020-01-25 19:14:33 +00:00
|
|
|
};
|
|
|
|
|
|
|
|
factory.registerStorage("Join", creator_fn, StorageFactory::StorageFeatures{ .supports_settings = true, });
|
2017-12-30 00:36:06 +00:00
|
|
|
}
|
|
|
|
|
2018-11-30 14:49:35 +00:00
|
|
|
template <typename T>
|
|
|
|
static const char * rawData(T & t)
|
|
|
|
{
|
|
|
|
return reinterpret_cast<const char *>(&t);
|
|
|
|
}
|
|
|
|
template <typename T>
|
|
|
|
static size_t rawSize(T &)
|
|
|
|
{
|
|
|
|
return sizeof(T);
|
|
|
|
}
|
|
|
|
template <>
|
|
|
|
const char * rawData(const StringRef & t)
|
|
|
|
{
|
|
|
|
return t.data;
|
|
|
|
}
|
|
|
|
template <>
|
|
|
|
size_t rawSize(const StringRef & t)
|
|
|
|
{
|
|
|
|
return t.size;
|
|
|
|
}
|
|
|
|
|
2022-05-20 19:49:31 +00:00
|
|
|
class JoinSource : public ISource
|
2018-11-30 14:49:35 +00:00
|
|
|
{
|
|
|
|
public:
|
2021-09-29 17:30:07 +00:00
|
|
|
JoinSource(HashJoinPtr join_, TableLockHolder lock_holder_, UInt64 max_block_size_, Block sample_block_)
|
2022-05-20 19:49:31 +00:00
|
|
|
: ISource(sample_block_)
|
2021-02-25 09:31:22 +00:00
|
|
|
, join(join_)
|
2021-09-29 17:30:07 +00:00
|
|
|
, lock_holder(lock_holder_)
|
2020-01-31 14:51:09 +00:00
|
|
|
, max_block_size(max_block_size_)
|
|
|
|
, sample_block(std::move(sample_block_))
|
2018-11-30 14:49:35 +00:00
|
|
|
{
|
2021-09-10 14:52:44 +00:00
|
|
|
if (!join->getTableJoin().oneDisjunct())
|
|
|
|
throw DB::Exception(ErrorCodes::NOT_IMPLEMENTED, "StorageJoin does not support OR for keys in JOIN ON section");
|
|
|
|
|
2018-11-30 14:49:35 +00:00
|
|
|
column_indices.resize(sample_block.columns());
|
2020-06-01 17:27:22 +00:00
|
|
|
|
2021-02-25 09:31:22 +00:00
|
|
|
auto & saved_block = join->getJoinedData()->sample_block;
|
2020-06-01 17:27:22 +00:00
|
|
|
|
2018-11-30 14:49:35 +00:00
|
|
|
for (size_t i = 0; i < sample_block.columns(); ++i)
|
|
|
|
{
|
|
|
|
auto & [_, type, name] = sample_block.getByPosition(i);
|
2021-02-25 09:31:22 +00:00
|
|
|
if (join->right_table_keys.has(name))
|
2018-11-30 14:49:35 +00:00
|
|
|
{
|
|
|
|
key_pos = i;
|
2021-02-25 09:31:22 +00:00
|
|
|
const auto & column = join->right_table_keys.getByName(name);
|
2020-06-01 17:27:22 +00:00
|
|
|
restored_block.insert(column);
|
2018-11-30 14:49:35 +00:00
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
2020-06-01 17:27:22 +00:00
|
|
|
size_t pos = saved_block.getPositionByName(name);
|
2018-11-30 14:49:35 +00:00
|
|
|
column_indices[i] = pos;
|
2020-06-01 17:27:22 +00:00
|
|
|
|
2020-06-01 19:38:51 +00:00
|
|
|
const auto & column = saved_block.getByPosition(pos);
|
2020-06-01 17:27:22 +00:00
|
|
|
restored_block.insert(column);
|
2018-11-30 14:49:35 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
String getName() const override { return "Join"; }
|
|
|
|
|
|
|
|
protected:
|
2020-01-31 14:51:09 +00:00
|
|
|
Chunk generate() override
|
2018-11-30 14:49:35 +00:00
|
|
|
{
|
2021-02-25 09:31:22 +00:00
|
|
|
if (join->data->blocks.empty())
|
2020-01-31 14:51:09 +00:00
|
|
|
return {};
|
2018-11-30 14:49:35 +00:00
|
|
|
|
2020-01-31 14:51:09 +00:00
|
|
|
Chunk chunk;
|
2021-06-25 12:03:10 +00:00
|
|
|
if (!joinDispatch(join->kind, join->strictness, join->data->maps.front(),
|
2020-01-31 14:51:09 +00:00
|
|
|
[&](auto kind, auto strictness, auto & map) { chunk = createChunk<kind, strictness>(map); }))
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::LOGICAL_ERROR, "Logical error: unknown JOIN strictness");
|
2020-01-31 14:51:09 +00:00
|
|
|
return chunk;
|
2018-11-30 14:49:35 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
private:
|
2021-02-25 09:31:22 +00:00
|
|
|
HashJoinPtr join;
|
2021-09-29 17:30:07 +00:00
|
|
|
TableLockHolder lock_holder;
|
2021-02-25 09:31:22 +00:00
|
|
|
|
2019-02-10 16:55:12 +00:00
|
|
|
UInt64 max_block_size;
|
2018-11-30 14:49:35 +00:00
|
|
|
Block sample_block;
|
2020-06-01 17:27:22 +00:00
|
|
|
Block restored_block; /// sample_block with parent column types
|
2018-11-30 14:49:35 +00:00
|
|
|
|
|
|
|
ColumnNumbers column_indices;
|
|
|
|
std::optional<size_t> key_pos;
|
|
|
|
|
|
|
|
std::unique_ptr<void, std::function<void(void *)>> position; /// type erasure
|
|
|
|
|
|
|
|
|
2022-07-29 16:30:50 +00:00
|
|
|
template <JoinKind KIND, JoinStrictness STRICTNESS, typename Maps>
|
2020-01-31 14:51:09 +00:00
|
|
|
Chunk createChunk(const Maps & maps)
|
2018-11-30 14:49:35 +00:00
|
|
|
{
|
2020-11-25 12:41:12 +00:00
|
|
|
MutableColumns mut_columns = restored_block.cloneEmpty().mutateColumns();
|
2018-11-30 14:49:35 +00:00
|
|
|
|
|
|
|
size_t rows_added = 0;
|
|
|
|
|
2021-02-25 09:31:22 +00:00
|
|
|
switch (join->data->type)
|
2018-11-30 14:49:35 +00:00
|
|
|
{
|
|
|
|
#define M(TYPE) \
|
2020-04-07 09:48:47 +00:00
|
|
|
case HashJoin::Type::TYPE: \
|
2020-11-25 12:41:12 +00:00
|
|
|
rows_added = fillColumns<KIND, STRICTNESS>(*maps.TYPE, mut_columns); \
|
2018-11-30 14:49:35 +00:00
|
|
|
break;
|
|
|
|
APPLY_FOR_JOIN_VARIANTS_LIMITED(M)
|
|
|
|
#undef M
|
|
|
|
|
|
|
|
default:
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::UNSUPPORTED_JOIN_KEYS, "Unsupported JOIN keys in StorageJoin. Type: {}",
|
|
|
|
static_cast<UInt32>(join->data->type));
|
2018-11-30 14:49:35 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
if (!rows_added)
|
|
|
|
return {};
|
|
|
|
|
2020-11-25 12:41:12 +00:00
|
|
|
Columns columns;
|
|
|
|
columns.reserve(mut_columns.size());
|
|
|
|
for (auto & col : mut_columns)
|
|
|
|
columns.emplace_back(std::move(col));
|
|
|
|
|
|
|
|
/// Correct nullability and LowCardinality types
|
2018-11-30 14:49:35 +00:00
|
|
|
for (size_t i = 0; i < columns.size(); ++i)
|
2020-06-01 17:27:22 +00:00
|
|
|
{
|
2020-11-25 12:41:12 +00:00
|
|
|
const auto & src = restored_block.getByPosition(i);
|
|
|
|
const auto & dst = sample_block.getByPosition(i);
|
2020-06-01 17:27:22 +00:00
|
|
|
|
2020-11-25 12:41:12 +00:00
|
|
|
if (!src.type->equals(*dst.type))
|
2018-11-30 14:49:35 +00:00
|
|
|
{
|
2020-11-25 12:41:12 +00:00
|
|
|
auto arg = src;
|
|
|
|
arg.column = std::move(columns[i]);
|
|
|
|
columns[i] = castColumn(arg, dst.type);
|
2018-11-30 14:49:35 +00:00
|
|
|
}
|
2020-06-01 17:27:22 +00:00
|
|
|
}
|
2018-11-30 14:49:35 +00:00
|
|
|
|
2020-06-01 17:27:22 +00:00
|
|
|
UInt64 num_rows = columns.at(0)->size();
|
|
|
|
return Chunk(std::move(columns), num_rows);
|
2018-11-30 14:49:35 +00:00
|
|
|
}
|
|
|
|
|
2022-07-29 16:30:50 +00:00
|
|
|
template <JoinKind KIND, JoinStrictness STRICTNESS, typename Map>
|
2020-06-01 17:27:22 +00:00
|
|
|
size_t fillColumns(const Map & map, MutableColumns & columns)
|
2018-11-30 14:49:35 +00:00
|
|
|
{
|
|
|
|
size_t rows_added = 0;
|
|
|
|
|
|
|
|
if (!position)
|
|
|
|
position = decltype(position)(
|
2023-02-19 22:15:09 +00:00
|
|
|
static_cast<void *>(new typename Map::const_iterator(map.begin())),
|
2018-11-30 14:49:35 +00:00
|
|
|
[](void * ptr) { delete reinterpret_cast<typename Map::const_iterator *>(ptr); });
|
|
|
|
|
|
|
|
auto & it = *reinterpret_cast<typename Map::const_iterator *>(position.get());
|
|
|
|
auto end = map.end();
|
|
|
|
|
|
|
|
for (; it != end; ++it)
|
|
|
|
{
|
2022-07-29 16:30:50 +00:00
|
|
|
if constexpr (STRICTNESS == JoinStrictness::RightAny)
|
2018-11-30 14:49:35 +00:00
|
|
|
{
|
2019-12-25 17:32:55 +00:00
|
|
|
fillOne<Map>(columns, column_indices, it, key_pos, rows_added);
|
|
|
|
}
|
2022-07-29 16:30:50 +00:00
|
|
|
else if constexpr (STRICTNESS == JoinStrictness::All)
|
2019-12-25 17:32:55 +00:00
|
|
|
{
|
|
|
|
fillAll<Map>(columns, column_indices, it, key_pos, rows_added);
|
2018-11-30 14:49:35 +00:00
|
|
|
}
|
2022-07-29 16:30:50 +00:00
|
|
|
else if constexpr (STRICTNESS == JoinStrictness::Any)
|
2019-11-07 21:32:44 +00:00
|
|
|
{
|
2022-07-29 16:30:50 +00:00
|
|
|
if constexpr (KIND == JoinKind::Left || KIND == JoinKind::Inner)
|
2019-12-25 17:32:55 +00:00
|
|
|
fillOne<Map>(columns, column_indices, it, key_pos, rows_added);
|
2022-07-29 16:30:50 +00:00
|
|
|
else if constexpr (KIND == JoinKind::Right)
|
2019-12-25 17:32:55 +00:00
|
|
|
fillAll<Map>(columns, column_indices, it, key_pos, rows_added);
|
|
|
|
}
|
2022-07-29 16:30:50 +00:00
|
|
|
else if constexpr (STRICTNESS == JoinStrictness::Semi)
|
2019-12-25 17:32:55 +00:00
|
|
|
{
|
2022-07-29 16:30:50 +00:00
|
|
|
if constexpr (KIND == JoinKind::Left)
|
2019-12-25 17:32:55 +00:00
|
|
|
fillOne<Map>(columns, column_indices, it, key_pos, rows_added);
|
2022-07-29 16:30:50 +00:00
|
|
|
else if constexpr (KIND == JoinKind::Right)
|
2019-12-25 17:32:55 +00:00
|
|
|
fillAll<Map>(columns, column_indices, it, key_pos, rows_added);
|
2019-11-07 21:32:44 +00:00
|
|
|
}
|
2022-07-29 16:30:50 +00:00
|
|
|
else if constexpr (STRICTNESS == JoinStrictness::Anti)
|
2019-03-26 22:05:51 +00:00
|
|
|
{
|
2022-07-29 16:30:50 +00:00
|
|
|
if constexpr (KIND == JoinKind::Left)
|
2019-12-25 17:32:55 +00:00
|
|
|
fillOne<Map>(columns, column_indices, it, key_pos, rows_added);
|
2022-07-29 16:30:50 +00:00
|
|
|
else if constexpr (KIND == JoinKind::Right)
|
2019-12-25 17:32:55 +00:00
|
|
|
fillAll<Map>(columns, column_indices, it, key_pos, rows_added);
|
2019-03-20 16:58:28 +00:00
|
|
|
}
|
2018-11-30 14:49:35 +00:00
|
|
|
else
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::NOT_IMPLEMENTED, "This JOIN is not implemented yet");
|
2018-11-30 14:49:35 +00:00
|
|
|
|
|
|
|
if (rows_added >= max_block_size)
|
|
|
|
{
|
|
|
|
++it;
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return rows_added;
|
|
|
|
}
|
2019-12-25 17:32:55 +00:00
|
|
|
|
|
|
|
template <typename Map>
|
|
|
|
static void fillOne(MutableColumns & columns, const ColumnNumbers & column_indices, typename Map::const_iterator & it,
|
|
|
|
const std::optional<size_t> & key_pos, size_t & rows_added)
|
|
|
|
{
|
|
|
|
for (size_t j = 0; j < columns.size(); ++j)
|
|
|
|
if (j == key_pos)
|
|
|
|
columns[j]->insertData(rawData(it->getKey()), rawSize(it->getKey()));
|
|
|
|
else
|
|
|
|
columns[j]->insertFrom(*it->getMapped().block->getByPosition(column_indices[j]).column.get(), it->getMapped().row_num);
|
|
|
|
++rows_added;
|
|
|
|
}
|
|
|
|
|
|
|
|
template <typename Map>
|
|
|
|
static void fillAll(MutableColumns & columns, const ColumnNumbers & column_indices, typename Map::const_iterator & it,
|
|
|
|
const std::optional<size_t> & key_pos, size_t & rows_added)
|
|
|
|
{
|
|
|
|
for (auto ref_it = it->getMapped().begin(); ref_it.ok(); ++ref_it)
|
|
|
|
{
|
|
|
|
for (size_t j = 0; j < columns.size(); ++j)
|
|
|
|
if (j == key_pos)
|
|
|
|
columns[j]->insertData(rawData(it->getKey()), rawSize(it->getKey()));
|
|
|
|
else
|
|
|
|
columns[j]->insertFrom(*ref_it->block->getByPosition(column_indices[j]).column.get(), ref_it->row_num);
|
|
|
|
++rows_added;
|
|
|
|
}
|
|
|
|
}
|
2018-11-30 14:49:35 +00:00
|
|
|
};
|
|
|
|
|
|
|
|
|
|
|
|
// TODO: multiple stream read and index read
|
2020-08-03 13:54:14 +00:00
|
|
|
Pipe StorageJoin::read(
|
2018-11-30 14:49:35 +00:00
|
|
|
const Names & column_names,
|
2021-07-09 03:15:41 +00:00
|
|
|
const StorageSnapshotPtr & storage_snapshot,
|
2020-09-20 17:52:17 +00:00
|
|
|
SelectQueryInfo & /*query_info*/,
|
2021-09-29 17:30:07 +00:00
|
|
|
ContextPtr context,
|
2018-11-30 14:49:35 +00:00
|
|
|
QueryProcessingStage::Enum /*processed_stage*/,
|
2019-02-18 23:38:44 +00:00
|
|
|
size_t max_block_size,
|
2022-10-07 10:46:45 +00:00
|
|
|
size_t /*num_streams*/)
|
2018-11-30 14:49:35 +00:00
|
|
|
{
|
2021-07-09 03:15:41 +00:00
|
|
|
storage_snapshot->check(column_names);
|
2020-01-31 14:51:09 +00:00
|
|
|
|
2021-07-09 03:15:41 +00:00
|
|
|
Block source_sample_block = storage_snapshot->getSampleBlockForColumns(column_names);
|
2021-09-29 17:30:07 +00:00
|
|
|
RWLockImpl::LockHolder holder = tryLockTimedWithContext(rwlock, RWLockImpl::Read, context);
|
|
|
|
return Pipe(std::make_shared<JoinSource>(join, std::move(holder), max_block_size, source_sample_block));
|
2018-11-30 14:49:35 +00:00
|
|
|
}
|
|
|
|
|
2015-01-27 21:24:24 +00:00
|
|
|
}
|