file_header.h

// 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.

#pragma once

#include <stdio.h>
#include <sys/stat.h>

#include <memory>
#include <string>
#include <vector>

#include "cpp/lru_cache.h"
#include "io/fs/file_reader.h"
#include "io/fs/file_writer.h"
#include "io/fs/local_file_system.h"
#include "storage/olap_common.h"
#include "storage/olap_define.h"
#include "storage/utils.h"
#include "util/debug_util.h"

namespace doris {

using FixedFileHeader = struct _FixedFileHeader {
    // the length of the entire file
    uint32_t file_length;
    // Checksum of the file's contents except the FileHeader
    uint32_t checksum;
    // Protobuf length of section
    uint32_t protobuf_length;
    // Checksum of Protobuf part
    uint32_t protobuf_checksum;
} __attribute__((packed));

using FixedFileHeaderV2 = struct _FixedFileHeaderV2 {
    uint64_t magic_number;
    uint32_t version;
    // the length of the entire file
    uint64_t file_length;
    // Checksum of the file's contents except the FileHeader
    uint32_t checksum;
    // Protobuf length of section
    uint64_t protobuf_length;
    // Checksum of Protobuf part
    uint32_t protobuf_checksum;
} __attribute__((packed));

template <typename MessageType, typename ExtraType = uint32_t>
class FileHeader {
public:
    FileHeader(const std::string& file_path) : _file_path(file_path) {
        memset(&_fixed_file_header, 0, sizeof(_fixed_file_header));
        memset(&_extra_fixed_header, 0, sizeof(_extra_fixed_header));
        _fixed_file_header_size = sizeof(_fixed_file_header);
    }
    ~FileHeader() = default;

    // To calculate the length of the proto part, it needs to be called after the proto is operated,
    // and prepare must be called before calling serialize
    Status prepare();

    // call prepare() first, serialize() will write fixed header and protobuffer.
    // Write the header to the starting position of the incoming file handle
    Status serialize();

    // read from file, validate file length, signature and alder32 of protobuffer.
    // Read the header from the beginning of the incoming file handle
    Status deserialize();

    // Check the validity of Header
    // it is actually call deserialize().
    Status validate();

    // write to memory
    Status serialize_to_memory(uint8_t* buffer, size_t buffer_size);

    // read from memory
    Status deserialize_from_memory(const uint8_t* buffer, size_t buffer_size);

    uint64_t file_length() const { return _fixed_file_header.file_length; }
    uint32_t checksum() const { return _fixed_file_header.checksum; }
    const ExtraType& extra() const { return _extra_fixed_header; }
    ExtraType* mutable_extra() { return &_extra_fixed_header; }
    const MessageType& message() const { return _proto; }
    MessageType* mutable_message() { return &_proto; }
    uint64_t size() const {
        return _fixed_file_header_size + sizeof(_extra_fixed_header) +
               _fixed_file_header.protobuf_length;
    }

    void set_file_length(uint64_t file_length) { _fixed_file_header.file_length = file_length; }
    void set_checksum(uint32_t checksum) { _fixed_file_header.checksum = checksum; }

private:
    std::string _file_path;
    FixedFileHeaderV2 _fixed_file_header;
    uint32_t _fixed_file_header_size;

