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

using namespace common;

namespace storage {

int DataRun::remove_tsfile(const FileID& to_remove) {
    ASSERT(run_type_ == DRT_TSFILE);
    int ret = E_OK;

    SimpleList<OpenFile*>::Iterator it = tsfile_list_.begin();
    OpenFile* of = nullptr;
    for (; it != tsfile_list_.end(); it++) {
        OpenFile* cur = it.get();
        if (cur->get_file_id() == to_remove) {
            of = cur;
            break;
        }
    }
    if (of != nullptr) {
        ret = tsfile_list_.remove(of);
    }
    return ret;
}

int DataRun::get_next(TsBlock* ret_block, TimeRange& ret_time_range,
                      bool alloc_tsblock) {
    if (run_type_ == DRT_TVLIST) {
        return tvlist_get_next(ret_block, ret_time_range, alloc_tsblock);
    } else {
        ASSERT(run_type_ == DRT_TSFILE);
        return tsfile_get_next(ret_block, ret_time_range, alloc_tsblock);
    }
}

TsBlock* DataRun::alloc_tsblock() {
    tuple_desc_.reset();
    // TODO default config of time_cd
    tuple_desc_.push_back(g_time_column_schema);
    tuple_desc_.push_back(*col_schema_);
    return (new TsBlock(&tuple_desc_));
}

int DataRun::tvlist_get_next(TsBlock* ret_block, TimeRange& ret_time_range,
                             bool tsblock) {
    // TODO @tsblock
    int ret = E_OK;
    if (UNLIKELY(!tvlist_list_iter_.is_inited())) {
        tvlist_list_iter_ = tvlist_list_.begin();
    }

    while (true) {
        if (tvlist_list_iter_ == tvlist_list_.end()) {
            return E_NO_MORE_DATA;
        } else {
            SeqTVListBase* tvlist_base = tvlist_list_iter_.get();
            if (!tvlist_base->is_immutable()) {
                tvlist_base->lock();
            }
            ret = fill_tsblock_from_tvlist(tvlist_base, ret_block,
                                           ret_time_range);
            if (ret == E_NO_MORE_DATA) {
                ret = E_OK;
            }
            if (!tvlist_base->is_immutable()) {
                tvlist_base->unlock();
            }
            tvlist_list_iter_++;
            return ret;
        }
    }
    return ret;
}

int DataRun::fill_tsblock_from_tvlist(SeqTVListBase* tvlist, TsBlock* ret_block,
                                      TimeRange& ret_time_range) {
    int ret = E_OK;
    switch (col_schema_->data_type_) {
        case common::BOOLEAN:
            ret = fill_tsblock_from_typed_tvlist<bool>(tvlist, ret_block,
                                                       ret_time_range);
            break;
        case common::INT32:
            ret = fill_tsblock_from_typed_tvlist<int32_t>(tvlist, ret_block,
                                                          ret_time_range);
            break;
        case common::INT64:
            ret = fill_tsblock_from_typed_tvlist<int64_t>(tvlist, ret_block,
                                                          ret_time_range);
            break;
        case common::FLOAT:
            ret = fill_tsblock_from_typed_tvlist<float>(tvlist, ret_block,
                                                        ret_time_range);
            break;
        case common::DOUBLE:
            ret = fill_tsblock_from_typed_tvlist<double>(tvlist, ret_block,
                                                         ret_time_range);
            break;
        default:
            ASSERT(false);
            break;
    }
    return ret;
}

template <typename T>
int DataRun::fill_tsblock_from_typed_tvlist(SeqTVListBase* tvlist,
                                            TsBlock* ret_block,
                                            TimeRange& ret_time_range) {
    int ret = E_OK;

    SeqTVList<T>* typed_tvlist = static_cast<SeqTVList<T>*>(tvlist);
    typename SeqTVList<T>::Iterator it;
    it = typed_tvlist->scan_without_lock();
    typename SeqTVList<T>::TV tv;

    // FIXME do not append all tvlist data into tsblock in one time.
    ret_time_range.start_time_ = typed_tvlist->time_at(0);
    RowAppender row_appender(ret_block);
#ifndef NDEBUG
    int count = 0;
#endif
    while (E_OK == (ret = it.next(tv))) {
        ret_time_range.end_time_ = tv.time_;
#ifndef NDEBUG
        std::cout << "DataRun::fill_tsblock_from_typed_tvlist: [" << count
                  << "] = <" << tv.time_ << ", " << tv.value_ << ">"
                  << std::endl;
#endif
        row_appender.add_row();
        row_appender.append(0, reinterpret_cast<char*>(&tv.time_),
                            sizeof(tv.time_));
        row_appender.append(1, reinterpret_cast<char*>(&tv.value_),
                            sizeof(tv.value_));
    }
    return ret;
}

int DataRun::reinit_io_reader(SimpleList<OpenFile*>::Iterator& it,
                              common::PageArena* pa) {
    int ret = E_OK;
    // maybe io_reader_ destroy before re-init
    OpenFile* open_file = it.get();
    io_reader_.reset();
    if (RET_FAIL(io_reader_.init(open_file->get_file_path()))) {
        ////log_err("io_reader init error, ret=%d, file_path=%s",
        // ret, open_file->get_file_path().c_str());
    } else {
        std::shared_ptr<IDeviceID> device_id =
            std::make_shared<StringArrayDeviceID>(
                col_schema_->get_device_name_str());
        std::string measurement_name = col_schema_->get_measurement_name_str();
        if (ssi_ != nullptr) {
            delete ssi_;
            ssi_ = nullptr;
        }
        if (RET_FAIL(
                io_reader_.alloc_ssi(device_id, measurement_name, ssi_, *pa))) {
        }
    }
    return ret;
}

int DataRun::tsfile_get_next(TsBlock* ret_tsblock, TimeRange& ret_time_range,
                             bool alloc_tsblock) {
    int ret = E_OK;
    if (UNLIKELY(!tsfile_list_iter_.is_inited())) {
        tsfile_list_iter_ = tsfile_list_.begin();
        if (tsfile_list_iter_ == tsfile_list_.end()) {  // all file iterated
            ret = E_NO_MORE_DATA;
        } else if (RET_FAIL(reinit_io_reader(tsfile_list_iter_))) {
        }
    }

    if (IS_SUCC(ret)) {
        // ret = io_reader_.get_next(*col_desc_, ret_tsblock, ret_time_range);
        ret = ssi_->get_next(ret_tsblock, alloc_tsblock);
        if (E_NO_MORE_DATA == ret) {  // current file reach end
            tsfile_list_iter_++;
            if (tsfile_list_iter_ == tsfile_list_.end()) {  // all file iterated
                ret = E_NO_MORE_DATA;
            } else if (RET_FAIL(reinit_io_reader(tsfile_list_iter_))) {
            }
        }
    }
    return ret;
}

DataRun* DataScanIterator::alloc_data_run(DataRunType run_type) {
    void* buf = page_arena_.alloc(sizeof(DataRun));
    if (IS_NULL(buf)) {
        return nullptr;
    }
    return (new (buf) DataRun(run_type, &col_schema_, &page_arena_));
}

#ifndef NDEBUG
void DataScanIterator::DEBUG_dump_data_run_list() {
    SimpleList<DataRun*>::Iterator it;
    std::cout << "\n/---- DEBUG_dump_data_run_list: size="
              << data_run_list_.size() << "----\\" << std::endl;
    int idx = 0;
    for (it = data_run_list_.begin(); it != data_run_list_.end(); it++) {
        std::cout << "[" << (idx++) << "]" << *it.get() << std::endl;
    }
    std::cout << "\\---- DEBUG_dump_data_run_list: size="
              << data_run_list_.size() << "----/\n"
              << std::endl;
}
#endif

int DataScanIterator::get_next(TsBlock* ret_block, bool alloc_tsblock) {
#ifndef NDEBUG
    DEBUG_dump_data_run_list();
#endif
    int ret = E_OK;

    if (UNLIKELY(!cursor_.is_inited())) {
        cursor_ = data_run_list_.begin();
    }

    while (true) {
        TimeRange time_range;
        DataRun* data_run = cursor_.get();
        ret = data_run->get_next(ret_block, time_range, alloc_tsblock);
        if (ret == E_OK) {
            return ret;
        } else if (ret == E_NO_MORE_DATA) {
            cursor_++;
            if (cursor_ == data_run_list_.end()) {
                return E_NO_MORE_DATA;
            }
        } else {
            // log_err("data run get next batch error, ret=%d", ret);
            break;
        }
    }
    return ret;
}

}  // end namespace storage