// 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 "cloud/delete_bitmap_file_reader.h"

#include "cloud/delete_bitmap_file_writer.h"

#include "common/status.h"

#include "io/fs/file_reader.h"

#include "io/fs/packed_file_reader.h"

#include "util/coding.h"

namespace doris {

DeleteBitmapFileReader::DeleteBitmapFileReader(int64_t tablet_id, const std::string& rowset_id,

                                               std::optional<StorageResource>& storage_resource)

        : _tablet_id(tablet_id), _rowset_id(rowset_id), _storage_resource(storage_resource) {}

DeleteBitmapFileReader::DeleteBitmapFileReader(int64_t tablet_id, const std::string& rowset_id,

                                               std::optional<StorageResource>& storage_resource,

                                               const PackedSliceLocationPB& packed_location)

        : _tablet_id(tablet_id),

          _rowset_id(rowset_id),

          _storage_resource(storage_resource),

          _is_packed(true),

          _packed_offset(packed_location.offset()),

          _packed_size(packed_location.size()),

          _packed_file_path(packed_location.packed_file_path()),

          _packed_file_size(packed_location.has_packed_file_size()

                                    ? packed_location.packed_file_size()

                                    : -1) {}

DeleteBitmapFileReader::~DeleteBitmapFileReader() = default;

Status DeleteBitmapFileReader::init() {

#ifdef BE_TEST

    _path = "./log/" + _rowset_id + "_delete_bitmap.db";

    bool exists = false;

    RETURN_IF_ERROR(io::global_local_filesystem()->exists(_path, &exists));

    if (!exists) {

        return Status::NotFound("{} doesn't exist", _path);

    }

    RETURN_IF_ERROR(io::global_local_filesystem()->open_file(_path, &_file_reader));

    return Status::OK();

#endif

    if (!_storage_resource) {

        return Status::InternalError("invalid storage resource for tablet_id={}", _tablet_id);

    }

    if (_is_packed) {

        // Read from packed file

        io::FileReaderSPtr inner_reader;

        io::FileReaderOptions opts;

        if (_packed_file_size > 0) {

            opts.file_size = _packed_file_size;

        }

        opts.cache_type = io::FileCachePolicy::NO_CACHE;

        RETURN_IF_ERROR(_storage_resource->fs->open_file(io::Path(_packed_file_path), &inner_reader,

                                                         &opts));

        _path = _storage_resource->remote_delete_bitmap_path(_tablet_id, _rowset_id);

        _file_reader = std::make_shared<io::PackedFileReader>(

                std::move(inner_reader), io::Path(_path), _packed_offset, _packed_size);

    } else {

        // Read from standalone file

        _path = _storage_resource->remote_delete_bitmap_path(_tablet_id, _rowset_id);

        io::FileReaderOptions opts;

        RETURN_IF_ERROR(_storage_resource->fs->open_file(_path, &_file_reader, &opts));

    }

    return Status::OK();

}

Status DeleteBitmapFileReader::close() {

    if (_file_reader) {

        return _file_reader->close();

    }

    return Status::OK();

}

Status DeleteBitmapFileReader::read(DeleteBitmapPB& delete_bitmap) {

    size_t bytes_read = 0;

    size_t offset = 0;

    // 0. read magic number

    {

        uint8_t magic_buf[DeleteBitmapFileWriter::MAGIC_SIZE];

        RETURN_IF_ERROR(_file_reader->read_at(

                offset, {magic_buf, DeleteBitmapFileWriter::MAGIC_SIZE}, &bytes_read));

        offset += DeleteBitmapFileWriter::MAGIC_SIZE;

        if (bytes_read != DeleteBitmapFileWriter::MAGIC_SIZE ||

            memcmp(magic_buf, DeleteBitmapFileWriter::DELETE_BITMAP_MAGIC,

                   DeleteBitmapFileWriter::MAGIC_SIZE) != 0) {

            return Status::InternalError(

                    "read delete bitmap failed from {} because magic is not "

                    "matched",

                    _path);

        }

    }

    // 1. read delete bitmap proto length

    size_t delete_bitmap_len = 0;

    {

        uint8_t len_buf[DeleteBitmapFileWriter::LENGTH_SIZE];

        RETURN_IF_ERROR(_file_reader->read_at(

                offset, {len_buf, DeleteBitmapFileWriter::LENGTH_SIZE}, &bytes_read));

        offset += DeleteBitmapFileWriter::LENGTH_SIZE;

        delete_bitmap_len = decode_fixed64_le(len_buf);

        if (delete_bitmap_len == 0) {

            return Status::InternalError("read delete bitmap failed from {} because length is 0",

                                         _path);

        }

        if (offset == _file_reader->size()) {

            LOG(WARNING) << "read delete bitmap failed because reach end of file=" << _path

                         << ", file size=" << _file_reader->size() << ", offset=" << offset

                         << ", delete_bitmap_len=" << delete_bitmap_len;

            return Status::InternalError(

                    "read delete bitmap failed from {} because reach end of file", _path);

        }

    }

    // 2. read delete bitmap

    std::string delete_bitmap_buf;

    {

        delete_bitmap_buf.resize(delete_bitmap_len);

        RETURN_IF_ERROR(_file_reader->read_at(

                offset, {delete_bitmap_buf.c_str(), delete_bitmap_len}, &bytes_read));

        offset += delete_bitmap_len;

        if (!delete_bitmap.ParseFromString(delete_bitmap_buf)) {

            LOG(WARNING) << "deserialize delete bitmap failed from file=" << _path

                         << ", file size=" << _file_reader->size()

                         << ", delete_bitmap_len=" << delete_bitmap_len

                         << ", bytes_read=" << bytes_read;

            return Status::InternalError("deserialize delete bitmap failed from {}", _path);

        }

    }

    // 3. checksum

    uint8_t checksum_len_buf[DeleteBitmapFileWriter::CHECKSUM_SIZE];

    RETURN_IF_ERROR(_file_reader->read_at(

            offset, {checksum_len_buf, DeleteBitmapFileWriter::CHECKSUM_SIZE}, &bytes_read));

    offset += DeleteBitmapFileWriter::CHECKSUM_SIZE;

    uint32_t checksum = decode_fixed32_le(checksum_len_buf);

    uint32_t computed_checksum = crc32c::Crc32c(delete_bitmap_buf.data(), delete_bitmap_len);

    if (computed_checksum != checksum) {

        return Status::InternalError("delete bitmap checksum failed from file=" + _path +

                                     ", computed checksum=" + std::to_string(computed_checksum) +

                                     ", expected=" + std::to_string(checksum));

    }

    return Status::OK();

}

} // namespace doris