compression_algorithm_zstd.h

/*
 * Copyright (c) 2019, 2025, Oracle and/or its affiliates.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License, version 2.0,
 * as published by the Free Software Foundation.
 *
 * This program is designed to work with certain software (including
 * but not limited to OpenSSL) that is licensed under separate terms,
 * as designated in a particular file or component or in included license
 * documentation. The authors of MySQL hereby grant you an additional
 * permission to link the program and your derivative works with the
 * separately licensed software that they have either included with
 * the program or referenced in the documentation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License, version 2.0, for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef PLUGIN_X_PROTOCOL_STREAM_COMPRESSION_COMPRESSION_ALGORITHM_ZSTD_H_
#definePLUGIN_X_PROTOCOL_STREAM_COMPRESSION_COMPRESSION_ALGORITHM_ZSTD_H_

#include<cstdint>
#include<memory>
#include<utility>

#include"my_compiler.h"
#include"my_dbug.h"// NOLINT(build/include_subdir)
#include"zstd.h"// NOLINT(build/include_subdir)

#include"plugin/x/protocol/stream/compression/compression_algorithm_interface.h"

namespaceprotocol {

classCompression_algorithm_zstd : publicCompression_algorithm_interface {
public:
explicitCompression_algorithm_zstd(constint32_t level)
 : m_stream{ZSTD_createCStream()} {
#if ZSTD_VERSION_NUMBER < 10400
is_error(ZSTD_initCStream(m_stream, level));
#else
if (is_error(ZSTD_CCtx_reset(m_stream, ZSTD_reset_session_only)) ||
is_error(ZSTD_CCtx_refCDict(m_stream, nullptr)) ||
is_error(
ZSTD_CCtx_setParameter(m_stream, ZSTD_c_compressionLevel, level)))
return;
#endif
 }

~Compression_algorithm_zstd() override { ZSTD_freeCStream(m_stream); }

voidset_pledged_source_size(constint src_size [[maybe_unused]]) override {
 DBUG_TRACE;
DBUG_LOG("debug", "set_pledged_source_size(" << src_size << ")");
#if ZSTD_VERSION_NUMBER < 10400
// is_error(ZSTD_resetCStream(m_stream, src_size));
#else
if (!is_error(ZSTD_CCtx_reset(m_stream, ZSTD_reset_session_only)))
is_error(ZSTD_CCtx_setPledgedSrcSize(m_stream, src_size));
#endif
 }

voidset_input(uint8_t *in_ptr, constint in_size) override {
 DBUG_TRACE;
 m_in_buffer = ZSTD_inBuffer{in_ptr, static_cast<size_t>(in_size), 0};
 m_flush_finished = false;
 }

boolcompress(uint8_t *out_ptr, int *out_size) override {
 DBUG_TRACE;
if (m_error) returnfalse;
 ZSTD_outBuffer out_buffer{out_ptr, static_cast<size_t>(*out_size), 0};
while (m_in_buffer.pos < m_in_buffer.size) {
#if ZSTD_VERSION_NUMBER < 10400
if (is_error(ZSTD_compressStream(m_stream, &out_buffer, &m_in_buffer))) {
#else
if (is_error(ZSTD_compressStream2(m_stream, &out_buffer, &m_in_buffer,
 ZSTD_e_continue))) {
#endif
 *out_size = 0;
returnfalse;
 }
 }
DBUG_LOG("debug", "zstandard(in_pos:" << m_in_buffer.pos << ")");
 *out_size = static_cast<int>(out_buffer.pos);
returntrue;
 }

boolflush(uint8_t *out_ptr, int *out_size) override {
 DBUG_TRACE;
if (m_error) returnfalse;

if (m_flush_finished) {
 *out_size = 0;
returntrue;
 }

 ZSTD_outBuffer out_buffer{out_ptr, static_cast<size_t>(*out_size), 0};
#if ZSTD_VERSION_NUMBER < 10400
constauto result = ZSTD_flushStream(m_stream, &out_buffer);
#else
constauto result =
ZSTD_compressStream2(m_stream, &out_buffer, &m_in_buffer, ZSTD_e_end);
#endif
if (is_error(result)) {
 *out_size = 0;
returnfalse;
 }
DBUG_LOG("debug", "zstandard(out_size:" << *out_size << "), executed with "
 << result);
 *out_size = static_cast<int>(out_buffer.pos);
 m_flush_finished = result == 0;
returntrue;
 }

#if ZSTD_VERSION_NUMBER < 10400
staticint32_tget_level_min() { return3; }
staticint32_tget_level_max() { return3; }
#else
staticint32_tget_level_min() { returnZSTD_minCLevel(); }
staticint32_tget_level_max() { returnZSTD_maxCLevel(); }
#endif

private:
boolis_error(constuint64_t result) {
if (!ZSTD_isError(result)) returnfalse;
DBUG_LOG("debug", "ZSTD error: " << ZSTD_getErrorName(result));
return m_error = true;
 }

 ZSTD_CStream *m_stream;
 ZSTD_inBuffer m_in_buffer{nullptr, 0, 0};
bool m_error{false};
bool m_flush_finished{false};
};

} // namespace protocol

#endif// PLUGIN_X_PROTOCOL_STREAM_COMPRESSION_COMPRESSION_ALGORITHM_ZSTD_H_