thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-21 15:12:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

refine setting with enable_zstd_qat_codec

Browse Source
This commit is contained in:
jinjunzh 2023-12-22 14:24:15 -05:00
parent b1a7cae515
commit aaf98f0ed7
16 changed files with 93 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
docs/en/sql-reference/statements/create/table.md
View File

@ -380,7 +380,8 @@ High compression levels are useful for asymmetric scenarios, like compress once,
`ZSTDQAT[(level)]` — ZSTD QAT (hardware-accelerated compression) implemented by [QAT-ZSTD-Plugin](https://github.com/intel/QAT-ZSTD-Plugin) with configurable level. Default level: 1. Setting `level <= 0` applies the default level. Possible levels: \[1, 12\]. Recommended level range: \[6, 12\].
ZSTDQAT tries to use an Intel® QAT offloading device ([QuickAssist Technology](https://www.intel.com/content/www/us/en/developer/topic-technology/open/quick-assist-technology/overview.html)). If no such device was found, it will fallback to ZSTD compression in software.
- ZSTDQAT is disabled by default and can only be used after setting configuration parameter `enable_zstd_qat_codec = 1`.
- ZSTDQAT tries to use an Intel® QAT offloading device ([QuickAssist Technology](https://www.intel.com/content/www/us/en/developer/topic-technology/open/quick-assist-technology/overview.html)). If no such device was found, it will fallback to ZSTD compression in software.
#### DEFLATE_QPL

8
src/Client/Connection.cpp
View File

@ -651,7 +651,13 @@ void Connection::sendQuery(
if (method == "ZSTD")
level = settings->network_zstd_compression_level;
CompressionCodecFactory::instance().validateCodec(method, level, !settings->allow_suspicious_codecs, settings->allow_experimental_codecs, settings->enable_deflate_qpl_codec);
CompressionCodecFactory::instance().validateCodec(
method,
level,
!settings->allow_suspicious_codecs,
settings->allow_experimental_codecs,
settings->enable_deflate_qpl_codec,
settings->enable_zstd_qat_codec);
compression_codec = CompressionCodecFactory::instance().get(method, level);
}
else

1
src/Compression/CompressionCodecZSTDQAT.cpp
View File

@ -30,6 +30,7 @@ public:
~CompressionCodecZSTDQAT() override;
protected:
bool isZSTDQAT() const override { return true; }
/// TODO: So far, QAT hardware only support compression. For next generation in future, it will support decompression as well.
UInt32 doCompressData(const char * source, UInt32 source_size, char * dest) const override;

4
src/Compression/CompressionFactory.h
View File

@ -40,10 +40,10 @@ public:
CompressionCodecPtr getDefaultCodec() const;
/// Validate codecs AST specified by user and parses codecs description (substitute default parameters)
ASTPtr validateCodecAndGetPreprocessedAST(const ASTPtr & ast, const DataTypePtr & column_type, bool sanity_check, bool allow_experimental_codecs, bool enable_deflate_qpl_codec) const;
ASTPtr validateCodecAndGetPreprocessedAST(const ASTPtr & ast, const DataTypePtr & column_type, bool sanity_check, bool allow_experimental_codecs, bool enable_deflate_qpl_codec, bool enable_zstd_qat_codec) const;
/// Validate codecs AST specified by user
void validateCodec(const String & family_name, std::optional<int> level, bool sanity_check, bool allow_experimental_codecs, bool enable_deflate_qpl_codec) const;
void validateCodec(const String & family_name, std::optional<int> level, bool sanity_check, bool allow_experimental_codecs, bool enable_deflate_qpl_codec, bool enable_zstd_qat_codec) const;
/// Get codec by AST and possible column_type. Some codecs can use
/// information about type to improve inner settings, but every codec should

14
src/Compression/CompressionFactoryAdditions.cpp
View File

@ -34,7 +34,7 @@ namespace ErrorCodes
void CompressionCodecFactory::validateCodec(
const String & family_name, std::optional<int> level, bool sanity_check, bool allow_experimental_codecs, bool enable_deflate_qpl_codec) const
const String & family_name, std::optional<int> level, bool sanity_check, bool allow_experimental_codecs, bool enable_deflate_qpl_codec, bool enable_zstd_qat_codec) const
{
if (family_name.empty())
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Compression codec name cannot be empty");
@ -43,13 +43,13 @@ void CompressionCodecFactory::validateCodec(
{
auto literal = std::make_shared<ASTLiteral>(static_cast<UInt64>(*level));
validateCodecAndGetPreprocessedAST(makeASTFunction("CODEC", makeASTFunction(Poco::toUpper(family_name), literal)),
{}, sanity_check, allow_experimental_codecs, enable_deflate_qpl_codec);
{}, sanity_check, allow_experimental_codecs, enable_deflate_qpl_codec, enable_zstd_qat_codec);
}
else
{
auto identifier = std::make_shared<ASTIdentifier>(Poco::toUpper(family_name));
validateCodecAndGetPreprocessedAST(makeASTFunction("CODEC", identifier),
{}, sanity_check, allow_experimental_codecs, enable_deflate_qpl_codec);
{}, sanity_check, allow_experimental_codecs, enable_deflate_qpl_codec, enable_zstd_qat_codec);
}
}
@ -77,7 +77,7 @@ bool innerDataTypeIsFloat(const DataTypePtr & type)
}
ASTPtr CompressionCodecFactory::validateCodecAndGetPreprocessedAST(
const ASTPtr & ast, const DataTypePtr & column_type, bool sanity_check, bool allow_experimental_codecs, bool enable_deflate_qpl_codec) const
const ASTPtr & ast, const DataTypePtr & column_type, bool sanity_check, bool allow_experimental_codecs, bool enable_deflate_qpl_codec, bool enable_zstd_qat_codec) const
{
if (const auto * func = ast->as<ASTFunction>())
{
@ -165,6 +165,12 @@ ASTPtr CompressionCodecFactory::validateCodecAndGetPreprocessedAST(
" You can enable it with the 'enable_deflate_qpl_codec' setting.",
codec_family_name);
if (!enable_zstd_qat_codec && result_codec->isZSTDQAT())
throw Exception(ErrorCodes::BAD_ARGUMENTS,
"Codec {} is disabled by default."
" You can enable it with the 'enable_zstd_qat_codec' setting.",
codec_family_name);
codecs_descriptions->children.emplace_back(result_codec->getCodecDesc());
}

3
src/Compression/ICompressionCodec.h
View File

@ -121,6 +121,9 @@ public:
/// Is this the DEFLATE_QPL codec?
virtual bool isDeflateQpl() const { return false; }
/// Is this the ZSTDQAT codec?
virtual bool isZSTDQAT() const { return false; }
/// If it does nothing.
virtual bool isNone() const { return false; }

1
src/Core/Settings.h
View File

@ -348,6 +348,7 @@ class IColumn;
M(Bool, allow_suspicious_codecs, false, "If it is set to true, allow to specify meaningless compression codecs.", 0) \
M(Bool, allow_experimental_codecs, false, "If it is set to true, allow to specify experimental compression codecs (but we don't have those yet and this option does nothing).", 0) \
M(Bool, enable_deflate_qpl_codec, false, "Enable/disable the DEFLATE_QPL codec.", 0) \
M(Bool, enable_zstd_qat_codec, false, "Enable/disable the ZSTDQAT codec.", 0) \
M(UInt64, query_profiler_real_time_period_ns, QUERY_PROFILER_DEFAULT_SAMPLE_RATE_NS, "Period for real clock timer of query profiler (in nanoseconds). Set 0 value to turn off the real clock query profiler. Recommended value is at least 10000000 (100 times a second) for single queries or 1000000000 (once a second) for cluster-wide profiling.", 0) \
M(UInt64, query_profiler_cpu_time_period_ns, QUERY_PROFILER_DEFAULT_SAMPLE_RATE_NS, "Period for CPU clock timer of query profiler (in nanoseconds). Set 0 value to turn off the CPU clock query profiler. Recommended value is at least 10000000 (100 times a second) for single queries or 1000000000 (once a second) for cluster-wide profiling.", 0) \
M(Bool, metrics_perf_events_enabled, false, "If enabled, some of the perf events will be measured throughout queries' execution.", 0) \

3
src/Interpreters/InterpreterCreateQuery.cpp
View File

@ -595,6 +595,7 @@ ColumnsDescription InterpreterCreateQuery::getColumnsDescription(
bool sanity_check_compression_codecs = !attach && !context_->getSettingsRef().allow_suspicious_codecs;
bool allow_experimental_codecs = attach || context_->getSettingsRef().allow_experimental_codecs;
bool enable_deflate_qpl_codec = attach || context_->getSettingsRef().enable_deflate_qpl_codec;
bool enable_zstd_qat_codec = attach || context_->getSettingsRef().enable_zstd_qat_codec;
ColumnsDescription res;
auto name_type_it = column_names_and_types.begin();
@ -655,7 +656,7 @@ ColumnsDescription InterpreterCreateQuery::getColumnsDescription(
if (col_decl.default_specifier == "ALIAS")
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot specify codec for column type ALIAS");
column.codec = CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(
col_decl.codec, column.type, sanity_check_compression_codecs, allow_experimental_codecs, enable_deflate_qpl_codec);
col_decl.codec, column.type, sanity_check_compression_codecs, allow_experimental_codecs, enable_deflate_qpl_codec, enable_zstd_qat_codec);
}
if (col_decl.stat_type)

2
src/Server/TCPHandler.cpp
View File

@ -2023,7 +2023,7 @@ void TCPHandler::initBlockOutput(const Block & block)
if (state.compression == Protocol::Compression::Enable)
{
CompressionCodecFactory::instance().validateCodec(method, level, !query_settings.allow_suspicious_codecs, query_settings.allow_experimental_codecs, query_settings.enable_deflate_qpl_codec);
CompressionCodecFactory::instance().validateCodec(method, level, !query_settings.allow_suspicious_codecs, query_settings.allow_experimental_codecs, query_settings.enable_deflate_qpl_codec, query_settings.enable_zstd_qat_codec);
state.maybe_compressed_out = std::make_shared<CompressedWriteBuffer>(
*out, CompressionCodecFactory::instance().get(method, level));

8
src/Storages/AlterCommands.cpp
View File

@ -431,7 +431,7 @@ void AlterCommand::apply(StorageInMemoryMetadata & metadata, ContextPtr context)
column.comment = *comment;
if (codec)
column.codec = CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(codec, data_type, false, true, true);
column.codec = CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(codec, data_type, false, true, true, true);
column.ttl = ttl;
@ -496,7 +496,7 @@ void AlterCommand::apply(StorageInMemoryMetadata & metadata, ContextPtr context)
else
{
if (codec)
column.codec = CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(codec, data_type ? data_type : column.type, false, true, true);
column.codec = CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(codec, data_type ? data_type : column.type, false, true, true, true);
if (comment)
column.comment = *comment;
@ -1237,7 +1237,7 @@ void AlterCommands::validate(const StoragePtr & table, ContextPtr context) const
"this column name is reserved for _block_number persisting feature", backQuote(column_name));
if (command.codec)
CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(command.codec, command.data_type, !context->getSettingsRef().allow_suspicious_codecs, context->getSettingsRef().allow_experimental_codecs, context->getSettingsRef().enable_deflate_qpl_codec);
CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(command.codec, command.data_type, !context->getSettingsRef().allow_suspicious_codecs, context->getSettingsRef().allow_experimental_codecs, context->getSettingsRef().enable_deflate_qpl_codec, context->getSettingsRef().enable_zstd_qat_codec);
all_columns.add(ColumnDescription(column_name, command.data_type));
}
@ -1262,7 +1262,7 @@ void AlterCommands::validate(const StoragePtr & table, ContextPtr context) const
{
if (all_columns.hasAlias(column_name))
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot specify codec for column type ALIAS");
CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(command.codec, command.data_type, !context->getSettingsRef().allow_suspicious_codecs, context->getSettingsRef().allow_experimental_codecs, context->getSettingsRef().enable_deflate_qpl_codec);
CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(command.codec, command.data_type, !context->getSettingsRef().allow_suspicious_codecs, context->getSettingsRef().allow_experimental_codecs, context->getSettingsRef().enable_deflate_qpl_codec, context->getSettingsRef().enable_zstd_qat_codec);
}
auto column_default = all_columns.getDefault(column_name);
if (column_default)

2
src/Storages/ColumnsDescription.cpp
View File

@ -151,7 +151,7 @@ void ColumnDescription::readText(ReadBuffer & buf)
comment = col_ast->comment->as<ASTLiteral &>().value.get<String>();
if (col_ast->codec)
codec = CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(col_ast->codec, type, false, true, true);
codec = CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(col_ast->codec, type, false, true, true, true);
if (col_ast->ttl)
ttl = col_ast->ttl;

2
src/Storages/Distributed/DistributedSink.cpp
View File

@ -740,7 +740,7 @@ void DistributedSink::writeToShard(const Cluster::ShardInfo & shard_info, const
if (compression_method == "ZSTD")
compression_level = settings.network_zstd_compression_level;
CompressionCodecFactory::instance().validateCodec(compression_method, compression_level, !settings.allow_suspicious_codecs, settings.allow_experimental_codecs, settings.enable_deflate_qpl_codec);
CompressionCodecFactory::instance().validateCodec(compression_method, compression_level, !settings.allow_suspicious_codecs, settings.allow_experimental_codecs, settings.enable_deflate_qpl_codec, settings.enable_zstd_qat_codec);
CompressionCodecPtr compression_codec = CompressionCodecFactory::instance().get(compression_method, compression_level);
/// tmp directory is used to ensure atomicity of transactions

2
src/Storages/TTLDescription.cpp
View File

@ -285,7 +285,7 @@ TTLDescription TTLDescription::getTTLFromAST(
{
result.recompression_codec =
CompressionCodecFactory::instance().validateCodecAndGetPreprocessedAST(
ttl_element->recompression_codec, {}, !context->getSettingsRef().allow_suspicious_codecs, context->getSettingsRef().allow_experimental_codecs, context->getSettingsRef().enable_deflate_qpl_codec);
ttl_element->recompression_codec, {}, !context->getSettingsRef().allow_suspicious_codecs, context->getSettingsRef().allow_experimental_codecs, context->getSettingsRef().enable_deflate_qpl_codec, context->getSettingsRef().enable_zstd_qat_codec);
}
}

1
tests/ci/stress.py
View File

@ -21,6 +21,7 @@ def get_options(i: int, upgrade_check: bool) -> str:
options.append(f'''--db-engine="Replicated('/test/db/test_{i}', 's1', 'r1')"''')
client_options.append("allow_experimental_database_replicated=1")
client_options.append("enable_deflate_qpl_codec=1")
client_options.append("enable_zstd_qat_codec=1")
# If database name is not specified, new database is created for each functional test.
# Run some threads with one database for all tests.

7
tests/integration/test_non_default_compression/configs/enable_zstdqat_codec.xml Normal file
View File

@ -0,0 +1,7 @@
<clickhouse>
<profiles>
<default>
<enable_zstd_qat_codec>1</enable_zstd_qat_codec>
</default>
</profiles>
</clickhouse>

49
tests/queries/0_stateless/00804_test_zstdqat_codec_compression.sql Normal file
View File

@ -0,0 +1,49 @@
--Tags: no-fasttest, no-cpu-aarch64, no-cpu-s390x
-- no-fasttest because ZSTDQAT isn't available in fasttest
-- no-cpu-aarch64 and no-cpu-s390x because ZSTDQAT is x86-only
-- A bunch of random DDLs to test the ZSTDQAT codec.
SET enable_zstd_qat_codec = 1;
-- Suppress test failures because stderr contains warning "Initialization of hardware-assisted DeflateQpl failed, falling
-- back to software DeflateQpl coded."
SET send_logs_level = 'fatal';
DROP TABLE IF EXISTS compression_codec;
CREATE TABLE compression_codec(
id UInt64 CODEC(ZSTDQAT),
data String CODEC(ZSTDQAT),
ddd Date CODEC(ZSTDQAT),
ddd32 Date32 CODEC(ZSTDQAT),
somenum Float64 CODEC(ZSTDQAT),
somestr FixedString(3) CODEC(ZSTDQAT),
othernum Int64 CODEC(ZSTDQAT),
somearray Array(UInt8) CODEC(ZSTDQAT),
somemap Map(String, UInt32) CODEC(ZSTDQAT),
sometuple Tuple(UInt16, UInt64) CODEC(ZSTDQAT),
) ENGINE = MergeTree() ORDER BY tuple();
SHOW CREATE TABLE compression_codec;
INSERT INTO compression_codec VALUES(1, 'hello', toDate('2018-12-14'), toDate32('2018-12-14'), 1.1, 'aaa', 5, [1,2,3], map('k1',1,'k2',2), tuple(1,2));
INSERT INTO compression_codec VALUES(2, 'world', toDate('2018-12-15'), toDate32('2018-12-15'), 2.2, 'bbb', 6, [4,5,6], map('k3',3,'k4',4), tuple(3,4));
INSERT INTO compression_codec VALUES(3, '!', toDate('2018-12-16'), toDate32('2018-12-16'), 3.3, 'ccc', 7, [7,8,9], map('k5',5,'k6',6), tuple(5,6));
SELECT * FROM compression_codec ORDER BY id;
OPTIMIZE TABLE compression_codec FINAL;
INSERT INTO compression_codec VALUES(2, '', toDate('2018-12-13'), toDate32('2018-12-13'), 4.4, 'ddd', 8, [10,11,12], map('k7',7,'k8',8), tuple(7,8));
DETACH TABLE compression_codec;
ATTACH TABLE compression_codec;
SELECT count(*) FROM compression_codec WHERE id = 2 GROUP BY id;
INSERT INTO compression_codec SELECT 3, '!', toDate('2018-12-16'), toDate32('2018-12-16'), 3.3, 'ccc', 7, [7,8,9], map('k5',5,'k6',6), tuple(5,6) FROM system.numbers LIMIT 10000;
SELECT count(*) FROM compression_codec WHERE id = 3 GROUP BY id;
DROP TABLE IF EXISTS compression_codec;