// 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 "storage/segment/page_io.h"

#include <crc32c/crc32c.h>

#include <gen_cpp/segment_v2.pb.h>

#include <stdint.h>

#include <algorithm>

#include <cstring>

#include <memory>

#include <ostream>

#include <string>

#include <utility>

#include "cloud/config.h"

#include "common/logging.h"

#include "cpp/sync_point.h"

#include "io/cache/block_file_cache.h"

#include "io/cache/block_file_cache_factory.h"

#include "io/cache/cached_remote_file_reader.h"

#include "io/fs/file_reader.h"

#include "io/fs/file_writer.h"

#include "runtime/runtime_profile.h"

#include "storage/cache/page_cache.h"

#include "storage/olap_common.h"

#include "storage/segment/encoding_info.h"

#include "storage/segment/page_handle.h"

#include "util/block_compression.h"

#include "util/coding.h"

#include "util/concurrency_stats.h"

#include "util/faststring.h"

namespace doris {

namespace segment_v2 {

Status PageIO::compress_page_body(BlockCompressionCodec* codec, double min_space_saving,

                                  const std::vector<Slice>& body, OwnedSlice* compressed_body) {

    size_t uncompressed_size = Slice::compute_total_size(body);

    if (codec != nullptr && !codec->exceed_max_compress_len(uncompressed_size)) {

        faststring buf;

        RETURN_IF_ERROR_OR_CATCH_EXCEPTION(codec->compress(body, uncompressed_size, &buf));

        double space_saving =

                1.0 - (cast_set<double>(buf.size()) / cast_set<double>(uncompressed_size));

        // return compressed body only when it saves more than min_space_saving

        if (space_saving > 0 && space_saving >= min_space_saving) {

            // shrink the buf to fit the len size to avoid taking

            // up the memory of the size MAX_COMPRESSED_SIZE

            RETURN_IF_CATCH_EXCEPTION(*compressed_body = buf.build());

            return Status::OK();

        }

    }

    // otherwise, do not compress

    OwnedSlice empty;

    *compressed_body = std::move(empty);

    return Status::OK();

}

Status PageIO::write_page(io::FileWriter* writer, const std::vector<Slice>& body,

                          const PageFooterPB& footer, PagePointer* result) {

    // sanity check of page footer

    CHECK(footer.has_type()) << "type must be set";

    CHECK(footer.has_uncompressed_size()) << "uncompressed_size must be set";

    switch (footer.type()) {

    case DATA_PAGE:

        CHECK(footer.has_data_page_footer());

        break;

    case INDEX_PAGE:

        CHECK(footer.has_index_page_footer());

        break;

    case DICTIONARY_PAGE:

        CHECK(footer.has_dict_page_footer());

        break;

    case SHORT_KEY_PAGE:

        CHECK(footer.has_short_key_page_footer());

        break;

    default:

        CHECK(false) << "Invalid page footer type: " << footer.type();

        break;

    }

    std::string footer_buf; // serialized footer + footer size

    footer.SerializeToString(&footer_buf);

    put_fixed32_le(&footer_buf, static_cast<uint32_t>(footer_buf.size()));

    std::vector<Slice> page = body;

    page.emplace_back(footer_buf);

    // checksum

    uint8_t checksum_buf[sizeof(uint32_t)];

    uint32_t checksum = 0;

    for (const auto& slice : page) {

        checksum = crc32c::Extend(checksum, (const uint8_t*)slice.data, slice.size);

    }

    encode_fixed32_le(checksum_buf, checksum);

    page.emplace_back(checksum_buf, sizeof(uint32_t));

    uint64_t offset = writer->bytes_appended();

    RETURN_IF_ERROR(writer->appendv(&page[0], page.size()));

    result->offset = offset;

    result->size = cast_set<uint32_t>(writer->bytes_appended() - offset);

    return Status::OK();

}

io::UInt128Wrapper file_cache_key_from_path(const std::string& seg_path) {

    std::string base = seg_path.substr(seg_path.rfind('/') + 1); // tricky: npos + 1 == 0

    return io::BlockFileCache::hash(base);

}

std::string file_cache_key_str(const std::string& seg_path) {

    return file_cache_key_from_path(seg_path).to_string();

}

Status PageIO::read_and_decompress_page_(const PageReadOptions& opts, PageHandle* handle,

                                         Slice* body, PageFooterPB* footer) {

    opts.sanity_check();

    opts.stats->total_pages_num++;

    auto cache = StoragePageCache::instance();

    PageCacheHandle cache_handle;

    StoragePageCache::CacheKey cache_key(opts.file_reader->path().native(),

                                         opts.file_reader->size(), opts.page_pointer.offset);

    VLOG_DEBUG << fmt::format("Reading page {}:{}:{}", cache_key.fname, cache_key.fsize,

                              cache_key.offset);

    if (opts.use_page_cache && cache && cache->lookup(cache_key, &cache_handle, opts.type)) {

        // we find page in cache, use it

        *handle = PageHandle(std::move(cache_handle));

        opts.stats->cached_pages_num++;

        // parse body and footer

        Slice page_slice = handle->data();

        uint32_t footer_size = decode_fixed32_le((uint8_t*)page_slice.data + page_slice.size - 4);

        std::string footer_buf(page_slice.data + page_slice.size - 4 - footer_size, footer_size);

        if (!footer->ParseFromString(footer_buf)) {

            return Status::Corruption("Bad page: invalid footer, footer_size={}, file={}",

                                      footer_size, opts.file_reader->path().native());

        }

        *body = Slice(page_slice.data, page_slice.size - 4 - footer_size);

        // If read from cache, then should also recorded in uncompressed bytes read counter.

        opts.stats->uncompressed_bytes_read += body->size;

        return Status::OK();

    }

    // every page contains 4 bytes footer length and 4 bytes checksum

    const uint32_t page_size = opts.page_pointer.size;

    if (page_size < 8) {

        return Status::Corruption("Bad page: too small size ({}), file={}", page_size,

                                  opts.file_reader->path().native());

    }

    // hold compressed page at first, reset to decompressed page later

    std::unique_ptr<DataPage> page =

            std::make_unique<DataPage>(page_size, opts.use_page_cache, opts.type);

    Slice page_slice(page->data(), page_size);

    {

        SCOPED_RAW_TIMER(&opts.stats->io_ns);

        size_t bytes_read = 0;

        RETURN_IF_ERROR(opts.file_reader->read_at(opts.page_pointer.offset, page_slice, &bytes_read,

                                                  &opts.io_ctx));

        DCHECK_EQ(bytes_read, page_size);

        opts.stats->compressed_bytes_read += page_size;

    }

    if (opts.verify_checksum) {

        uint32_t expect = decode_fixed32_le((uint8_t*)page_slice.data + page_slice.size - 4);

        uint32_t actual = crc32c::Crc32c(page_slice.data, page_slice.size - 4);

        // here const_cast is used for testing.

        InjectionContext ctx = {&actual, const_cast<PageReadOptions*>(&opts)};

        (void)ctx;

        TEST_INJECTION_POINT_CALLBACK("PageIO::read_and_decompress_page:crc_failure_inj", &ctx);

        if (expect != actual) {

            return Status::Corruption(

                    "Bad page: checksum mismatch (actual={} vs expect={}), file={}", actual, expect,

                    opts.file_reader->path().native());

        }

    }

    // remove checksum suffix

    page_slice.size -= 4;

    // parse and set footer

    uint32_t footer_size = decode_fixed32_le((uint8_t*)page_slice.data + page_slice.size - 4);

    if (!footer->ParseFromArray(page_slice.data + page_slice.size - 4 - footer_size, footer_size)) {

        return Status::Corruption("Bad page: invalid footer, footer_size={}, file={}", footer_size,

                                  opts.file_reader->path().native());

    }

    auto body_size = cast_set<uint32_t>(page_slice.size - 4 - footer_size);

    if (body_size != footer->uncompressed_size()) { // need decompress body

        if (opts.codec == nullptr) {

            return Status::Corruption(

                    "Bad page: page is compressed but codec is NO_COMPRESSION, file={}",

                    opts.file_reader->path().native());

        }

        SCOPED_CONCURRENCY_COUNT(ConcurrencyStatsManager::instance().page_io_decompress);

        SCOPED_RAW_TIMER(&opts.stats->decompress_ns);

        std::unique_ptr<DataPage> decompressed_page = std::make_unique<DataPage>(

                footer->uncompressed_size() + footer_size + 4, opts.use_page_cache, opts.type);

        // decompress page body

        Slice compressed_body(page_slice.data, body_size);

        Slice decompressed_body(decompressed_page->data(), footer->uncompressed_size());

        RETURN_IF_ERROR(opts.codec->decompress(compressed_body, &decompressed_body));

        if (decompressed_body.size != footer->uncompressed_size()) {

            return Status::Corruption(

                    "Bad page: record uncompressed size={} vs real decompressed size={}, file={}",

                    footer->uncompressed_size(), decompressed_body.size,

                    opts.file_reader->path().native());

        }

        // append footer and footer size

        memcpy(decompressed_body.data + decompressed_body.size, page_slice.data + body_size,

               footer_size + 4);

        // free memory of compressed page

        page = std::move(decompressed_page);

        page_slice = Slice(page->data(), footer->uncompressed_size() + footer_size + 4);

    }

    if (opts.pre_decode) {

        const auto* encoding_info = opts.encoding_info;

        if (opts.is_dict_page) {

            // for dict page, we need to use encoding_info based on footer->dict_page_footer().encoding()

            // to get its pre_decoder

            RETURN_IF_ERROR(EncodingInfo::get(FieldType::OLAP_FIELD_TYPE_VARCHAR,

                                              footer->dict_page_footer().encoding(), {},

                                              &encoding_info));

        }

        if (encoding_info) {

            auto* pre_decoder = encoding_info->get_data_page_pre_decoder();

            if (pre_decoder) {

                SCOPED_CONCURRENCY_COUNT(ConcurrencyStatsManager::instance().page_io_pre_decode);

                RETURN_IF_ERROR(pre_decoder->decode(

                        &page, &page_slice,

                        footer->data_page_footer().nullmap_size() + footer_size + 4,

                        opts.use_page_cache, opts.type, opts.file_reader->path().native()));

            }

        }

    }

    *body = Slice(page_slice.data, page_slice.size - 4 - footer_size);

    page->reset_size(page_slice.size);

    // Uncompressed has 2 meanings: uncompress and decode. The buffer in pagecache maybe

    // uncompressed or decoded. So that should update the uncompressed_bytes_read counter

    // just before add it to pagecache, it will be consistency with reading data from page cache.

    opts.stats->uncompressed_bytes_read += body->size;

    if (opts.use_page_cache && cache) {

        SCOPED_CONCURRENCY_COUNT(ConcurrencyStatsManager::instance().page_io_insert_page_cache);

        // insert this page into cache and return the cache handle

        cache->insert(cache_key, page.get(), &cache_handle, opts.type, opts.kept_in_memory);

        *handle = PageHandle(std::move(cache_handle));

    } else {

        *handle = PageHandle(page.get());

    }

    page.release(); // memory now managed by handle

    return Status::OK();

}

Status PageIO::read_and_decompress_page(const PageReadOptions& opts, PageHandle* handle,

                                        Slice* body, PageFooterPB* footer) {

    // First try to read with file cache

    Status st = do_read_and_decompress_page(opts, handle, body, footer);

    if (!st.is<ErrorCode::CORRUPTION>() || !config::is_cloud_mode()) {

        return st;

    }

    auto* cached_file_reader = dynamic_cast<io::CachedRemoteFileReader*>(opts.file_reader);

    if (cached_file_reader == nullptr) {

        return st;

    }

    // If we get CORRUPTION error and using file cache, clear cache and retry

    LOG(WARNING) << "Bad page may be read from file cache, need retry."

                 << " error msg: " << st.msg()

                 << " file path: " << opts.file_reader->path().native()

                 << " offset: " << opts.page_pointer.offset;

    // Remove cache if exists

    const std::string path = opts.file_reader->path().string();

    auto file_key = file_cache_key_from_path(path);

    auto* file_cache = io::FileCacheFactory::instance()->get_by_path(file_key);

    if (file_cache) {

        file_cache->remove_if_cached(file_key);

    }

    // Retry with file cache

    st = do_read_and_decompress_page(opts, handle, body, footer);

    if (!st.is<ErrorCode::CORRUPTION>()) {

        return st;

    }

    LOG(WARNING) << "Corruption again with retry downloading cache,"

                 << " error msg: " << st.msg()

                 << " file path: " << opts.file_reader->path().native()

                 << " offset: " << opts.page_pointer.offset;

    PageReadOptions new_opts = opts;

    new_opts.file_reader = cached_file_reader->get_remote_reader();

    st = do_read_and_decompress_page(new_opts, handle, body, footer);

    if (!st.ok()) {

        LOG(WARNING) << "Corruption again with retry read directly from remote,"

                     << " error msg: " << st.msg()

                     << " file path: " << opts.file_reader->path().native()

                     << " offset: " << opts.page_pointer.offset << " Give up.";

    }

    return st;

}

} // namespace segment_v2

} // namespace doris