ClickHouse/src/Storages/StoragePostgreSQL.cpp

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

537 lines
20 KiB
C++
Raw Normal View History

#include "StoragePostgreSQL.h"
#if USE_LIBPQXX
#include <Processors/Sources/PostgreSQLSource.h>
2021-09-01 17:59:11 +00:00
#include <Common/parseAddress.h>
#include <Common/assert_cast.h>
#include <Common/parseRemoteDescription.h>
2022-04-27 15:05:45 +00:00
#include <Common/logger_useful.h>
#include <Common/NamedCollections/NamedCollections.h>
#include <Core/Settings.h>
#include <Core/PostgreSQL/PoolWithFailover.h>
#include <DataTypes/DataTypeString.h>
#include <DataTypes/DataTypeArray.h>
#include <DataTypes/DataTypesDecimal.h>
#include <DataTypes/DataTypeNullable.h>
2021-09-01 17:59:11 +00:00
#include <Columns/ColumnString.h>
#include <Columns/ColumnArray.h>
#include <Columns/ColumnsNumber.h>
#include <Columns/ColumnDecimal.h>
2020-11-25 14:52:11 +00:00
#include <Columns/ColumnNullable.h>
2021-09-01 17:59:11 +00:00
2020-11-25 14:52:11 +00:00
#include <Formats/FormatSettings.h>
2021-09-01 17:59:11 +00:00
2020-11-25 21:06:33 +00:00
#include <IO/WriteHelpers.h>
2021-08-24 12:29:42 +00:00
#include <IO/Operators.h>
2021-09-01 17:59:11 +00:00
#include <Interpreters/evaluateConstantExpression.h>
#include <Interpreters/Context.h>
#include <Parsers/getInsertQuery.h>
#include <Parsers/ASTFunction.h>
#include <Processors/Sinks/SinkToStorage.h>
2022-05-20 19:49:31 +00:00
#include <QueryPipeline/Pipe.h>
2021-09-01 17:59:11 +00:00
#include <Storages/StorageFactory.h>
#include <Storages/transformQueryForExternalDatabase.h>
#include <Storages/checkAndGetLiteralArgument.h>
#include <Storages/NamedCollectionsHelpers.h>
2021-09-01 17:59:11 +00:00
namespace DB
{
namespace ErrorCodes
{
extern const int NOT_IMPLEMENTED;
2021-09-04 18:46:09 +00:00
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
}
StoragePostgreSQL::StoragePostgreSQL(
const StorageID & table_id_,
2021-05-07 11:18:49 +00:00
postgres::PoolWithFailoverPtr pool_,
const String & remote_table_name_,
const ColumnsDescription & columns_,
const ConstraintsDescription & constraints_,
2021-04-23 12:18:23 +00:00
const String & comment,
2021-08-24 12:29:42 +00:00
const String & remote_table_schema_,
const String & on_conflict_)
: IStorage(table_id_)
, remote_table_name(remote_table_name_)
, remote_table_schema(remote_table_schema_)
2021-08-24 12:29:42 +00:00
, on_conflict(on_conflict_)
2021-05-07 11:18:49 +00:00
, pool(std::move(pool_))
, log(&Poco::Logger::get("StoragePostgreSQL (" + table_id_.table_name + ")"))
{
StorageInMemoryMetadata storage_metadata;
storage_metadata.setColumns(columns_);
storage_metadata.setConstraints(constraints_);
2021-04-23 12:18:23 +00:00
storage_metadata.setComment(comment);
setInMemoryMetadata(storage_metadata);
}
Pipe StoragePostgreSQL::read(
const Names & column_names_,
const StorageSnapshotPtr & storage_snapshot,
SelectQueryInfo & query_info_,
ContextPtr context_,
QueryProcessingStage::Enum /*processed_stage*/,
size_t max_block_size_,
size_t /*num_streams*/)
{
storage_snapshot->check(column_names_);
/// Connection is already made to the needed database, so it should not be present in the query;
/// remote_table_schema is empty if it is not specified, will access only table_name.
String query = transformQueryForExternalDatabase(
query_info_,
column_names_,
storage_snapshot->metadata->getColumns().getOrdinary(),
IdentifierQuotingStyle::DoubleQuotes, remote_table_schema, remote_table_name, context_);
LOG_TRACE(log, "Query: {}", query);
Block sample_block;
for (const String & column_name : column_names_)
{
auto column_data = storage_snapshot->metadata->getColumns().getPhysical(column_name);
WhichDataType which(column_data.type);
if (which.isEnum())
column_data.type = std::make_shared<DataTypeString>();
sample_block.insert({ column_data.type, column_data.name });
}
return Pipe(std::make_shared<PostgreSQLSource<>>(pool->get(), query, sample_block, max_block_size_));
}
2021-07-23 19:33:59 +00:00
class PostgreSQLSink : public SinkToStorage
2020-11-25 14:52:11 +00:00
{
2021-08-24 12:29:42 +00:00
using Row = std::vector<std::optional<std::string>>;
public:
2021-07-23 19:33:59 +00:00
explicit PostgreSQLSink(
const StorageMetadataPtr & metadata_snapshot_,
2021-05-07 11:18:49 +00:00
postgres::ConnectionHolderPtr connection_holder_,
2021-05-22 08:47:19 +00:00
const String & remote_table_name_,
2021-08-24 12:29:42 +00:00
const String & remote_table_schema_,
const String & on_conflict_)
2021-07-26 10:08:40 +00:00
: SinkToStorage(metadata_snapshot_->getSampleBlock())
2021-07-23 19:33:59 +00:00
, metadata_snapshot(metadata_snapshot_)
2021-05-07 11:18:49 +00:00
, connection_holder(std::move(connection_holder_))
, remote_table_name(remote_table_name_)
2021-05-22 08:47:19 +00:00
, remote_table_schema(remote_table_schema_)
2021-08-24 12:29:42 +00:00
, on_conflict(on_conflict_)
{
}
2020-11-25 14:52:11 +00:00
2021-07-23 19:33:59 +00:00
String getName() const override { return "PostgreSQLSink"; }
2020-11-25 14:52:11 +00:00
2021-07-23 19:33:59 +00:00
void consume(Chunk chunk) override
{
auto block = getHeader().cloneWithColumns(chunk.detachColumns());
2021-05-22 08:47:19 +00:00
if (!inserter)
2021-08-24 12:29:42 +00:00
{
if (on_conflict.empty())
{
inserter = std::make_unique<StreamTo>(connection_holder->get(),
remote_table_schema.empty() ? pqxx::table_path({remote_table_name})
: pqxx::table_path({remote_table_schema, remote_table_name}), block.getNames());
}
else
{
inserter = std::make_unique<PreparedInsert>(connection_holder->get(), remote_table_name,
remote_table_schema, block.getColumnsWithTypeAndName(), on_conflict);
}
}
2021-01-10 13:01:15 +00:00
const auto columns = block.getColumns();
const size_t num_rows = block.rows(), num_cols = block.columns();
const auto data_types = block.getDataTypes();
2020-11-25 14:52:11 +00:00
/// std::optional lets libpqxx to know if value is NULL
std::vector<std::optional<std::string>> row(num_cols);
2020-11-25 14:52:11 +00:00
2021-06-15 19:55:21 +00:00
for (const auto i : collections::range(0, num_rows))
{
2021-06-15 19:55:21 +00:00
for (const auto j : collections::range(0, num_cols))
{
if (columns[j]->isNullAt(i))
{
row[j] = std::nullopt;
}
else
{
WriteBufferFromOwnString ostr;
if (isArray(data_types[j]))
{
parseArray((*columns[j])[i], data_types[j], ostr);
}
else
{
2021-03-09 14:46:52 +00:00
data_types[j]->getDefaultSerialization()->serializeText(*columns[j], i, ostr, FormatSettings{});
}
row[j] = ostr.str();
}
}
2021-01-10 10:03:01 +00:00
2021-08-24 12:29:42 +00:00
inserter->insert(row);
}
}
2021-07-23 19:33:59 +00:00
void onFinish() override
{
2021-05-22 08:47:19 +00:00
if (inserter)
inserter->complete();
}
2020-11-25 14:52:11 +00:00
/// Cannot just use serializeAsText for array data type even though it converts perfectly
2021-06-28 17:02:22 +00:00
/// any dimension number array into text format, because it encloses in '[]' and for postgres it must be '{}'.
2021-01-16 08:43:59 +00:00
/// Check if array[...] syntax from PostgreSQL will be applicable.
2023-05-17 13:13:57 +00:00
static void parseArray(const Field & array_field, const DataTypePtr & data_type, WriteBuffer & ostr)
2020-11-25 14:52:11 +00:00
{
const auto * array_type = typeid_cast<const DataTypeArray *>(data_type.get());
2021-01-10 10:03:01 +00:00
const auto & nested = array_type->getNestedType();
const auto & array = array_field.get<Array>();
if (!isArray(nested))
2020-11-25 14:52:11 +00:00
{
2023-05-17 13:13:57 +00:00
parseArrayContent(array, data_type, ostr);
return;
}
writeChar('{', ostr);
const auto * nested_array_type = typeid_cast<const DataTypeArray *>(nested.get());
for (auto iter = array.begin(); iter != array.end(); ++iter)
{
if (iter != array.begin())
writeText(", ", ostr);
if (!isArray(nested_array_type->getNestedType()))
2020-11-25 14:52:11 +00:00
{
2023-05-17 13:13:57 +00:00
parseArrayContent(iter->get<Array>(), nested, ostr);
2020-11-25 14:52:11 +00:00
}
else
{
parseArray(*iter, nested, ostr);
2020-11-25 14:52:11 +00:00
}
}
writeChar('}', ostr);
2020-11-25 14:52:11 +00:00
}
/// Conversion is done via column casting because with writeText(Array..) got incorrect conversion
/// of Date and DateTime data types and it added extra quotes for values inside array.
2023-05-17 13:13:57 +00:00
static void parseArrayContent(const Array & array_field, const DataTypePtr & data_type, WriteBuffer & ostr)
{
2023-05-17 13:13:57 +00:00
auto nested_type = typeid_cast<const DataTypeArray *>(data_type.get())->getNestedType();
auto array_column = ColumnArray::create(createNested(nested_type));
2021-01-15 20:09:31 +00:00
array_column->insert(array_field);
2023-05-17 13:13:57 +00:00
const IColumn & nested_column = array_column->getData();
const auto serialization = nested_type->getDefaultSerialization();
FormatSettings settings;
settings.pretty.charset = FormatSettings::Pretty::Charset::ASCII;
if (nested_type->isNullable())
nested_type = static_cast<const DataTypeNullable *>(nested_type.get())->getNestedType();
const bool quoted = !isUUID(nested_type);
writeChar('{', ostr);
for (size_t i = 0, size = array_field.size(); i < size; ++i)
{
if (i != 0)
writeChar(',', ostr);
if (quoted)
serialization->serializeTextQuoted(nested_column, i, ostr, settings);
else
serialization->serializeText(nested_column, i, ostr, settings);
}
writeChar('}', ostr);
}
2021-01-15 20:09:31 +00:00
static MutableColumnPtr createNested(DataTypePtr nested)
{
bool is_nullable = false;
if (nested->isNullable())
{
is_nullable = true;
2021-01-10 15:33:35 +00:00
nested = static_cast<const DataTypeNullable *>(nested.get())->getNestedType();
}
WhichDataType which(nested);
2021-01-15 20:09:31 +00:00
MutableColumnPtr nested_column;
if (which.isString() || which.isFixedString()) nested_column = ColumnString::create();
else if (which.isInt8() || which.isInt16()) nested_column = ColumnInt16::create();
else if (which.isUInt8() || which.isUInt16()) nested_column = ColumnUInt16::create();
else if (which.isInt32()) nested_column = ColumnInt32::create();
else if (which.isUInt32()) nested_column = ColumnUInt32::create();
else if (which.isInt64()) nested_column = ColumnInt64::create();
else if (which.isUInt64()) nested_column = ColumnUInt64::create();
else if (which.isFloat32()) nested_column = ColumnFloat32::create();
else if (which.isFloat64()) nested_column = ColumnFloat64::create();
else if (which.isDate()) nested_column = ColumnUInt16::create();
else if (which.isDateTime()) nested_column = ColumnUInt32::create();
2023-05-15 21:40:10 +00:00
else if (which.isUUID()) nested_column = ColumnUUID::create();
else if (which.isDateTime64())
{
nested_column = ColumnDecimal<DateTime64>::create(0, 6);
}
else if (which.isDecimal32())
{
const auto & type = typeid_cast<const DataTypeDecimal<Decimal32> *>(nested.get());
nested_column = ColumnDecimal<Decimal32>::create(0, type->getScale());
}
else if (which.isDecimal64())
{
const auto & type = typeid_cast<const DataTypeDecimal<Decimal64> *>(nested.get());
nested_column = ColumnDecimal<Decimal64>::create(0, type->getScale());
}
else if (which.isDecimal128())
{
const auto & type = typeid_cast<const DataTypeDecimal<Decimal128> *>(nested.get());
nested_column = ColumnDecimal<Decimal128>::create(0, type->getScale());
}
2021-01-10 10:03:01 +00:00
else if (which.isDecimal256())
{
const auto & type = typeid_cast<const DataTypeDecimal<Decimal256> *>(nested.get());
nested_column = ColumnDecimal<Decimal256>::create(0, type->getScale());
}
else
throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Type conversion not supported");
if (is_nullable)
2021-01-15 20:09:31 +00:00
return ColumnNullable::create(std::move(nested_column), ColumnUInt8::create(nested_column->size(), 0));
return nested_column;
}
private:
2021-08-24 12:29:42 +00:00
struct Inserter
2021-05-22 08:47:19 +00:00
{
2021-08-24 12:29:42 +00:00
pqxx::connection & connection;
2021-05-22 08:47:19 +00:00
pqxx::work tx;
2021-08-24 12:29:42 +00:00
2021-08-25 21:51:43 +00:00
explicit Inserter(pqxx::connection & connection_)
2021-08-24 12:29:42 +00:00
: connection(connection_)
, tx(connection) {}
virtual ~Inserter() = default;
virtual void insert(const Row & row) = 0;
virtual void complete() = 0;
};
struct StreamTo : Inserter
{
2021-06-09 07:46:49 +00:00
Names columns;
2021-05-22 08:47:19 +00:00
pqxx::stream_to stream;
2021-08-24 12:29:42 +00:00
StreamTo(pqxx::connection & connection_, pqxx::table_path table_, Names columns_)
: Inserter(connection_)
2021-06-09 07:46:49 +00:00
, columns(std::move(columns_))
2021-06-16 09:11:31 +00:00
, stream(pqxx::stream_to::raw_table(tx, connection.quote_table(table_), connection.quote_columns(columns)))
2021-06-09 07:46:49 +00:00
{
}
2021-05-22 08:47:19 +00:00
2021-08-24 12:29:42 +00:00
void complete() override
2021-05-22 08:47:19 +00:00
{
stream.complete();
tx.commit();
}
2021-08-24 12:29:42 +00:00
void insert(const Row & row) override
{
stream.write_values(row);
}
};
struct PreparedInsert : Inserter
{
PreparedInsert(pqxx::connection & connection_, const String & table, const String & schema,
const ColumnsWithTypeAndName & columns, const String & on_conflict_)
: Inserter(connection_)
, statement_name("insert_" + getHexUIntLowercase(thread_local_rng()))
2021-08-24 12:29:42 +00:00
{
WriteBufferFromOwnString buf;
buf << getInsertQuery(schema, table, columns, IdentifierQuotingStyle::DoubleQuotes);
buf << " (";
for (size_t i = 1; i <= columns.size(); ++i)
{
if (i > 1)
buf << ", ";
buf << "$" << i;
}
buf << ") ";
buf << on_conflict_;
connection.prepare(statement_name, buf.str());
prepared = true;
2021-08-24 12:29:42 +00:00
}
void complete() override
{
connection.unprepare(statement_name);
prepared = false;
2021-08-24 12:29:42 +00:00
tx.commit();
}
void insert(const Row & row) override
{
pqxx::params params;
params.reserve(row.size());
params.append_multi(row);
tx.exec_prepared(statement_name, params);
2021-08-24 12:29:42 +00:00
}
~PreparedInsert() override
{
try
{
if (prepared)
connection.unprepare(statement_name);
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__);
}
}
const String statement_name;
bool prepared = false;
2021-05-22 08:47:19 +00:00
};
StorageMetadataPtr metadata_snapshot;
2021-05-07 11:18:49 +00:00
postgres::ConnectionHolderPtr connection_holder;
2021-08-24 12:29:42 +00:00
const String remote_db_name, remote_table_name, remote_table_schema, on_conflict;
std::unique_ptr<Inserter> inserter;
};
2020-11-25 14:52:11 +00:00
2021-07-23 19:33:59 +00:00
SinkToStoragePtr StoragePostgreSQL::write(
const ASTPtr & /*query*/, const StorageMetadataPtr & metadata_snapshot, ContextPtr /* context */)
2020-11-25 14:52:11 +00:00
{
2021-08-24 12:29:42 +00:00
return std::make_shared<PostgreSQLSink>(metadata_snapshot, pool->get(), remote_table_name, remote_table_schema, on_conflict);
2020-11-25 14:52:11 +00:00
}
2023-02-20 20:37:38 +00:00
StoragePostgreSQL::Configuration StoragePostgreSQL::processNamedCollectionResult(const NamedCollection & named_collection, bool require_table)
{
StoragePostgreSQL::Configuration configuration;
2023-03-01 16:00:10 +00:00
ValidateKeysMultiset<ExternalDatabaseEqualKeysSet> required_arguments = {"user", "username", "password", "database", "db"};
2023-02-20 20:37:38 +00:00
if (require_table)
required_arguments.insert("table");
2023-02-24 19:38:40 +00:00
validateNamedCollection<ValidateKeysMultiset<ExternalDatabaseEqualKeysSet>>(
2023-05-03 23:40:12 +00:00
named_collection, required_arguments, {"schema", "on_conflict", "addresses_expr", "host", "hostname", "port", "use_table_cache"});
2023-02-20 20:37:38 +00:00
configuration.addresses_expr = named_collection.getOrDefault<String>("addresses_expr", "");
if (configuration.addresses_expr.empty())
{
2023-02-24 19:57:37 +00:00
configuration.host = named_collection.getAny<String>({"host", "hostname"});
2023-02-20 20:37:38 +00:00
configuration.port = static_cast<UInt16>(named_collection.get<UInt64>("port"));
configuration.addresses = {std::make_pair(configuration.host, configuration.port)};
}
2023-02-24 19:57:37 +00:00
configuration.username = named_collection.getAny<String>({"username", "user"});
2023-02-20 20:37:38 +00:00
configuration.password = named_collection.get<String>("password");
2023-02-24 19:57:37 +00:00
configuration.database = named_collection.getAny<String>({"db", "database"});
2023-02-20 20:37:38 +00:00
if (require_table)
configuration.table = named_collection.get<String>("table");
configuration.schema = named_collection.getOrDefault<String>("schema", "");
configuration.on_conflict = named_collection.getOrDefault<String>("on_conflict", "");
return configuration;
}
2020-11-25 14:52:11 +00:00
StoragePostgreSQL::Configuration StoragePostgreSQL::getConfiguration(ASTs engine_args, ContextPtr context)
{
StoragePostgreSQL::Configuration configuration;
2023-03-05 11:50:29 +00:00
if (auto named_collection = tryGetNamedCollectionWithOverrides(engine_args, context))
{
2023-02-20 20:37:38 +00:00
configuration = StoragePostgreSQL::processNamedCollectionResult(*named_collection);
2021-09-01 17:59:11 +00:00
}
else
{
2021-08-24 12:29:42 +00:00
if (engine_args.size() < 5 || engine_args.size() > 7)
2022-12-21 23:27:22 +00:00
{
throw Exception(
ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH,
"Storage PostgreSQL requires from 5 to 7 parameters: "
"PostgreSQL('host:port', 'database', 'table', 'username', 'password' "
"[, 'schema', 'ON CONFLICT ...']. Got: {}",
engine_args.size());
}
for (auto & engine_arg : engine_args)
2021-09-01 17:59:11 +00:00
engine_arg = evaluateConstantExpressionOrIdentifierAsLiteral(engine_arg, context);
2023-03-05 11:50:29 +00:00
configuration.addresses_expr = checkAndGetLiteralArgument<String>(engine_args[0], "host:port");
2021-09-01 17:59:11 +00:00
size_t max_addresses = context->getSettingsRef().glob_expansion_max_elements;
2021-03-27 19:18:05 +00:00
2023-03-05 11:50:29 +00:00
configuration.addresses = parseRemoteDescriptionForExternalDatabase(configuration.addresses_expr, max_addresses, 5432);
2021-09-03 11:16:32 +00:00
if (configuration.addresses.size() == 1)
{
configuration.host = configuration.addresses[0].first;
configuration.port = configuration.addresses[0].second;
}
configuration.database = checkAndGetLiteralArgument<String>(engine_args[1], "database");
configuration.table = checkAndGetLiteralArgument<String>(engine_args[2], "table");
2022-12-20 21:33:54 +00:00
configuration.username = checkAndGetLiteralArgument<String>(engine_args[3], "user");
configuration.password = checkAndGetLiteralArgument<String>(engine_args[4], "password");
2021-08-24 12:29:42 +00:00
if (engine_args.size() >= 6)
configuration.schema = checkAndGetLiteralArgument<String>(engine_args[5], "schema");
2021-08-24 12:29:42 +00:00
if (engine_args.size() >= 7)
configuration.on_conflict = checkAndGetLiteralArgument<String>(engine_args[6], "on_conflict");
2021-09-01 17:59:11 +00:00
}
for (const auto & address : configuration.addresses)
2022-03-11 15:33:23 +00:00
context->getRemoteHostFilter().checkHostAndPort(address.first, toString(address.second));
2021-09-01 17:59:11 +00:00
return configuration;
}
2021-09-01 17:59:11 +00:00
void registerStoragePostgreSQL(StorageFactory & factory)
{
factory.registerStorage("PostgreSQL", [](const StorageFactory::Arguments & args)
{
2021-09-15 18:11:49 +00:00
auto configuration = StoragePostgreSQL::getConfiguration(args.engine_args, args.getLocalContext());
2022-06-23 21:44:59 +00:00
const auto & settings = args.getContext()->getSettingsRef();
2021-09-01 23:17:15 +00:00
auto pool = std::make_shared<postgres::PoolWithFailover>(configuration,
2022-06-23 21:44:59 +00:00
settings.postgresql_connection_pool_size,
settings.postgresql_connection_pool_wait_timeout,
POSTGRESQL_POOL_WITH_FAILOVER_DEFAULT_MAX_TRIES,
settings.postgresql_connection_pool_auto_close_connection);
2021-01-15 21:18:51 +00:00
return std::make_shared<StoragePostgreSQL>(
2021-05-11 16:22:24 +00:00
args.table_id,
std::move(pool),
2021-09-01 17:59:11 +00:00
configuration.table,
2021-05-11 16:22:24 +00:00
args.columns,
args.constraints,
args.comment,
2021-09-01 17:59:11 +00:00
configuration.schema,
configuration.on_conflict);
},
{
.source_access_type = AccessType::POSTGRES,
});
}
}
#endif