/*
 * 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 "page_writer.h"

#include "common/config/config.h"
#include "common/logger/elog.h"
#include "compress/compressor_factory.h"
#include "encoding/encoder_factory.h"

using namespace common;

namespace storage {

/* ================ PageData ================ */
int PageData::init(ByteStream &time_bs, ByteStream &value_bs,
                   Compressor *compressor) {
    int ret = E_OK;
    time_buf_size_ = time_bs.total_size();
    value_buf_size_ = value_bs.total_size();
    uint32_t var_size = get_var_uint_size(time_buf_size_);
    uncompressed_size_ = var_size + time_buf_size_ + value_buf_size_;
    uncompressed_buf_ =
        (char *)mem_alloc(uncompressed_size_, MOD_PAGE_WRITER_OUTPUT_STREAM);
    compressor_ = compressor;
    if (IS_NULL(uncompressed_buf_)) {
        return E_OOM;
    }
    if (time_buf_size_ == 0 || value_buf_size_ == 0) {
        return E_INVALID_ARG;
    }
    if (RET_FAIL(SerializationUtil::write_var_uint(
            time_buf_size_, uncompressed_buf_, var_size))) {
    } else if (RET_FAIL(
                   common::copy_bs_to_buf(time_bs, uncompressed_buf_ + var_size,
                                          uncompressed_size_ - var_size))) {
    } else if (RET_FAIL(common::copy_bs_to_buf(
                   value_bs, uncompressed_buf_ + var_size + time_buf_size_,
                   uncompressed_size_ - var_size - time_buf_size_))) {
    } else {
        // TODO
        // NOTE: different compressor may have different compress API
        // Be carefull about the memory.
        if (RET_FAIL(compressor->reset(true))) {
        } else if (RET_FAIL(compressor->compress(
                       uncompressed_buf_, uncompressed_size_, compressed_buf_,
                       compressed_size_))) {
        }
    }
#if DEBUG_SE
    std::cout << "PageData::init. time_buf_size=" << time_buf_size_
              << ", value_buf_size=" << value_buf_size_
              << ", uncompressed_size=" << uncompressed_size_
              << ", compressed_size=" << compressed_size_ << std::endl;
    // DEBUG_hex_dump_buf("compressed_buf=", compressed_buf_, compressed_size_);
#endif
    return ret;
}

/* ================ PageWriter ================ */
int PageWriter::init(TSDataType data_type, TSEncoding encoding,
                     CompressionType compression) {
    int ret = E_OK;
    data_type_ = data_type;
    if (nullptr == (time_encoder_ = EncoderFactory::alloc_time_encoder())) {
        ret = E_OOM;
    } else if (nullptr == (value_encoder_ = EncoderFactory::alloc_value_encoder(
                               encoding, data_type))) {
        ret = E_OOM;
    } else if (nullptr ==
               (statistic_ = StatisticFactory::alloc_statistic(data_type))) {
        ret = E_OOM;
    } else if (nullptr == (compressor_ = CompressorFactory::alloc_compressor(
                               compression))) {
        ret = E_OOM;
    }
    if (ret != E_OK) {
        if (time_encoder_ != nullptr) {
            EncoderFactory::free(time_encoder_);
            time_encoder_ = nullptr;
        }
        if (value_encoder_ != nullptr) {
            EncoderFactory::free(value_encoder_);
            value_encoder_ = nullptr;
        }
        if (statistic_ != nullptr) {
            statistic_->destroy();
            StatisticFactory::free(statistic_);
            statistic_ = nullptr;
        }
    }
    if (ret == E_OK) {
        is_inited_ = true;
    }
    return ret;
}

/*
 * free out_stream memory, reset statistic_,
 */
void PageWriter::reset() {
    if (time_encoder_ != nullptr) {
        time_encoder_->reset();
    }
    if (value_encoder_ != nullptr) {
        value_encoder_->reset();
    }
    if (statistic_ != nullptr) {
        statistic_->reset();
    }
    time_out_stream_.reset();
    value_out_stream_.reset();
}

void PageWriter::destroy() {
    if (is_inited_) {
        is_inited_ = false;
        time_encoder_->destroy();
        value_encoder_->destroy();
        statistic_->destroy();

        EncoderFactory::free(time_encoder_);
        time_encoder_ = nullptr;
        EncoderFactory::free(value_encoder_);
        value_encoder_ = nullptr;
        statistic_->destroy();
        StatisticFactory::free(statistic_);
        statistic_ = nullptr;
        compressor_->destroy();
        CompressorFactory::free(compressor_);
        compressor_ = nullptr;
    }
}

int PageWriter::write_to_chunk(ByteStream &pages_data, bool write_header,
                               bool write_statistic,
                               bool write_data_to_chunk_data) {
#if DEBUG_SE
    std::cout << "PageWriter::write_to_chunk at position "
              << pages_data.total_size() << " of chunk_data." << std::endl;
#endif
    int ret = E_OK;
    if (RET_FAIL(prepare_end_page())) {
        return ret;
    }
    if (RET_FAIL(cur_page_data_.init(time_out_stream_, value_out_stream_,
                                     compressor_))) {
    }

    if (IS_SUCC(ret) && write_header) {
        if (RET_FAIL(SerializationUtil::write_var_uint(
                cur_page_data_.uncompressed_size_, pages_data))) {
        } else if (RET_FAIL(SerializationUtil::write_var_uint(
                       cur_page_data_.compressed_size_, pages_data))) {
        }
    }
    // std::cout << "PageWriter::write_to_chunk after write_header. pos=" <<
    // pages_data.total_size() << std::endl;
    if (IS_SUCC(ret) && write_statistic) {
        if (RET_FAIL(statistic_->serialize_to(pages_data))) {
        }
    }
    // std::cout << "PageWriter::write_to_chunk after write_stat. pos=" <<
    // pages_data.total_size() << std::endl; DEBUG_print_byte_stream("In
    // PageWriter::write_to_chunk, before writer page data: pages_data = ",
    // pages_data);
    if (IS_SUCC(ret) && write_data_to_chunk_data) {
        // DEBUG_hex_dump_buf("cur_page_data_.compressed_buf_ = ",
        // cur_page_data_.compressed_buf_, cur_page_data_.compressed_size_);
        if (RET_FAIL(pages_data.write_buf(cur_page_data_.compressed_buf_,
                                          cur_page_data_.compressed_size_))) {
        }
    }
    // DEBUG_print_byte_stream("In PageWriter::write_to_chunk, after writer page
    // data: pages_data = ", pages_data); std::cout <<
    // "PageWriter::write_to_chunk after write_data. pos=" <<
    // pages_data.total_size() << std::endl;
#if DEBUG_SE
    std::cout << "PageWriter write_to_chunk: ret=" << ret
              << ", flags(HSD)=" << write_header << write_statistic
              << write_data_to_chunk_data
              << ", uncompressed_size=" << cur_page_data_.uncompressed_size_
              << ", compressed_size=" << cur_page_data_.compressed_size_
              << ", write_header=" << write_header
              << ", statistic_=" << statistic_->to_string() << std::endl;
#endif
    return ret;
}

}  // end namespace storage