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

#include <fcntl.h>

#include <memory>

#include "common/device_id.h"
#include "common/global.h"
#include "common/logger/elog.h"
#include "writer/chunk_writer.h"

using namespace common;

namespace storage {

#if 0
#define OFFSET_DEBUG(msg)                \
    std::cout << "OFFSET_DEBUG: " << msg \
              << ". offset=" << write_stream_.total_size() << std::endl
#else
#define OFFSET_DEBUG(msg) void(msg)
#endif

int TsFileIOWriter::init(WriteFile* write_file) {
    int ret = E_OK;
    const uint32_t page_size = 1024;
    meta_allocator_.init(page_size, MOD_TSFILE_WRITER_META);
    chunk_meta_count_ = 0;
    recovery_chunk_meta_prefix_.clear();
    destroyed_ = false;
    file_ = write_file;
    return ret;
}

void TsFileIOWriter::destroy() {
    if (destroyed_) {
        return;
    }
    // Recovery attaches a prefix of ChunkGroupMeta; device_id and chunk stats
    // in that snapshot live in reader/recovery memory. After open, new chunks
    // may be pushed into the same ChunkGroupMeta (same device); only those
    // appended ChunkMeta need statistic_->destroy() (see
    // recovery_chunk_meta_prefix_).
    for (auto iter = chunk_group_meta_list_.begin();
         iter != chunk_group_meta_list_.end(); iter++) {
        ChunkGroupMeta* cgm = iter.get();
        auto prefix_it = recovery_chunk_meta_prefix_.find(cgm);
        const bool is_recovery_cgm =
            chunk_group_meta_from_recovery_ && cgm != nullptr &&
            prefix_it != recovery_chunk_meta_prefix_.end();
        uint32_t recovered_cm_count = is_recovery_cgm ? prefix_it->second : 0;

        if (!is_recovery_cgm) {
            if (cgm != nullptr && cgm->device_id_) {
                cgm->device_id_.reset();
            }
        }

        if (cgm == nullptr) {
            continue;
        }
        uint32_t cm_idx = 0;
        for (auto chunk_meta = cgm->chunk_meta_list_.begin();
             chunk_meta != cgm->chunk_meta_list_.end();
             chunk_meta++, cm_idx++) {
            if (chunk_meta.get() == nullptr ||
                chunk_meta.get()->statistic_ == nullptr) {
                continue;
            }
            if (is_recovery_cgm && cm_idx < recovered_cm_count) {
                continue;
            }
            chunk_meta.get()->statistic_->destroy();
        }
    }

    if (cur_chunk_meta_ != nullptr && cur_chunk_meta_->statistic_ != nullptr) {
        cur_chunk_meta_->statistic_->destroy();
        cur_chunk_meta_ = nullptr;
    }

    meta_allocator_.destroy();
    write_stream_.destroy();
    if (write_file_created_ && file_ != nullptr) {
        delete file_;
        file_ = nullptr;
    }
    destroyed_ = true;
}

int TsFileIOWriter::start_file() {
    int ret = E_OK;
    if (RET_FAIL(write_buf(MAGIC_STRING_TSFILE, MAGIC_STRING_TSFILE_LEN))) {
    } else if (RET_FAIL(write_byte(VERSION_NUM_BYTE))) {
    } else if (RET_FAIL(flush_stream_to_file())) {
    }
    return ret;
}

int TsFileIOWriter::start_flush_chunk_group(
    std::shared_ptr<IDeviceID> device_name, bool is_aligned) {
    int ret = write_byte(CHUNK_GROUP_HEADER_MARKER);
    if (ret != common::E_OK) {
        return ret;
    }
    ret = device_name->serialize(write_stream_);
    if (ret != common::E_OK) {
        return ret;
    }
    is_aligned_ = is_aligned;
    cur_device_name_ = device_name;
    ASSERT(cur_chunk_group_meta_ == nullptr);
    use_prev_alloc_cgm_ = false;
    for (auto iter = chunk_group_meta_list_.begin();
         iter != chunk_group_meta_list_.end(); iter++) {
        if (*iter.get()->device_id_ == *cur_device_name_) {
            use_prev_alloc_cgm_ = true;
            cur_chunk_group_meta_ = iter.get();
            break;
        }
    }
    if (!use_prev_alloc_cgm_) {
        void* buf = meta_allocator_.alloc(sizeof(*cur_chunk_group_meta_));
        if (IS_NULL(buf)) {
            ret = E_OOM;
        } else {
            cur_chunk_group_meta_ = new (buf) ChunkGroupMeta(&meta_allocator_);
            cur_chunk_group_meta_->init(device_name);
        }
    }
    return ret;
}

int TsFileIOWriter::start_flush_chunk(ByteStream& chunk_data,
                                      ColumnSchema& col_schema,
                                      int32_t num_of_pages) {
    std::string measurement_name = col_schema.get_measurement_name_str();
    return start_flush_chunk(chunk_data, measurement_name,
                             col_schema.data_type_, col_schema.encoding_,
                             col_schema.compression_, num_of_pages);
}

int TsFileIOWriter::start_flush_chunk(common::ByteStream& chunk_data,
                                      std::string& measurement_name,
                                      common::TSDataType data_type,
                                      common::TSEncoding encoding,
                                      common::CompressionType compression,
                                      int32_t num_of_pages) {
    int ret = E_OK;

    // Step 1. record chunk meta
    const int mask = 0;  // for common chunk
    ASSERT(cur_chunk_meta_ == nullptr);
    void* buf1 = meta_allocator_.alloc(sizeof(*cur_chunk_meta_));
    void* buf2 = meta_allocator_.alloc(get_typed_statistic_sizeof(data_type));
    if (IS_NULL(buf1) || IS_NULL(buf2)) {
        ret = E_OOM;
    } else {
        cur_chunk_meta_ = new (buf1) ChunkMeta();
        Statistic* chunk_statistic_copy =
            placement_new_statistic(data_type, buf2);
        String mname((char*)measurement_name.c_str(),
                     strlen(measurement_name.c_str()));
        ret = cur_chunk_meta_->init(mname, data_type, cur_file_position(),
                                    chunk_statistic_copy, mask, encoding,
                                    compression, meta_allocator_);
    }

    // Step 2. serialize chunk header to write_stream_
    if (IS_SUCC(ret)) {
        ChunkHeader chunk_header;
        chunk_header.measurement_name_ = measurement_name,
        chunk_header.data_size_ = chunk_data.total_size();
        chunk_header.data_type_ = data_type;
        chunk_header.compression_type_ = compression;
        chunk_header.encoding_type_ = encoding;
        chunk_header.num_of_pages_ = num_of_pages;
        chunk_header.chunk_type_ =
            (num_of_pages <= 1 ? ONLY_ONE_PAGE_CHUNK_HEADER_MARKER
                               : CHUNK_HEADER_MARKER);
        if (is_aligned_) {
            chunk_header.chunk_type_ |= (data_type == common::VECTOR)
                                            ? TIME_COLUMN_MASK
                                            : VALUE_COLUMN_MASK;
        }
        OFFSET_DEBUG("start flush chunk header");
        ret = chunk_header.serialize_to(write_stream_);
        OFFSET_DEBUG("after flush chunk header");
#if DEBUG_SE
        std::cout << "TsFileIOWriter writer ChunkHeader: " << chunk_header
                  << std::endl;
#endif
    }
    return ret;
}

int TsFileIOWriter::flush_chunk(ByteStream& chunk_data) {
    int ret = E_OK;
    if (RET_FAIL(write_chunk_data(chunk_data))) {
        // log_err("writer chunk data error, ret=%d", ret);
    } else if (RET_FAIL(flush_stream_to_file())) {
        // log_err("flush stream error, ret=%d", ret);
    }
    return ret;
}

int TsFileIOWriter::write_chunk_data(ByteStream& chunk_data) {
    OFFSET_DEBUG("before writer chunk data");
    // may print chunk_data here for debug.
#if DEBUG_SE
    DEBUG_print_byte_stream("WriteChunkData chunk_data=", chunk_data);
    std::cout << "WriteChunkData: write_stream_.total_size="
              << write_stream_.total_size()
              << ", chunk_data.total_size=" << chunk_data.total_size()
              << std::endl;
#endif
    int ret = merge_byte_stream(write_stream_, chunk_data, true);
    OFFSET_DEBUG("before writer chunk data");
    return ret;
}

int TsFileIOWriter::end_flush_chunk(Statistic* chunk_statistic) {
    int ret = E_OK;
    chunk_meta_count_++;
    cur_chunk_meta_->clone_statistic_from(chunk_statistic);
    if (RET_FAIL(cur_chunk_group_meta_->push(cur_chunk_meta_))) {
    } else {
        cur_chunk_meta_ = nullptr;
    }
    return ret;
}

int TsFileIOWriter::end_flush_chunk_group(bool is_aligned) {
    if (generate_table_schema_) {
        schema_->update_table_schema(cur_chunk_group_meta_);
    }

    if (use_prev_alloc_cgm_) {
        cur_chunk_group_meta_ = nullptr;
        return common::E_OK;
    }
    int ret = chunk_group_meta_list_.push_back(cur_chunk_group_meta_);
    cur_chunk_group_meta_ = nullptr;
    return ret;
}

int TsFileIOWriter::end_file() {
    int ret = E_OK;
    if (file_->get_fd() == -1) {
        return E_OK;
    }
    OFFSET_DEBUG("before end file");
    if (RET_FAIL(write_log_index_range())) {
        std::cout << "writer range index error, ret =" << ret << std::endl;
    } else if (RET_FAIL(write_file_index())) {
        std::cout << "writer file index error, ret = " << ret << std::endl;
    } else if (RET_FAIL(write_file_footer())) {
        std::cout << "writer file footer error, ret = " << ret << std::endl;
    } else if (RET_FAIL(sync_file())) {
        std::cout << "sync file error, ret = " << ret << std::endl;
    } else if (RET_FAIL(close_file())) {
        std::cout << "close file error, ret = " << ret << std::endl;
    }
    return ret;
}

int TsFileIOWriter::write_log_index_range() {
    int ret = E_OK;
    // Currently, Java IoTDB leaves it as default value.
    const int64_t min_plan_index = 0;  // 0x7fffffffffffffff;
    const int64_t max_plan_index = 0;  // 0x8000000000000000;
    //  OFFSET_DEBUG("before writer log index");
    if (RET_FAIL(write_byte(OPERATION_INDEX_RANGE))) {
        std::cout << "writer byte error " << ret << std::endl;
    } else if (RET_FAIL(SerializationUtil::write_i64(min_plan_index,
                                                     write_stream_))) {
        std::cout << "min index error " << ret << std::endl;
    } else if (RET_FAIL(SerializationUtil::write_i64(max_plan_index,
                                                     write_stream_))) {
        std::cout << "max index error " << ret << std::endl;
    }
    return ret;
}

#if DEBUG_SE
void debug_print_chunk_group_meta(ChunkGroupMeta* cgm) {
    std::cout << "ChunkGroupMeta = {device_id_=" << cgm->device_id_
              << ", chunk_meta_list_={";
    SimpleList<ChunkMeta*>::Iterator cm_it = cgm->chunk_meta_list_.begin();
    for (; cm_it != cgm->chunk_meta_list_.end(); cm_it++) {
        ChunkMeta* cm = cm_it.get();
        std::cout << "{measurement=" << cm->measurement_name_
                  << ", offset_of_chunk_header=" << cm->offset_of_chunk_header_
                  << ", statistic=" << cm->statistic_->to_string() << "}}}";
    }
    std::cout << std::endl;
}

void debug_print_chunk_group_meta_list(
    common::SimpleList<ChunkGroupMeta*>& cgm_list) {
    SimpleList<ChunkGroupMeta*>::Iterator cgm_it = cgm_list.begin();
    for (; cgm_it != cgm_list.end(); cgm_it++) {
        ChunkGroupMeta* cgm = cgm_it.get();
        debug_print_chunk_group_meta(cgm);
    }
}
#endif

// TODO better memory management.
int TsFileIOWriter::write_file_index() {
#if DEBUG_SE
    debug_print_chunk_group_meta_list(chunk_group_meta_list_);
#endif

    int ret = E_OK;
    int64_t meta_offset = 0;
    BloomFilter filter;

    // TODO: better memory manage for this while-loop, cur_index_node_queue
    FileIndexWritingMemManager writing_mm;
    std::shared_ptr<IDeviceID> device_id;
    String measurement_name;
    TimeseriesIndex ts_index;
    std::shared_ptr<IDeviceID> prev_device_id;
    int entry_count_in_cur_device = 0;
    std::shared_ptr<IMetaIndexEntry> meta_index_entry = nullptr;
    std::shared_ptr<MetaIndexNode> cur_index_node = nullptr;
    SimpleList<std::shared_ptr<MetaIndexNode>>* cur_index_node_queue = nullptr;
    DeviceNodeMap device_map;

    TSMIterator tsm_iter(chunk_group_meta_list_);

    if (RET_FAIL(write_separator_marker(meta_offset))) {
        return ret;
    } else if (RET_FAIL(init_bloom_filter(filter))) {
        return ret;
    }

    if (RET_FAIL(tsm_iter.init())) {
        return ret;
    }

    while (IS_SUCC(ret) && tsm_iter.has_next()) {
        ts_index.reset();  // TODO reuse
        if (RET_FAIL(
                tsm_iter.get_next(device_id, measurement_name, ts_index))) {
            break;
        }
#if DEBUG_SE
        std::cout << "tsm_iter get next = {device_name=" << device_id
                  << ", measurement_name=" << measurement_name
                  << ", ts_index=" << ts_index << std::endl;
#endif

        // prepare if it is an entry of a new device
        if (prev_device_id == nullptr || prev_device_id != device_id) {
            if (prev_device_id != nullptr) {
                if (RET_FAIL(add_cur_index_node_to_queue(
                        cur_index_node, cur_index_node_queue))) {
                } else if (RET_FAIL(add_device_node(device_map, prev_device_id,
                                                    cur_index_node_queue,
                                                    writing_mm))) {
                }
            }
            if (IS_SUCC(ret)) {
                destroy_node_list(cur_index_node_queue);
                if (RET_FAIL(alloc_meta_index_node_queue(
                        writing_mm, cur_index_node_queue))) {
                } else if (RET_FAIL(alloc_and_init_meta_index_node(
                               writing_mm, cur_index_node, LEAF_MEASUREMENT))) {
                }
            }
            prev_device_id = device_id;
            entry_count_in_cur_device = 0;
        }

        // pick an entry as index entry every max_degree_of_index_node entries.
        if (IS_SUCC(ret) && entry_count_in_cur_device %
                                    g_config_value_.max_degree_of_index_node_ ==
                                0) {
            if (cur_index_node->is_full()) {
                if (RET_FAIL(add_cur_index_node_to_queue(
                        cur_index_node, cur_index_node_queue))) {
                } else if (RET_FAIL(alloc_and_init_meta_index_node(
                               writing_mm, cur_index_node, LEAF_MEASUREMENT))) {
                }
            }

            if (IS_SUCC(ret)) {
                if (RET_FAIL(alloc_and_init_meta_index_entry(
                        writing_mm, meta_index_entry, measurement_name))) {
                } else if (RET_FAIL(
                               cur_index_node->push_entry(meta_index_entry))) {
                }
            }
        }

        if (IS_SUCC(ret)) {
            OFFSET_DEBUG("before ts_index written");
            common::String tmp_device_name;
            tmp_device_name.dup_from(device_id->get_device_name(),
                                     meta_allocator_);
            // Time column also need add to bloom filter.
            ret = filter.add_path_entry(tmp_device_name, measurement_name);

            if (RET_FAIL(ts_index.serialize_to(write_stream_))) {
            } else {
#if DEBUG_SE
                std::cout << "ts_index.serialize. ts_index=" << ts_index
                          << " file_pos=" << cur_file_position() << std::endl;
#endif
                entry_count_in_cur_device++;
                // add_ts_time_index_entry(ts_index);
            }
        }
    }  // end while
    if (ret == E_NO_MORE_DATA) {
        ret = E_OK;
    }
    ASSERT(ret == E_OK);
    if (IS_SUCC(ret) && cur_index_node != nullptr &&
        cur_index_node_queue != nullptr) {  // iter finish
        ASSERT(cur_index_node != nullptr);
        ASSERT(cur_index_node_queue != nullptr);
        if (RET_FAIL(add_cur_index_node_to_queue(cur_index_node,
                                                 cur_index_node_queue))) {
        } else if (RET_FAIL(add_device_node(device_map, prev_device_id,
                                            cur_index_node_queue,
                                            writing_mm))) {
        }
    }

    if (IS_SUCC(ret)) {
        TsFileMeta tsfile_meta;
        tsfile_meta.meta_offset_ = meta_offset;
        tsfile_meta.bloom_filter_ = &filter;
        // split device by table
        std::map<std::string, DeviceNodeMap> table_device_nodes_map;
        for (const auto& entry : device_map) {
            std::string table_name = entry.first->get_table_name();
            auto& table_map = table_device_nodes_map[table_name];
            if (table_map.empty() ||
                table_map.find(entry.first) == table_map.end()) {
                table_map[entry.first] = entry.second;
            }
        }
        std::map<std::string, std::shared_ptr<MetaIndexNode>> table_nodes_map;
        for (auto& entry : table_device_nodes_map) {
            auto meta_index_node =
                std::make_shared<MetaIndexNode>(&meta_allocator_);
            build_device_level(entry.second, meta_index_node, writing_mm);
            table_nodes_map[entry.first] = meta_index_node;
        }
        tsfile_meta.table_metadata_index_node_map_ = table_nodes_map;
        tsfile_meta.table_schemas_ = schema_->table_schema_map_;
        tsfile_meta.tsfile_properties_.insert(
            std::make_pair("encryptLevel", new std::string(encrypt_level_)));
        tsfile_meta.tsfile_properties_.insert(
            std::make_pair("encryptType", new std::string(encrypt_type_)));
        tsfile_meta.tsfile_properties_.insert(
            std::make_pair("encryptKey", nullptr));
#if DEBUG_SE
        auto tsfile_meta_offset = write_stream_.total_size();
#endif
        auto total_write_size = tsfile_meta.serialize_to(write_stream_);
        if (RET_FAIL(common::SerializationUtil::write_i32(total_write_size,
                                                          write_stream_))) {
            return ret;
        }
        tsfile_meta.bloom_filter_ = nullptr;
#if DEBUG_SE
        std::cout << "writer tsfile_meta: " << tsfile_meta
                  << ", tsfile_meta_offset=" << tsfile_meta_offset
                  << ", size=" << total_write_size << std::endl;
        DEBUG_print_byte_stream("byte_stream", write_stream_);
#endif
    }
    destroy_node_list(cur_index_node_queue);
    return ret;
}

int TsFileIOWriter::build_device_level(DeviceNodeMap& device_map,
                                       std::shared_ptr<MetaIndexNode>& ret_root,
                                       FileIndexWritingMemManager& wmm) {
    int ret = E_OK;

    SimpleList<std::shared_ptr<MetaIndexNode>> node_queue(
        1024,
        MOD_TSFILE_WRITER_META);  // FIXME
    DeviceNodeMapIterator device_map_iter;

    std::shared_ptr<MetaIndexNode> cur_index_node = nullptr;
    if (RET_FAIL(
            alloc_and_init_meta_index_node(wmm, cur_index_node, LEAF_DEVICE))) {
        return ret;
    }

    for (device_map_iter = device_map.begin();
         device_map_iter != device_map.end() && IS_SUCC(ret);
         device_map_iter++) {
        auto device_id = device_map_iter->first;
        std::shared_ptr<IMetaIndexEntry> entry = nullptr;
        if (cur_index_node->is_full()) {
            cur_index_node->end_offset_ = cur_file_position();
#if DEBUG_SE
            std::cout << "TsFileIOWriter::build_device_level, cur_index_node="
                      << *cur_index_node << std::endl;
#endif
            if (RET_FAIL(node_queue.push_back(cur_index_node))) {
            } else if (RET_FAIL(alloc_and_init_meta_index_node(
                           wmm, cur_index_node, LEAF_DEVICE))) {
            }
        }
        if (RET_FAIL(alloc_and_init_meta_index_entry(wmm, entry, device_id))) {
        } else if (RET_FAIL(
                       device_map_iter->second->serialize_to(write_stream_))) {
        } else if (RET_FAIL(cur_index_node->push_entry(entry))) {
        }
    }  // end for
    if (IS_SUCC(ret)) {
        if (!cur_index_node->is_empty()) {
            cur_index_node->end_offset_ = cur_file_position();
            ret = node_queue.push_back(cur_index_node);
        }
    }

    if (IS_SUCC(ret)) {
        if (node_queue.size() > 0) {
            if (RET_FAIL(generate_root(&node_queue, ret_root, INTERNAL_DEVICE,
                                       wmm))) {
            }
        } else {
            ret_root = cur_index_node;
            ret_root->end_offset_ = cur_file_position();
            ret_root->node_type_ = LEAF_DEVICE;
        }
    }
    destroy_node_list(&node_queue);
    return ret;
}

int TsFileIOWriter::write_separator_marker(int64_t& meta_offset) {
    meta_offset = cur_file_position();
    return write_byte(SEPARATOR_MARKER);
}

int TsFileIOWriter::init_bloom_filter(BloomFilter& filter) {
    int ret = E_OK;
    int32_t path_count = get_path_count(chunk_group_meta_list_);
    if (RET_FAIL(filter.init(
            g_config_value_.tsfile_index_bloom_filter_error_percent_,
            path_count))) {
    }
    return ret;
}

int32_t TsFileIOWriter::get_path_count(
    common::SimpleList<ChunkGroupMeta*>& cgm_list) {
    int32_t path_count = 0;
    String prev_measurement;

    SimpleList<ChunkGroupMeta*>::Iterator cgm_it = cgm_list.begin();
    for (; cgm_it != cgm_list.end(); cgm_it++) {
        ChunkGroupMeta* cgm = cgm_it.get();
        SimpleList<ChunkMeta*>::Iterator cm_it = cgm->chunk_meta_list_.begin();
        for (; cm_it != cgm->chunk_meta_list_.end(); cm_it++) {
            ChunkMeta* cm = cm_it.get();
            // TODO currently, we do not have multi chunks of same timeseries in
            // tsfile.
            if (!cm->measurement_name_.equal_to(prev_measurement)) {
                path_count++;
                prev_measurement = cm->measurement_name_;
            }
        }
    }
    return path_count;
}

int TsFileIOWriter::write_file_footer() {
    int ret = E_OK;
    if (RET_FAIL(write_buf(MAGIC_STRING_TSFILE, MAGIC_STRING_TSFILE_LEN))) {
    } else if (RET_FAIL(flush_stream_to_file())) {
    }
    return ret;
}

int TsFileIOWriter::alloc_and_init_meta_index_entry(
    FileIndexWritingMemManager& wmm,
    std::shared_ptr<IMetaIndexEntry>& ret_entry,
    const std::shared_ptr<IDeviceID>& device_id) {
    void* buf = wmm.pa_.alloc(sizeof(DeviceMetaIndexEntry));
    if (IS_NULL(buf)) {
        return E_OOM;
    }
    auto entry_ptr = static_cast<DeviceMetaIndexEntry*>(buf);
    new (entry_ptr) DeviceMetaIndexEntry(device_id, cur_file_position());
    ret_entry = std::shared_ptr<IMetaIndexEntry>(
        entry_ptr, IMetaIndexEntry::self_destructor);
#if DEBUG_SE
    std::cout << "alloc_and_init_meta_index_entry, MetaIndexEntry="
              << *ret_entry << std::endl;
#endif
    return E_OK;
}

int TsFileIOWriter::alloc_and_init_meta_index_entry(
    FileIndexWritingMemManager& wmm,
    std::shared_ptr<IMetaIndexEntry>& ret_entry, common::String& name) {
    void* buf = wmm.pa_.alloc(sizeof(MeasurementMetaIndexEntry));
    if (IS_NULL(buf)) {
        return E_OOM;
    }
    auto entry_ptr = static_cast<MeasurementMetaIndexEntry*>(buf);
    new (entry_ptr)
        MeasurementMetaIndexEntry(name, cur_file_position(), wmm.pa_);
    ret_entry = std::shared_ptr<IMetaIndexEntry>(
        entry_ptr, IMetaIndexEntry::self_destructor);
#if DEBUG_SE
    std::cout << "alloc_and_init_meta_index_entry, MetaIndexEntry="
              << *ret_entry << std::endl;
#endif
    return E_OK;
}

int TsFileIOWriter::alloc_and_init_meta_index_node(
    FileIndexWritingMemManager& wmm, std::shared_ptr<MetaIndexNode>& ret_node,
    MetaIndexNodeType node_type) {
    //    void *buf = wmm.pa_.alloc(sizeof(MetaIndexNode));
    //    if (IS_NULL(buf)) {
    //        return E_OOM;
    //    }
    //    auto *node_ptr = new (buf) MetaIndexNode(&wmm.pa_);
    //    node_ptr->node_type_ = node_type;
    //    ret_node = std::shared_ptr<MetaIndexNode>(node_ptr, [](MetaIndexNode
    //    *ptr) {
    //        if (ptr) {
    //            ptr->~MetaIndexNode();
    //        }
    //    });
    ret_node = std::make_shared<MetaIndexNode>(&wmm.pa_);
    ret_node->node_type_ = node_type;
    wmm.all_index_nodes_.push_back(ret_node);
    return E_OK;
}

int TsFileIOWriter::add_cur_index_node_to_queue(
    std::shared_ptr<MetaIndexNode> node,
    SimpleList<std::shared_ptr<MetaIndexNode>>* queue) const {
    node->end_offset_ = cur_file_position();
#if DEBUG_SE
    std::cout << "add_cur_index_node_to_queue, node=" << *node
              << ", set offset=" << cur_file_position() << std::endl;
#endif
    return queue->push_back(node);
}

int TsFileIOWriter::alloc_meta_index_node_queue(
    FileIndexWritingMemManager& wmm,
    SimpleList<std::shared_ptr<MetaIndexNode>>*& queue) {
    void* buf = wmm.pa_.alloc(sizeof(*queue));
    if (IS_NULL(buf)) {
        return E_OOM;
    }
    queue = new (buf) SimpleList<std::shared_ptr<MetaIndexNode>>(&wmm.pa_);
    return E_OK;
}

int TsFileIOWriter::add_device_node(
    DeviceNodeMap& device_map, std::shared_ptr<IDeviceID> device_id,
    common::SimpleList<std::shared_ptr<MetaIndexNode>>*
        measurement_index_node_queue,
    FileIndexWritingMemManager& wmm) {
    ASSERT(measurement_index_node_queue->size() > 0);
    int ret = E_OK;
    auto find_iter = device_map.find(device_id);
    if (find_iter != device_map.end()) {
        return E_ALREADY_EXIST;
    }

    std::shared_ptr<MetaIndexNode> root = nullptr;
    if (RET_FAIL(generate_root(measurement_index_node_queue, root,
                               INTERNAL_MEASUREMENT, wmm))) {
    } else {
        std::pair<DeviceNodeMapIterator, bool> ins_res =
            device_map.insert(std::make_pair(device_id, root));
        if (!ins_res.second) {
            ASSERT(false);
        }
    }
    return ret;
}

void TsFileIOWriter::set_generate_table_schema(bool generate_table_schema) {
    generate_table_schema_ = generate_table_schema;
}

int TsFileIOWriter::generate_root(
    SimpleList<std::shared_ptr<MetaIndexNode>>* node_queue,
    std::shared_ptr<MetaIndexNode>& root_node, MetaIndexNodeType node_type,
    FileIndexWritingMemManager& wmm) {
    int ret = E_OK;

    ASSERT(node_queue->size() > 0);
    if (node_queue->size() == 1) {
        root_node = node_queue->front();
        return ret;
    }

    const uint32_t LIST_PAGE_SIZE = 256;
    const AllocModID mid = MOD_TSFILE_WRITER_META;
    SimpleList<std::shared_ptr<MetaIndexNode>> list_x(LIST_PAGE_SIZE, mid);
    SimpleList<std::shared_ptr<MetaIndexNode>> list_y(LIST_PAGE_SIZE, mid);

    if (RET_FAIL(clone_node_list(node_queue, &list_x))) {
        return ret;
    }

    common::SimpleList<std::shared_ptr<MetaIndexNode>>* from = &list_x;
    common::SimpleList<std::shared_ptr<MetaIndexNode>>* to = &list_y;

    std::shared_ptr<MetaIndexNode> cur_index_node = nullptr;
    if (RET_FAIL(
            alloc_and_init_meta_index_node(wmm, cur_index_node, node_type))) {
    }
    while (IS_SUCC(ret)) {
        for (auto iter = to->begin(); iter != to->end(); iter++) {
            iter.get().reset();
        }
        to->clear();
        SimpleList<std::shared_ptr<MetaIndexNode>>::Iterator from_iter;
        for (from_iter = from->begin();
             IS_SUCC(ret) && from_iter != from->end(); from_iter++) {
            auto iter_node = from_iter.get();
            std::shared_ptr<IMetaIndexEntry> entry = nullptr;
            auto first_child = iter_node->peek();
            if (const auto derived_entry =
                    std::dynamic_pointer_cast<DeviceMetaIndexEntry>(
                        first_child)) {
                ret = alloc_and_init_meta_index_entry(
                    wmm, entry, derived_entry->device_id_);
            } else if (const auto measurement_entry =
                           std::dynamic_pointer_cast<MeasurementMetaIndexEntry>(
                               first_child)) {
                ret = alloc_and_init_meta_index_entry(wmm, entry,
                                                      measurement_entry->name_);
            } else {
                ret = E_INVALID_DATA_POINT;
            }
            if (IS_FAIL(ret)) {
                continue;
            }

            if (cur_index_node->is_full()) {
                cur_index_node->end_offset_ = cur_file_position();
                if (RET_FAIL(to->push_back(cur_index_node))) {
                } else {
#if DEBUG_SE
                    std::cout
                        << "generate root, alloc_and_init_meta_index_node. "
                           "cur_index_node="
                        << *cur_index_node << std::endl;
#endif
                    if (RET_FAIL(alloc_and_init_meta_index_node(
                            wmm, cur_index_node, node_type))) {
                    }
                }
            }
            if (IS_SUCC(ret)) {
                if (RET_FAIL(cur_index_node->push_entry(entry))) {
                } else {
                    OFFSET_DEBUG("before writer index_node in generate_root");
                    ret = iter_node->serialize_to(write_stream_);
                    OFFSET_DEBUG("after writer index_node in generate_root");
                }
            }
        }  // end for
        if (IS_SUCC(ret)) {
            if (!cur_index_node->is_empty()) {
                cur_index_node->end_offset_ = cur_file_position();
                if (RET_FAIL(to->push_back(cur_index_node))) {
                }
#if DEBUG_SE
                std::cout << "genereate root 2, "
                             "alloc_and_init_meta_index_node. cur_index_node="
                          << *cur_index_node << std::endl;
#endif
                if (RET_FAIL(alloc_and_init_meta_index_node(wmm, cur_index_node,
                                                            node_type))) {
                }
            }
        }
        if (IS_SUCC(ret)) {
            ASSERT(from->size() > to->size());
            if (to->size() == 1) {
                root_node = to->front();
                break;
            } else {
                swap_list(from, to);
            }
        }
    }  // end while
    destroy_node_list(&list_x);
    destroy_node_list(&list_y);
    return ret;
}

void TsFileIOWriter::destroy_node_list(
    common::SimpleList<std::shared_ptr<MetaIndexNode>>* list) {
    if (list) {
        for (auto iter = list->begin(); iter != list->end(); iter++) {
            if (iter.get()) {
                iter.get().reset();
            }
        }
    }
}

int TsFileIOWriter::clone_node_list(
    SimpleList<std::shared_ptr<MetaIndexNode>>* src,
    SimpleList<std::shared_ptr<MetaIndexNode>>* dest) {
    int ret = E_OK;
    SimpleList<std::shared_ptr<MetaIndexNode>>::Iterator it;
    for (it = src->begin(); IS_SUCC(ret) && it != src->end(); it++) {
        ret = dest->push_back(it.get());
    }
    return ret;
}

int TsFileIOWriter::restore_recovered_file_position(int64_t recovered_size) {
    if (recovered_size < 0) {
        return E_INVALID_ARG;
    }
    file_base_offset_ = recovered_size;
    return E_OK;
}

// #if DEBUG_SE
// void DEBUG_print_byte_stream_buf(const char *tag,
//                                  const char *buf,
//                                  const uint32_t len,
//                                  bool reset_print)
// {
//   static int print_count = 0;
//   if (reset_print) {
//     print_count = 0;
//   }
//
//   for (uint32_t i = 0; i < len; i++) {
//     if (print_count++ % 16 == 0) {
//       printf("\n%s: BUF=", tag);
//     }
//     printf("%02x ", (uint8_t)buf[i]);
//   }
//   printf("\n\n");
// }
// #endif

/*
 * TODO:
 * when finish flushing stream to file, reclaim memory used by stream
 */
int TsFileIOWriter::flush_stream_to_file() {
    int ret = E_OK;
    while (true) {
        ByteStream::Buffer b =
            write_stream_consumer_.get_next_buf(write_stream_);
        if (b.buf_ == nullptr) {
            break;
        }
        if (b.len_ > 0 && RET_FAIL(file_->write(b.buf_, b.len_))) {
            break;
        }
    }

    write_stream_.purge_prev_pages();

    return ret;
}

}  // namespace storage