    std::string _proto_string;
    ExtraType _extra_fixed_header;
    MessageType _proto;
};

// FileHeader implementation
template <typename MessageType, typename ExtraType>
Status FileHeader<MessageType, ExtraType>::prepare() {
    try {
        if (!_proto.SerializeToString(&_proto_string)) {
            return Status::Error<ErrorCode::SERIALIZE_PROTOBUF_ERROR>(
                    "serialize file header to string error. [path={}]", _file_path);
        }
    } catch (...) {
        return Status::Error<ErrorCode::SERIALIZE_PROTOBUF_ERROR>(
                "serialize file header to string error. [path={}]", _file_path);
    }

    _fixed_file_header.protobuf_checksum =
            olap_adler32(olap_adler32_init(), _proto_string.c_str(), _proto_string.size());

    _fixed_file_header.checksum = 0;
    _fixed_file_header.protobuf_length = _proto_string.size();
    _fixed_file_header.file_length = size();
    _fixed_file_header.version = OLAP_DATA_VERSION_APPLIED;
    _fixed_file_header.magic_number = OLAP_FIX_HEADER_MAGIC_NUMBER;

    return Status::OK();
}

template <typename MessageType, typename ExtraType>
Status FileHeader<MessageType, ExtraType>::serialize() {
    // write to file
    io::FileWriterPtr file_writer;
    RETURN_IF_ERROR(io::global_local_filesystem()->create_file(_file_path, &file_writer));
    RETURN_IF_ERROR(
            file_writer->append({(const uint8_t*)&_fixed_file_header, _fixed_file_header_size}));
    RETURN_IF_ERROR(file_writer->append(
            {(const uint8_t*)&_extra_fixed_header, sizeof(_extra_fixed_header)}));
    RETURN_IF_ERROR(file_writer->append({_proto_string}));
    return file_writer->close();
}

template <typename MessageType, typename ExtraType>
Status FileHeader<MessageType, ExtraType>::serialize_to_memory(uint8_t* buffer,
                                                               size_t buffer_size) {
    if (buffer_size < size()) {
        return Status::Error<ErrorCode::INVALID_ARGUMENT>(
                "buffer size is too small. required={}, provided={}", size(), buffer_size);
    }

    uint8_t* ptr = buffer;
    memcpy(ptr, &_fixed_file_header, _fixed_file_header_size);
    ptr += _fixed_file_header_size;

    memcpy(ptr, &_extra_fixed_header, sizeof(_extra_fixed_header));
    ptr += sizeof(_extra_fixed_header);

    memcpy(ptr, _proto_string.data(), _proto_string.size());
    return Status::OK();
}

template <typename MessageType, typename ExtraType>
Status FileHeader<MessageType, ExtraType>::deserialize() {
    io::FileReaderSPtr file_reader;
    RETURN_IF_ERROR(io::global_local_filesystem()->open_file(_file_path, &file_reader));
    off_t real_file_length = 0;
    uint32_t real_protobuf_checksum = 0;
    size_t bytes_read = 0;
    RETURN_IF_ERROR(file_reader->read_at(
            0, {(const uint8_t*)&_fixed_file_header, _fixed_file_header_size}, &bytes_read));
    DCHECK(_fixed_file_header_size == bytes_read)
            << " deserialize read bytes dismatch, request bytes " << _fixed_file_header_size
            << " actual read " << bytes_read;

    //Status read_at(size_t offset, Slice result, size_t* bytes_read,
    //             const IOContext* io_ctx = nullptr);

    if (_fixed_file_header.magic_number != OLAP_FIX_HEADER_MAGIC_NUMBER) {
        VLOG_TRACE << "old fix header found, magic num=" << _fixed_file_header.magic_number;
        FixedFileHeader tmp_header;
        RETURN_IF_ERROR(file_reader->read_at(0, {(const uint8_t*)&tmp_header, sizeof(tmp_header)},
                                             &bytes_read));
        DCHECK(sizeof(tmp_header) == bytes_read)
                << " deserialize read bytes dismatch, request bytes " << sizeof(tmp_header)
                << " actual read " << bytes_read;
        _fixed_file_header.file_length = tmp_header.file_length;
        _fixed_file_header.checksum = tmp_header.checksum;
        _fixed_file_header.protobuf_length = tmp_header.protobuf_length;
        _fixed_file_header.protobuf_checksum = tmp_header.protobuf_checksum;
        _fixed_file_header.magic_number = OLAP_FIX_HEADER_MAGIC_NUMBER;
        _fixed_file_header.version = OLAP_DATA_VERSION_APPLIED;
        _fixed_file_header_size = sizeof(tmp_header);
    }

    VLOG_NOTICE << "fix head loaded. file_length=" << _fixed_file_header.file_length
                << ", checksum=" << _fixed_file_header.checksum
                << ", protobuf_length=" << _fixed_file_header.protobuf_length
                << ", magic_number=" << _fixed_file_header.magic_number
                << ", version=" << _fixed_file_header.version;

    RETURN_IF_ERROR(file_reader->read_at(
            _fixed_file_header_size,
            {(const uint8_t*)&_extra_fixed_header, sizeof(_extra_fixed_header)}, &bytes_read));

    std::unique_ptr<char[]> buf(new (std::nothrow) char[_fixed_file_header.protobuf_length]);
    if (nullptr == buf) {
        char errmsg[64];
        return Status::Error<ErrorCode::MEM_ALLOC_FAILED>(
                "malloc protobuf buf error. file={}, error={}", file_reader->path().native(),
                strerror_r(errno, errmsg, 64));
    }
    RETURN_IF_ERROR(file_reader->read_at(_fixed_file_header_size + sizeof(_extra_fixed_header),
                                         {buf.get(), _fixed_file_header.protobuf_length},
                                         &bytes_read));
    real_file_length = file_reader->size();

    if (file_length() != static_cast<uint64_t>(real_file_length)) {
        return Status::InternalError(
                "file length is not match. file={}, file_length={}, real_file_length={}",
                file_reader->path().native(), file_length(), real_file_length);
    }

    // check proto checksum
    real_protobuf_checksum =
            olap_adler32(olap_adler32_init(), buf.get(), _fixed_file_header.protobuf_length);

    if (real_protobuf_checksum != _fixed_file_header.protobuf_checksum) {
        // When compiling using gcc there woule be error like:
        // Cannot bind packed field '_FixedFileHeaderV2::protobuf_checksum' to 'unsigned int&'
        // so we need to using unary operator+ to evaluate one value to pass
        // to status to successfully compile.
        return Status::InternalError("checksum is not match. file={}, expect={}, actual={}",
                                     file_reader->path().native(),
                                     +_fixed_file_header.protobuf_checksum, real_protobuf_checksum);
    }

    try {
        std::string protobuf_str(buf.get(), _fixed_file_header.protobuf_length);

        if (!_proto.ParseFromString(protobuf_str)) {
            return Status::Error<ErrorCode::PARSE_PROTOBUF_ERROR>(
                    "fail to parse file content to protobuf object. file={}",
                    file_reader->path().native());
        }
    } catch (...) {
        LOG(WARNING) << "fail to load protobuf. file='" << file_reader->path().native();
        return Status::Error<ErrorCode::PARSE_PROTOBUF_ERROR>("fail to load protobuf. file={}",
                                                              file_reader->path().native());
    }

    return Status::OK();
}

template <typename MessageType, typename ExtraType>
Status FileHeader<MessageType, ExtraType>::deserialize_from_memory(const uint8_t* buffer,
                                                                   size_t buffer_size) {
    if (buffer_size < sizeof(FixedFileHeaderV2)) {
        return Status::Error<ErrorCode::INVALID_ARGUMENT>(
                "buffer size is too small to contain a valid header. provided={}", buffer_size);
    }

    const uint8_t* ptr = buffer;
    memcpy(&_fixed_file_header, ptr, _fixed_file_header_size);
    ptr += _fixed_file_header_size;

    if (_fixed_file_header.magic_number != OLAP_FIX_HEADER_MAGIC_NUMBER) {
        VLOG_TRACE << "old fix header found, magic num=" << _fixed_file_header.magic_number;
        if (buffer_size < sizeof(FixedFileHeader)) {
            return Status::Error<ErrorCode::INVALID_ARGUMENT>(
                    "buffer size is too small to contain a valid old header. provided={}",
                    buffer_size);
        }
        FixedFileHeader tmp_header;
        memcpy(&tmp_header, buffer, sizeof(tmp_header));
        _fixed_file_header.file_length = tmp_header.file_length;
        _fixed_file_header.checksum = tmp_header.checksum;
        _fixed_file_header.protobuf_length = tmp_header.protobuf_length;
        _fixed_file_header.protobuf_checksum = tmp_header.protobuf_checksum;
        _fixed_file_header.magic_number = OLAP_FIX_HEADER_MAGIC_NUMBER;
        _fixed_file_header.version = OLAP_DATA_VERSION_APPLIED;
        _fixed_file_header_size = sizeof(tmp_header);
        ptr = buffer + sizeof(tmp_header);
    }

    VLOG_NOTICE << "fix head loaded. file_length=" << _fixed_file_header.file_length
                << ", checksum=" << _fixed_file_header.checksum
                << ", protobuf_length=" << _fixed_file_header.protobuf_length
                << ", magic_number=" << _fixed_file_header.magic_number
                << ", version=" << _fixed_file_header.version;

    if (buffer_size < _fixed_file_header_size + sizeof(_extra_fixed_header) +
                              _fixed_file_header.protobuf_length) {
        return Status::Error<ErrorCode::INVALID_ARGUMENT>(
                "buffer size is too small to contain the entire header and protobuf. required={}, "
                "provided={}",
                _fixed_file_header_size + sizeof(_extra_fixed_header) +
                        _fixed_file_header.protobuf_length,
                buffer_size);
    }

    memcpy(&_extra_fixed_header, ptr, sizeof(_extra_fixed_header));
    ptr += sizeof(_extra_fixed_header);

    std::unique_ptr<char[]> buf(new (std::nothrow) char[_fixed_file_header.protobuf_length]);
    if (nullptr == buf) {
        char errmsg[64];
        return Status::Error<ErrorCode::MEM_ALLOC_FAILED>("malloc protobuf buf error. error={}",
                                                          strerror_r(errno, errmsg, 64));
    }
    memcpy(buf.get(), ptr, _fixed_file_header.protobuf_length);

    // check proto checksum
    uint32_t real_protobuf_checksum =
            olap_adler32(olap_adler32_init(), buf.get(), _fixed_file_header.protobuf_length);

    if (real_protobuf_checksum != _fixed_file_header.protobuf_checksum) {
        return Status::InternalError("checksum is not match. expect={}, actual={}",
                                     +_fixed_file_header.protobuf_checksum, real_protobuf_checksum);
    }

    try {
        std::string protobuf_str(buf.get(), _fixed_file_header.protobuf_length);

        if (!_proto.ParseFromString(protobuf_str)) {
            return Status::Error<ErrorCode::PARSE_PROTOBUF_ERROR>(
                    "fail to parse buffer content to protobuf object.");
        }
    } catch (...) {
        LOG(WARNING) << "fail to load protobuf from buffer.";
        return Status::Error<ErrorCode::PARSE_PROTOBUF_ERROR>("fail to load protobuf from buffer.");
    }

    return Status::OK();
}

template <typename MessageType, typename ExtraType>
Status FileHeader<MessageType, ExtraType>::validate() {
    return deserialize();
}

} // namespace doris