2019-06-24 11:17:15 +00:00
|
|
|
#include "StorageMySQL.h"
|
2017-12-30 00:36:06 +00:00
|
|
|
|
2017-12-29 15:48:20 +00:00
|
|
|
#if USE_MYSQL
|
2018-05-10 09:23:38 +00:00
|
|
|
|
2017-12-30 00:36:06 +00:00
|
|
|
#include <Storages/StorageFactory.h>
|
2017-12-26 21:34:06 +00:00
|
|
|
#include <Storages/transformQueryForExternalDatabase.h>
|
2019-02-15 11:46:07 +00:00
|
|
|
#include <Formats/MySQLBlockInputStream.h>
|
2017-12-30 00:36:06 +00:00
|
|
|
#include <Interpreters/evaluateConstantExpression.h>
|
2019-03-22 12:08:30 +00:00
|
|
|
#include <Core/Settings.h>
|
2018-05-11 04:15:22 +00:00
|
|
|
#include <Interpreters/Context.h>
|
2018-05-10 09:23:38 +00:00
|
|
|
#include <DataStreams/IBlockOutputStream.h>
|
2018-06-10 19:22:49 +00:00
|
|
|
#include <Formats/FormatFactory.h>
|
2017-12-30 00:36:06 +00:00
|
|
|
#include <Common/parseAddress.h>
|
2018-05-10 09:23:38 +00:00
|
|
|
#include <IO/Operators.h>
|
|
|
|
#include <IO/WriteHelpers.h>
|
2017-12-30 00:36:06 +00:00
|
|
|
#include <Parsers/ASTLiteral.h>
|
2018-05-10 09:23:38 +00:00
|
|
|
#include <mysqlxx/Transaction.h>
|
2020-01-31 16:46:27 +00:00
|
|
|
#include <Processors/Sources/SourceFromInputStream.h>
|
|
|
|
#include <Processors/Pipe.h>
|
2017-12-05 13:32:02 +00:00
|
|
|
|
2017-12-26 18:32:17 +00:00
|
|
|
|
2017-12-05 13:32:02 +00:00
|
|
|
namespace DB
|
|
|
|
{
|
2017-12-26 18:32:17 +00:00
|
|
|
|
2017-12-30 00:36:06 +00:00
|
|
|
namespace ErrorCodes
|
|
|
|
{
|
|
|
|
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
|
2018-05-14 11:00:22 +00:00
|
|
|
extern const int BAD_ARGUMENTS;
|
2017-12-30 00:36:06 +00:00
|
|
|
}
|
|
|
|
|
2019-12-15 06:34:43 +00:00
|
|
|
static String backQuoteMySQL(const String & x)
|
2019-06-21 05:22:04 +00:00
|
|
|
{
|
|
|
|
String res(x.size(), '\0');
|
|
|
|
{
|
|
|
|
WriteBufferFromString wb(res);
|
2019-06-24 14:51:37 +00:00
|
|
|
writeBackQuotedStringMySQL(x, wb);
|
2019-06-21 05:22:04 +00:00
|
|
|
}
|
|
|
|
return res;
|
|
|
|
}
|
2017-12-30 00:36:06 +00:00
|
|
|
|
2019-07-09 15:40:21 +00:00
|
|
|
StorageMySQL::StorageMySQL(
|
2019-12-04 16:06:55 +00:00
|
|
|
const StorageID & table_id_,
|
2019-08-03 11:02:40 +00:00
|
|
|
mysqlxx::Pool && pool_,
|
|
|
|
const std::string & remote_database_name_,
|
|
|
|
const std::string & remote_table_name_,
|
|
|
|
const bool replace_query_,
|
|
|
|
const std::string & on_duplicate_clause_,
|
2018-05-11 04:15:22 +00:00
|
|
|
const ColumnsDescription & columns_,
|
2019-08-24 21:20:20 +00:00
|
|
|
const ConstraintsDescription & constraints_,
|
2019-08-03 11:02:40 +00:00
|
|
|
const Context & context_)
|
2019-12-04 16:06:55 +00:00
|
|
|
: IStorage(table_id_)
|
2019-08-03 11:02:40 +00:00
|
|
|
, remote_database_name(remote_database_name_)
|
|
|
|
, remote_table_name(remote_table_name_)
|
|
|
|
, replace_query{replace_query_}
|
|
|
|
, on_duplicate_clause{on_duplicate_clause_}
|
|
|
|
, pool(std::move(pool_))
|
|
|
|
, global_context(context_)
|
2017-12-05 13:32:02 +00:00
|
|
|
{
|
2020-06-15 16:55:33 +00:00
|
|
|
StorageInMemoryMetadata metadata_;
|
|
|
|
metadata_.setColumns(columns_);
|
|
|
|
metadata_.setConstraints(constraints_);
|
|
|
|
setInMemoryMetadata(metadata_);
|
2017-12-05 13:32:02 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
2020-02-19 16:07:28 +00:00
|
|
|
Pipes StorageMySQL::read(
|
2019-08-03 11:02:40 +00:00
|
|
|
const Names & column_names_,
|
2020-06-16 15:51:29 +00:00
|
|
|
const StorageMetadataPtr & metadata_snapshot,
|
2019-08-03 11:02:40 +00:00
|
|
|
const SelectQueryInfo & query_info_,
|
|
|
|
const Context & context_,
|
2018-09-08 11:29:23 +00:00
|
|
|
QueryProcessingStage::Enum /*processed_stage*/,
|
2019-08-03 11:02:40 +00:00
|
|
|
size_t max_block_size_,
|
2017-12-26 21:34:06 +00:00
|
|
|
unsigned)
|
2017-12-05 13:32:02 +00:00
|
|
|
{
|
2019-08-03 11:02:40 +00:00
|
|
|
check(column_names_);
|
2018-07-16 01:02:46 +00:00
|
|
|
String query = transformQueryForExternalDatabase(
|
2020-06-16 15:51:29 +00:00
|
|
|
query_info_,
|
|
|
|
metadata_snapshot->getColumns().getOrdinary(),
|
|
|
|
IdentifierQuotingStyle::BackticksMySQL,
|
|
|
|
remote_database_name,
|
|
|
|
remote_table_name,
|
|
|
|
context_);
|
2017-12-28 05:26:45 +00:00
|
|
|
|
|
|
|
Block sample_block;
|
2019-08-03 11:02:40 +00:00
|
|
|
for (const String & column_name : column_names_)
|
2017-12-28 05:26:45 +00:00
|
|
|
{
|
2020-06-16 15:51:29 +00:00
|
|
|
auto column_data = metadata_snapshot->getColumns().getPhysical(column_name);
|
2017-12-28 05:26:45 +00:00
|
|
|
sample_block.insert({ column_data.type, column_data.name });
|
|
|
|
}
|
|
|
|
|
2020-01-31 16:46:27 +00:00
|
|
|
Pipes pipes;
|
|
|
|
/// TODO: rewrite MySQLBlockInputStream
|
|
|
|
pipes.emplace_back(std::make_shared<SourceFromInputStream>(
|
2020-03-23 02:12:31 +00:00
|
|
|
std::make_shared<MySQLBlockInputStream>(pool.get(), query, sample_block, max_block_size_)));
|
2020-01-31 16:46:27 +00:00
|
|
|
|
|
|
|
return pipes;
|
2017-12-05 13:32:02 +00:00
|
|
|
}
|
2017-12-26 21:34:06 +00:00
|
|
|
|
2017-12-30 00:36:06 +00:00
|
|
|
|
2018-05-10 09:23:38 +00:00
|
|
|
class StorageMySQLBlockOutputStream : public IBlockOutputStream
|
|
|
|
{
|
|
|
|
public:
|
2020-06-16 15:51:29 +00:00
|
|
|
explicit StorageMySQLBlockOutputStream(
|
|
|
|
const StorageMySQL & storage_,
|
|
|
|
const StorageMetadataPtr & metadata_snapshot_,
|
2019-08-03 11:02:40 +00:00
|
|
|
const std::string & remote_database_name_,
|
|
|
|
const std::string & remote_table_name_,
|
|
|
|
const mysqlxx::PoolWithFailover::Entry & entry_,
|
2018-05-14 11:00:22 +00:00
|
|
|
const size_t & mysql_max_rows_to_insert)
|
2019-08-03 11:02:40 +00:00
|
|
|
: storage{storage_}
|
2020-06-16 15:51:29 +00:00
|
|
|
, metadata_snapshot{metadata_snapshot_}
|
2019-08-03 11:02:40 +00:00
|
|
|
, remote_database_name{remote_database_name_}
|
|
|
|
, remote_table_name{remote_table_name_}
|
|
|
|
, entry{entry_}
|
2018-05-10 09:23:38 +00:00
|
|
|
, max_batch_rows{mysql_max_rows_to_insert}
|
|
|
|
{
|
|
|
|
}
|
|
|
|
|
2020-06-16 15:51:29 +00:00
|
|
|
Block getHeader() const override { return metadata_snapshot->getSampleBlock(); }
|
2018-05-14 11:00:22 +00:00
|
|
|
|
|
|
|
void write(const Block & block) override
|
|
|
|
{
|
|
|
|
auto blocks = splitBlocks(block, max_batch_rows);
|
|
|
|
mysqlxx::Transaction trans(entry);
|
|
|
|
try
|
|
|
|
{
|
|
|
|
for (const Block & batch_data : blocks)
|
|
|
|
{
|
|
|
|
writeBlockData(batch_data);
|
|
|
|
}
|
|
|
|
trans.commit();
|
|
|
|
}
|
2018-11-23 18:52:00 +00:00
|
|
|
catch (...)
|
2018-05-14 11:00:22 +00:00
|
|
|
{
|
|
|
|
trans.rollback();
|
|
|
|
throw;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
void writeBlockData(const Block & block)
|
|
|
|
{
|
|
|
|
WriteBufferFromOwnString sqlbuf;
|
|
|
|
sqlbuf << (storage.replace_query ? "REPLACE" : "INSERT") << " INTO ";
|
2019-06-21 05:22:04 +00:00
|
|
|
sqlbuf << backQuoteMySQL(remote_database_name) << "." << backQuoteMySQL(remote_table_name);
|
2018-11-24 01:48:06 +00:00
|
|
|
sqlbuf << " (" << dumpNamesWithBackQuote(block) << ") VALUES ";
|
2018-05-14 11:00:22 +00:00
|
|
|
|
2020-06-16 15:51:29 +00:00
|
|
|
auto writer = FormatFactory::instance().getOutput("Values", sqlbuf, metadata_snapshot->getSampleBlock(), storage.global_context);
|
2018-05-14 11:00:22 +00:00
|
|
|
writer->write(block);
|
|
|
|
|
|
|
|
if (!storage.on_duplicate_clause.empty())
|
|
|
|
sqlbuf << " ON DUPLICATE KEY " << storage.on_duplicate_clause;
|
|
|
|
|
|
|
|
sqlbuf << ";";
|
|
|
|
|
|
|
|
auto query = this->entry->query(sqlbuf.str());
|
|
|
|
query.execute();
|
|
|
|
}
|
|
|
|
|
|
|
|
Blocks splitBlocks(const Block & block, const size_t & max_rows) const
|
|
|
|
{
|
|
|
|
/// Avoid Excessive copy when block is small enough
|
|
|
|
if (block.rows() <= max_rows)
|
|
|
|
return Blocks{std::move(block)};
|
|
|
|
|
|
|
|
const size_t splited_block_size = ceil(block.rows() * 1.0 / max_rows);
|
|
|
|
Blocks splitted_blocks(splited_block_size);
|
|
|
|
|
|
|
|
for (size_t idx = 0; idx < splited_block_size; ++idx)
|
|
|
|
splitted_blocks[idx] = block.cloneEmpty();
|
|
|
|
|
|
|
|
const size_t columns = block.columns();
|
|
|
|
const size_t rows = block.rows();
|
|
|
|
size_t offsets = 0;
|
2019-02-10 15:17:45 +00:00
|
|
|
UInt64 limits = max_batch_rows;
|
2018-05-14 11:00:22 +00:00
|
|
|
for (size_t idx = 0; idx < splited_block_size; ++idx)
|
|
|
|
{
|
|
|
|
/// For last batch, limits should be the remain size
|
|
|
|
if (idx == splited_block_size - 1) limits = rows - offsets;
|
|
|
|
for (size_t col_idx = 0; col_idx < columns; ++col_idx)
|
|
|
|
{
|
|
|
|
splitted_blocks[idx].getByPosition(col_idx).column = block.getByPosition(col_idx).column->cut(offsets, limits);
|
|
|
|
}
|
|
|
|
offsets += max_batch_rows;
|
|
|
|
}
|
|
|
|
|
|
|
|
return splitted_blocks;
|
|
|
|
}
|
|
|
|
|
2020-03-18 00:57:00 +00:00
|
|
|
static std::string dumpNamesWithBackQuote(const Block & block)
|
2018-05-14 11:00:22 +00:00
|
|
|
{
|
|
|
|
WriteBufferFromOwnString out;
|
|
|
|
for (auto it = block.begin(); it != block.end(); ++it)
|
|
|
|
{
|
|
|
|
if (it != block.begin())
|
|
|
|
out << ", ";
|
2019-06-21 05:22:04 +00:00
|
|
|
out << backQuoteMySQL(it->name);
|
2018-05-14 11:00:22 +00:00
|
|
|
}
|
|
|
|
return out.str();
|
|
|
|
}
|
2018-05-10 09:23:38 +00:00
|
|
|
|
|
|
|
private:
|
2018-05-14 11:00:22 +00:00
|
|
|
const StorageMySQL & storage;
|
2020-06-16 15:51:29 +00:00
|
|
|
StorageMetadataPtr metadata_snapshot;
|
2018-05-14 11:00:22 +00:00
|
|
|
std::string remote_database_name;
|
|
|
|
std::string remote_table_name;
|
|
|
|
mysqlxx::PoolWithFailover::Entry entry;
|
|
|
|
size_t max_batch_rows;
|
2018-05-10 09:23:38 +00:00
|
|
|
};
|
|
|
|
|
|
|
|
|
2020-06-16 15:51:29 +00:00
|
|
|
BlockOutputStreamPtr StorageMySQL::write(const ASTPtr & /*query*/, const StorageMetadataPtr & metadata_snapshot, const Context & context)
|
2018-05-10 09:23:38 +00:00
|
|
|
{
|
2020-06-16 15:51:29 +00:00
|
|
|
return std::make_shared<StorageMySQLBlockOutputStream>(*this, metadata_snapshot, remote_database_name, remote_table_name, pool.get(), context.getSettingsRef().mysql_max_rows_to_insert);
|
2018-05-10 09:23:38 +00:00
|
|
|
}
|
|
|
|
|
2017-12-30 00:36:06 +00:00
|
|
|
void registerStorageMySQL(StorageFactory & factory)
|
|
|
|
{
|
|
|
|
factory.registerStorage("MySQL", [](const StorageFactory::Arguments & args)
|
|
|
|
{
|
|
|
|
ASTs & engine_args = args.engine_args;
|
|
|
|
|
2018-05-13 02:34:49 +00:00
|
|
|
if (engine_args.size() < 5 || engine_args.size() > 7)
|
2017-12-30 00:36:06 +00:00
|
|
|
throw Exception(
|
2018-05-14 11:00:22 +00:00
|
|
|
"Storage MySQL requires 5-7 parameters: MySQL('host:port', database, table, 'user', 'password'[, replace_query, 'on_duplicate_clause']).",
|
2017-12-30 00:36:06 +00:00
|
|
|
ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
|
|
|
|
|
2020-03-09 01:22:33 +00:00
|
|
|
for (auto & engine_arg : engine_args)
|
|
|
|
engine_arg = evaluateConstantExpressionOrIdentifierAsLiteral(engine_arg, args.local_context);
|
2017-12-30 00:36:06 +00:00
|
|
|
|
|
|
|
/// 3306 is the default MySQL port.
|
2019-03-15 17:09:14 +00:00
|
|
|
auto parsed_host_port = parseAddress(engine_args[0]->as<ASTLiteral &>().value.safeGet<String>(), 3306);
|
2017-12-30 00:36:06 +00:00
|
|
|
|
2019-03-15 17:09:14 +00:00
|
|
|
const String & remote_database = engine_args[1]->as<ASTLiteral &>().value.safeGet<String>();
|
|
|
|
const String & remote_table = engine_args[2]->as<ASTLiteral &>().value.safeGet<String>();
|
|
|
|
const String & username = engine_args[3]->as<ASTLiteral &>().value.safeGet<String>();
|
|
|
|
const String & password = engine_args[4]->as<ASTLiteral &>().value.safeGet<String>();
|
2017-12-30 00:36:06 +00:00
|
|
|
|
|
|
|
mysqlxx::Pool pool(remote_database, parsed_host_port.first, username, password, parsed_host_port.second);
|
|
|
|
|
2018-05-13 02:34:49 +00:00
|
|
|
bool replace_query = false;
|
|
|
|
std::string on_duplicate_clause;
|
2018-05-14 11:00:22 +00:00
|
|
|
if (engine_args.size() >= 6)
|
2019-03-15 17:09:14 +00:00
|
|
|
replace_query = engine_args[5]->as<ASTLiteral &>().value.safeGet<UInt64>();
|
2018-05-14 11:00:22 +00:00
|
|
|
if (engine_args.size() == 7)
|
2019-03-15 17:09:14 +00:00
|
|
|
on_duplicate_clause = engine_args[6]->as<ASTLiteral &>().value.safeGet<String>();
|
2018-05-14 11:00:22 +00:00
|
|
|
|
|
|
|
if (replace_query && !on_duplicate_clause.empty())
|
|
|
|
throw Exception(
|
|
|
|
"Only one of 'replace_query' and 'on_duplicate_clause' can be specified, or none of them",
|
|
|
|
ErrorCodes::BAD_ARGUMENTS);
|
2018-05-13 02:34:49 +00:00
|
|
|
|
2017-12-30 00:36:06 +00:00
|
|
|
return StorageMySQL::create(
|
2019-12-04 16:06:55 +00:00
|
|
|
args.table_id,
|
2017-12-30 00:36:06 +00:00
|
|
|
std::move(pool),
|
|
|
|
remote_database,
|
|
|
|
remote_table,
|
2018-05-13 02:34:49 +00:00
|
|
|
replace_query,
|
|
|
|
on_duplicate_clause,
|
2018-05-11 04:15:22 +00:00
|
|
|
args.columns,
|
2019-08-24 21:20:20 +00:00
|
|
|
args.constraints,
|
2018-05-11 04:15:22 +00:00
|
|
|
args.context);
|
2020-04-06 05:19:40 +00:00
|
|
|
},
|
|
|
|
{
|
|
|
|
.source_access_type = AccessType::MYSQL,
|
2017-12-30 00:36:06 +00:00
|
|
|
});
|
|
|
|
}
|
|
|
|
|
2017-12-26 18:32:17 +00:00
|
|
|
}
|
2017-12-29 15:48:20 +00:00
|
|
|
|
|
|
|
#endif
|