// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements. See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership. The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License. You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied. See the License for the

// specific language governing permissions and limitations

// under the License.

#include "load/group_commit/wal/wal_writer.h"

#include <crc32c/crc32c.h>

#include <gen_cpp/AgentService_types.h>

#include <gen_cpp/FrontendService_types.h>

#include "common/config.h"

#include "common/status.h"

#include "io/fs/encrypted_fs_factory.h"

#include "io/fs/file_system.h"

#include "io/fs/file_writer.h"

#include "io/fs/local_file_system.h"

#include "io/fs/path.h"

#include "load/group_commit/wal/wal_manager.h"

#include "storage/storage_engine.h"

#include "util/thrift_rpc_helper.h"

namespace doris {

const char* k_wal_magic = "WAL1";

const uint32_t k_wal_magic_length = 4;

WalWriter::WalWriter(const std::string& file_name) : _file_name(file_name) {}

WalWriter::~WalWriter() {}

Status determine_wal_fs(int64_t db_id, int64_t tb_id, io::FileSystemSPtr& fs) {

if (!config::enable_wal_tde) {

fs = io::global_local_filesystem();

return Status::OK();

}

#ifndef BE_TEST

TNetworkAddress master_addr = ExecEnv::GetInstance()->cluster_info()->master_fe_addr;

TGetTableTDEInfoRequest req;

req.__set_db_id(db_id);

req.__set_table_id(tb_id);

TGetTableTDEInfoResult ret;

RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(

master_addr.hostname, master_addr.port,

[&req, &ret](FrontendServiceConnection& client) {

client->getTableTDEInfo(ret, req);

}));

if (auto st = Status::create(ret.status); !st) {

return st;

}

auto encrypt_algorithm = [&ret]() -> EncryptionAlgorithmPB {

switch (ret.algorithm) {

case doris::TEncryptionAlgorithm::AES256:

return EncryptionAlgorithmPB::AES_256_CTR;

case doris::TEncryptionAlgorithm::SM4:

return EncryptionAlgorithmPB::SM4_128_CTR;

default:

return EncryptionAlgorithmPB::PLAINTEXT;

}

}();

auto local_fs = io::global_local_filesystem();

fs = io::make_file_system(local_fs, encrypt_algorithm);

#else

fs = io::global_local_filesystem();

#endif

return Status::OK();

}

Status WalWriter::init(const io::FileSystemSPtr& fs) {

io::Path wal_path = _file_name;

auto parent_path = wal_path.parent_path();

bool exists = false;

RETURN_IF_ERROR(fs->exists(parent_path, &exists));

if (!exists) {

RETURN_IF_ERROR(fs->create_directory(parent_path));

}

RETURN_IF_ERROR(fs->create_file(_file_name, &_file_writer));

LOG(INFO) << "create wal " << _file_name;

return Status::OK();

}

Status WalWriter::finalize() {

if (!_file_writer) {

return Status::InternalError("wal writer is null,fail to close file={}", _file_name);

}

auto st = _file_writer->close();

if (!st.ok()) {

LOG(WARNING) << "fail to close wal " << _file_name;

}

return Status::OK();

}

Status WalWriter::append_blocks(const PBlockArray& blocks) {

if (!_file_writer) {

return Status::InternalError("wal writer is null,fail to write file={}", _file_name);

}

size_t total_size = 0;

size_t offset = 0;

for (const auto& block : blocks) {

uint8_t len_buf[sizeof(uint64_t)];

uint64_t block_length = block->ByteSizeLong();

total_size += LENGTH_SIZE + block_length + CHECKSUM_SIZE;

encode_fixed64_le(len_buf, block_length);

RETURN_IF_ERROR(_file_writer->append({len_buf, sizeof(uint64_t)}));

offset += LENGTH_SIZE;

std::string content = block->SerializeAsString();

RETURN_IF_ERROR(_file_writer->append(content));

offset += block_length;

uint8_t checksum_buf[sizeof(uint32_t)];

uint32_t checksum = crc32c::Crc32c(content.data(), block_length);

encode_fixed32_le(checksum_buf, checksum);

RETURN_IF_ERROR(_file_writer->append({checksum_buf, sizeof(uint32_t)}));

offset += CHECKSUM_SIZE;

}

if (offset != total_size) {

return Status::InternalError(

"failed to write block to wal expected= " + std::to_string(total_size) +

",actually=" + std::to_string(offset));

}

return Status::OK();

}

Status WalWriter::append_header(std::string col_ids) {

if (!_file_writer) {

return Status::InternalError("wal writer is null,fail to write file={}", _file_name);

}

size_t total_size = 0;

uint64_t length = col_ids.size();

total_size += k_wal_magic_length;

total_size += VERSION_SIZE;

total_size += LENGTH_SIZE;

total_size += length;

size_t offset = 0;

RETURN_IF_ERROR(_file_writer->append({k_wal_magic, k_wal_magic_length}));

offset += k_wal_magic_length;

uint8_t version_buf[sizeof(uint32_t)];

encode_fixed32_le(version_buf, WAL_VERSION);

RETURN_IF_ERROR(_file_writer->append({version_buf, sizeof(uint32_t)}));

offset += VERSION_SIZE;

uint8_t len_buf[sizeof(uint64_t)];

encode_fixed64_le(len_buf, length);

RETURN_IF_ERROR(_file_writer->append({len_buf, sizeof(uint64_t)}));

offset += LENGTH_SIZE;

RETURN_IF_ERROR(_file_writer->append(col_ids));

offset += length;

if (offset != total_size) {

return Status::InternalError(

"failed to write header to wal expected= " + std::to_string(total_size) +

",actually=" + std::to_string(offset));

}

return Status::OK();

}

} // namespace doris