cpp/src/reader/tsfile_reader.cc

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

#include <stdexcept>
#include <string>
#include <utility>
#include <vector>

#include "common/allocator/byte_stream.h"
#include "common/schema.h"
#include "common/tsfile_common.h"
#include "file/local_random_access_read_file.h"
#include "filter/time_operator.h"
#include "tsfile_executor.h"

using namespace common;
using namespace storage;

namespace {

struct DeviceMetaEntry {
    std::shared_ptr<IDeviceID> device_id;
    int64_t start_offset;
    int64_t end_offset;
};

int parse_paths(const std::vector<std::string>& path_list,
                std::vector<Path>& parsed_paths) {
    try {
        parsed_paths.reserve(path_list.size());
        for (const auto& path : path_list) {
            parsed_paths.emplace_back(path, true);
        }
    } catch (const std::runtime_error&) {
        return E_INVALID_PATH;
    }
    return E_OK;
}

int get_all_device_entries(std::vector<DeviceMetaEntry>& entries,
                           std::shared_ptr<MetaIndexNode> index_node,
                           RandomAccessReadFile* read_file, PageArena& pa) {
    int ret = E_OK;
    if (index_node == nullptr) {
        return ret;
    }
    if (index_node->node_type_ == LEAF_DEVICE) {
        for (size_t i = 0; i < index_node->children_.size(); i++) {
            DeviceMetaEntry entry;
            entry.device_id = index_node->children_[i]->get_device_id();
            entry.start_offset = index_node->children_[i]->get_offset();
            entry.end_offset = (i + 1 < index_node->children_.size())
                                   ? index_node->children_[i + 1]->get_offset()
                                   : index_node->end_offset_;
            entries.push_back(entry);
        }
    } else {
        for (size_t idx = 0; idx < index_node->children_.size(); idx++) {
            auto meta_index_entry = index_node->children_[idx];
            int64_t start_offset = meta_index_entry->get_offset();
            int64_t end_offset = index_node->end_offset_;
            if (idx + 1 < index_node->children_.size()) {
                end_offset = index_node->children_[idx + 1]->get_offset();
            }
            ASSERT(end_offset - start_offset > 0);
            const int32_t read_size = (int32_t)(end_offset - start_offset);
            int32_t ret_read_len = 0;
            char* data_buf = (char*)pa.alloc(read_size);
            void* m_idx_node_buf = pa.alloc(sizeof(MetaIndexNode));
            if (IS_NULL(data_buf) || IS_NULL(m_idx_node_buf)) {
                return E_OOM;
            }
            auto* top_node_ptr = new (m_idx_node_buf) MetaIndexNode(&pa);
            auto top_node = std::shared_ptr<MetaIndexNode>(
                top_node_ptr, [](MetaIndexNode* ptr) {
                    if (ptr) {
                        ptr->~MetaIndexNode();
                    }
                });
            if (RET_FAIL(read_file->read(start_offset, data_buf, read_size,
                                         ret_read_len))) {
            } else if (ret_read_len != read_size) {
                ret = E_FILE_READ_ERR;
            } else if (RET_FAIL(top_node->device_deserialize_from(data_buf,
                                                                  read_size))) {
            } else {
                ret = get_all_device_entries(entries, top_node, read_file, pa);
            }
            if (ret != E_OK) {
                return ret;
            }
        }
    }
    return ret;
}

}  // namespace

namespace storage {
TsFileReader::TsFileReader()
    : read_file_(nullptr),
      tsfile_executor_(nullptr),
      table_query_executor_(nullptr) {
    tsfile_reader_meta_pa_.init(512, MOD_TSFILE_READER);
}

TsFileReader::~TsFileReader() { close(); }

int TsFileReader::open(const std::string& file_path) {
    std::unique_ptr<LocalRandomAccessReadFile> read_file(
        new LocalRandomAccessReadFile());
    int ret = E_OK;
    // Keep reader diagnostics in the caller's error channel.  Printing here
    // would leak an unstructured line to process stdout/stderr before the CLI
    // can attach its stable error text and exit code.
    if (RET_FAIL(read_file->open(file_path))) {
        return ret;
    }
    const unsigned char file_version = read_file->file_version();
    return open_source(std::move(read_file), file_version);
}

int TsFileReader::open(std::unique_ptr<RandomAccessReadFile> read_file) {
    if (read_file == nullptr || !read_file->is_opened()) {
        return E_INVALID_ARG;
    }
    unsigned char file_version = 0;
    const int ret = validate_tsfile(*read_file, &file_version);
    if (ret != E_OK) {
        return ret;
    }
    return open_source(std::move(read_file), file_version);
}

int TsFileReader::open_source(std::unique_ptr<RandomAccessReadFile> read_file,
                              unsigned char file_version) {
    close();
    read_file_ = std::move(read_file);
    file_version_ = file_version;
    tsfile_executor_ = new storage::TsFileExecutor();
    const int ret = tsfile_executor_->init(read_file_.get());
    if (ret != E_OK) {
        close();
    }
    return ret;
}

unsigned char TsFileReader::get_file_version() const { return file_version_; }

int TsFileReader::ensure_table_query_executor(int batch_size) {
    if (table_query_executor_ != nullptr &&
        table_query_executor_batch_size_ == batch_size) {
        return E_OK;
    }

    if (table_query_executor_ != nullptr) {
        delete table_query_executor_;
        table_query_executor_ = nullptr;
    }

    table_query_executor_ =
        new TableQueryExecutor(read_file_.get(), batch_size);
    table_query_executor_batch_size_ = batch_size;
    return E_OK;
}

int TsFileReader::close() {
    int ret = E_OK;
    if (tsfile_executor_ != nullptr) {
        delete tsfile_executor_;
        tsfile_executor_ = nullptr;
    }
    if (table_query_executor_ != nullptr) {
        delete table_query_executor_;
        table_query_executor_ = nullptr;
    }
    if (read_file_ != nullptr) {
        read_file_->close();
        read_file_.reset();
    }
    file_version_ = 0;
    return ret;
}

int TsFileReader::query(QueryExpression* qe, ResultSet*& ret_qds) {
    return tsfile_executor_->execute(qe, ret_qds);
}

int TsFileReader::query(std::vector<std::string>& path_list, int64_t start_time,
                        int64_t end_time, ResultSet*& result_set) {
    std::vector<Path> path_list_vec;
    int ret = parse_paths(path_list, path_list_vec);
    if (ret != E_OK) {
        return ret;
    }

    Filter* time_filter = new TimeBetween(start_time, end_time, false);
    Expression* exp =
        new storage::Expression(storage::GLOBALTIME_EXPR, time_filter);
    QueryExpression* query_expression =
        QueryExpression::create(path_list_vec, exp);
    ret = tsfile_executor_->execute(query_expression, result_set);
    return ret;
}

int TsFileReader::query(const std::string& table_name,
                        const std::vector<std::string>& columns_names,
                        int64_t start_time, int64_t end_time,
                        ResultSet*& result_set, int batch_size) {
    return this->query(table_name, columns_names, start_time, end_time,
                       result_set, nullptr, batch_size);
}

int TsFileReader::query(const std::string& table_name,
                        const std::vector<std::string>& columns_names,
                        int64_t start_time, int64_t end_time,
                        ResultSet*& result_set, Filter* tag_filter,
                        int batch_size) {
    return query(table_name, columns_names, start_time, end_time, 0, -1,
                 result_set, tag_filter, batch_size);
}

int TsFileReader::query(const std::string& table_name,
                        const std::vector<std::string>& columns_names,
                        int64_t start_time, int64_t end_time, int offset,
                        int limit, ResultSet*& result_set, Filter* tag_filter,
                        int batch_size) {
    int ret = E_OK;
    TsFileMeta* tsfile_meta = nullptr;
    if (RET_FAIL(tsfile_executor_->get_tsfile_meta(tsfile_meta))) {
        return ret;
    }
    auto schema_it = tsfile_meta->table_schemas_.find(to_lower(table_name));
    if (schema_it == tsfile_meta->table_schemas_.end() ||
        schema_it->second == nullptr) {
        return E_TABLE_NOT_EXIST;
    }
    if (offset < 0) {
        return E_INVALID_ARG;
    }
    if (limit < 0) {
        limit = -1;
    }

    Filter* time_filter = new TimeBetween(start_time, end_time, false);
    ensure_table_query_executor(batch_size);
    ret = table_query_executor_->query(to_lower(table_name), columns_names,
                                       time_filter, tag_filter, nullptr, offset,
                                       limit, result_set);
    return ret;
}

int TsFileReader::queryByRow(std::vector<std::string>& path_list, int offset,
                             int limit, ResultSet*& result_set) {
    std::vector<Path> path_list_vec;
    int ret = parse_paths(path_list, path_list_vec);
    if (ret != E_OK) {
        return ret;
    }
    QueryExpression* query_expression =
        QueryExpression::create(path_list_vec, nullptr);
    ret =
        tsfile_executor_->execute(query_expression, result_set, offset, limit);
    return ret;
}

int TsFileReader::prepare_series(const FileGeneration& generation,
                                 const PreparedLocator& locator,
                                 std::shared_ptr<PreparedSeries>& prepared) {
    return tsfile_executor_->prepare_series(generation, locator, prepared);
}

int TsFileReader::prepare_series(
    const FileGeneration& generation, const PreparedLocator& locator,
    const std::shared_ptr<PreparedSeries>& aligned_time_owner,
    std::shared_ptr<PreparedSeries>& prepared) {
    return tsfile_executor_->prepare_series(generation, locator,
                                            aligned_time_owner, prepared);
}

int TsFileReader::query_prepared(
    const std::shared_ptr<PreparedSeries>& prepared, int64_t start_time,
    int64_t end_time, int offset, int limit, ResultSet*& result_set) {
    if (prepared == nullptr || prepared->index() == nullptr) {
        return E_INVALID_ARG;
    }
    return tsfile_executor_->execute_prepared(
        prepared, start_time, end_time, offset, limit,
        prepared->index()->get_measurement_name().to_std_string(), result_set);
}

int TsFileReader::query_prepared_multi(
    const std::vector<std::shared_ptr<PreparedSeries>>& prepared,
    int64_t start_time, int64_t end_time, int offset, int limit,
    ResultSet*& result_set) {
    return tsfile_executor_->execute_prepared_multi(
        prepared, start_time, end_time, offset, limit, result_set);
}

int TsFileReader::queryByRow(const std::string& table_name,
                             const std::vector<std::string>& column_names,
                             int offset, int limit, ResultSet*& result_set,
                             Filter* tag_filter, int batch_size) {
    int ret = E_OK;
    TsFileMeta* tsfile_meta = nullptr;
    if (RET_FAIL(tsfile_executor_->get_tsfile_meta(tsfile_meta))) {
        return ret;
    }
    auto it = tsfile_meta->table_schemas_.find(to_lower(table_name));
    if (it == tsfile_meta->table_schemas_.end() || it->second == nullptr) {
        return E_TABLE_NOT_EXIST;
    }

    ensure_table_query_executor(batch_size);
    ret = table_query_executor_->query(to_lower(table_name), column_names,
                                       /*time_filter=*/nullptr, tag_filter,
                                       /*field_filter=*/nullptr, offset, limit,
                                       result_set);
    return ret;
}

int TsFileReader::query_table_on_tree(
    const std::vector<std::string>& measurement_names, int64_t star_time,
    int64_t end_time, ResultSet*& result_set) {
    int ret = E_OK;
    TsFileMeta* tsfile_meta = nullptr;
    if (RET_FAIL(tsfile_executor_->get_tsfile_meta(tsfile_meta))) {
        return ret;
    }
    std::vector<std::shared_ptr<IDeviceID>> device_ids;
    if (RET_FAIL(get_all_devices(device_ids))) {
        return ret;
    }
    std::vector<std::shared_ptr<IDeviceID>> satisfied_device_ids;
    std::unordered_set<std::string> measurement_names_set_to_query;
    size_t device_max_len = 0;

    if (measurement_names.empty()) {
        for (auto& device_name : device_ids) {
            std::vector<MeasurementSchema> schemas;
            if (RET_FAIL(get_timeseries_schema(device_name, schemas))) {
                return ret;
            }
            satisfied_device_ids.push_back(device_name);
            for (auto& schema : schemas) {
                measurement_names_set_to_query.insert(schema.measurement_name_);
            }
            device_name->split_table_name();
            if (device_name->get_split_seg_num() > device_max_len) {
                device_max_len = device_name->get_split_seg_num();
            }
        }
    } else {
        std::unordered_set<std::string> found_measurement_names;
        std::unordered_set<std::string> required_measurement_names(
            measurement_names.begin(), measurement_names.end());
        for (auto& device_name : device_ids) {
            std::vector<MeasurementSchema> schemas;
            if (RET_FAIL(get_timeseries_schema(device_name, schemas))) {
                return ret;
            }

            bool device_has_required_measurement_names = false;
            for (auto& schema : schemas) {
                if (required_measurement_names.find(schema.measurement_name_) !=
                    required_measurement_names.end()) {
                    found_measurement_names.insert(schema.measurement_name_);
                    device_has_required_measurement_names = true;
                }
            }
            if (device_has_required_measurement_names) {
                device_name->split_table_name();
                satisfied_device_ids.push_back(device_name);
                if (device_name->get_split_seg_num() > device_max_len) {
                    device_max_len = device_name->get_split_seg_num();
                }
            }
        }

        if (found_measurement_names.size() <
            required_measurement_names.size()) {
            return E_COLUMN_NOT_EXIST;
        }
        measurement_names_set_to_query = found_measurement_names;
    }
    std::vector<std::string> measurement_names_to_query;
    // Get all columns.
    if (measurement_names.empty() && !measurement_names_set_to_query.empty()) {
        for (auto& measurement_name : measurement_names_set_to_query) {
            measurement_names_to_query.push_back(measurement_name);
        }
    } else {
        measurement_names_to_query = measurement_names;
    }
    std::vector<std::string> columns_names(device_max_len);
    for (int i = 0; i < device_max_len; i++) {
        columns_names[i] = "col_" + std::to_string(i);
    }
    Filter* time_filter = new TimeBetween(star_time, end_time, false);
    ensure_table_query_executor(-1);
    ret = table_query_executor_->query_on_tree(
        satisfied_device_ids, columns_names, measurement_names_to_query,
        time_filter, result_set);
    return ret;
}

void TsFileReader::destroy_query_data_set(storage::ResultSet* qds) {
    tsfile_executor_->destroy_query_data_set(qds);
}

std::vector<std::shared_ptr<IDeviceID>> TsFileReader::get_all_devices(
    std::string table_name) {
    TsFileMeta* tsfile_meta = tsfile_executor_->get_tsfile_meta();
    std::vector<std::shared_ptr<IDeviceID>> device_ids;
    if (tsfile_meta != nullptr) {
        PageArena pa;
        pa.init(512, MOD_TSFILE_READER);
        to_lowercase_inplace(table_name);
        auto it = tsfile_meta->table_metadata_index_node_map_.find(table_name);
        if (it != tsfile_meta->table_metadata_index_node_map_.end() &&
            it->second != nullptr) {
            get_all_devices(device_ids, it->second, pa);
        }
    }
    return device_ids;
}

std::vector<std::shared_ptr<IDeviceID>> TsFileReader::get_all_device_ids() {
    std::vector<std::shared_ptr<IDeviceID>> device_ids;
    get_all_devices(device_ids);
    return device_ids;
}

std::vector<std::shared_ptr<IDeviceID>> TsFileReader::get_all_devices() {
    return get_all_device_ids();
}

int TsFileReader::get_all_devices(
    std::vector<std::shared_ptr<IDeviceID>>& device_ids) {
    device_ids.clear();
    if (tsfile_executor_ == nullptr) {
        return E_INVALID_ARG;
    }
    TsFileMeta* tsfile_meta = nullptr;
    int ret = tsfile_executor_->get_tsfile_meta(tsfile_meta);
    if (ret != E_OK) {
        return ret;
    }
    PageArena pa;
    pa.init(512, MOD_TSFILE_READER);
    for (const auto& entry : tsfile_meta->table_metadata_index_node_map_) {
        if (RET_FAIL(get_all_devices(device_ids, entry.second, pa))) {
            device_ids.clear();
            return ret;
        }
    }
    return E_OK;
}

int TsFileReader::get_all_devices(
    std::vector<std::shared_ptr<IDeviceID>>& device_ids,
    std::shared_ptr<MetaIndexNode> index_node, PageArena& pa) {
    int ret = E_OK;
    if (index_node != nullptr) {
        if (index_node->node_type_ == LEAF_DEVICE) {
            for (const auto& meta_index_entry : index_node->children_) {
                device_ids.push_back(meta_index_entry->get_device_id());
            }
        } else {
            for (size_t idx = 0; idx < index_node->children_.size(); idx++) {
                auto meta_index_entry = index_node->children_[idx];
                int start_offset = meta_index_entry->get_offset();
                int end_offset = index_node->end_offset_;
                if (idx + 1 < index_node->children_.size()) {
                    end_offset = index_node->children_[idx + 1]->get_offset();
                }
                ASSERT(end_offset - start_offset > 0);
                const int32_t read_size = (int32_t)end_offset - start_offset;
                int32_t ret_read_len = 0;
                char* data_buf = (char*)pa.alloc(read_size);
                void* m_idx_node_buf = pa.alloc(sizeof(MetaIndexNode));
                if (IS_NULL(data_buf) || IS_NULL(m_idx_node_buf)) {
                    return E_OOM;
                }
                auto* top_node_ptr = new (m_idx_node_buf) MetaIndexNode(&pa);
                auto top_node = std::shared_ptr<MetaIndexNode>(
                    top_node_ptr, [](MetaIndexNode* ptr) {
                        if (ptr) {
                            ptr->~MetaIndexNode();
                        }
                    });

                if (RET_FAIL(read_file_->read(start_offset, data_buf, read_size,
                                              ret_read_len))) {
                    return ret;
                } else if (ret_read_len != read_size) {
                    return E_FILE_READ_ERR;
                } else if (RET_FAIL(top_node->device_deserialize_from(
                               data_buf, read_size))) {
                    return ret;
                } else if (RET_FAIL(
                               get_all_devices(device_ids, top_node, pa))) {
                    return ret;
                }
            }
        }
    }
    return ret;
}

namespace {

// Encoding and compression are per-chunk properties that only live in the
// chunk header on disk: ChunkMeta::deserialize_from() reads nothing but
// offset_of_chunk_header_, so a ChunkMeta obtained from the metadata index
// never carries them.  Read the header back from the file instead.
int read_chunk_header_codec(RandomAccessReadFile* read_file,
                            int64_t chunk_header_offset,
                            size_t measurement_name_len,
                            common::TSEncoding& encoding,
                            common::CompressionType& compression) {
    if (read_file == nullptr || chunk_header_offset < 0) {
        return E_INVALID_ARG;
    }
    // A serialized chunk header is chunk_type (1 byte) + the measurement name
    // as a var_str (varint length + bytes) + data size as a var_uint + the
    // data type, compressor and encoding (1 byte each).  Size the buffer from
    // the name we already know so an unusually long measurement name cannot
    // truncate the header and send us back to the defaults.
    const size_t kMaxVarIntLen = 5;
    std::vector<char> buf(1 + kMaxVarIntLen + measurement_name_len +
                          kMaxVarIntLen + 3);
    int32_t read_len = 0;
    int ret = read_file->read(chunk_header_offset, buf.data(),
                              static_cast<int32_t>(buf.size()), read_len);
    if (ret != E_OK) {
        return ret;
    }
    if (read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
        return E_TSFILE_CORRUPTED;
    }
    common::ByteStream in;
    in.wrap_from(buf.data(), read_len);
    ChunkHeader chunk_header;
    if (RET_FAIL(chunk_header.deserialize_from(in))) {
        return ret;
    }
    encoding = chunk_header.encoding_type_;
    compression = chunk_header.compression_type_;
    return E_OK;
}

}  // namespace

int TsFileReader::get_timeseries_schema(
    std::shared_ptr<IDeviceID> device_id,
    std::vector<MeasurementSchema>& result) {
    int ret = E_OK;
    std::vector<ITimeseriesIndex*> timeseries_indexs;
    PageArena pa;
    pa.init(512, MOD_TSFILE_READER);
    if (RET_FAIL(tsfile_executor_->get_tsfile_io_reader()
                     ->get_device_timeseries_meta_without_chunk_meta(
                         device_id, timeseries_indexs, pa))) {
    } else {
        for (auto timeseries_index : timeseries_indexs) {
            // AlignedTimeseriesIndex::get_data_type() returns the time
            // column type (VECTOR) so the aligned/non-aligned dispatch in
            // SSI can keep using the existing accessor.  For schema
            // exposure we need the actual value column type — without this
            // unwrap, INT32/FLOAT/... would all surface as VECTOR.
            common::TSDataType dt = timeseries_index->get_data_type();
            if (dt == common::VECTOR) {
                auto* aligned =
                    dynamic_cast<AlignedTimeseriesIndex*>(timeseries_index);
                if (aligned != nullptr && aligned->value_ts_idx_ != nullptr) {
                    dt = aligned->value_ts_idx_->get_data_type();
                }
            }
            // Report the codecs stored in the final chunk. The two-argument
            // MeasurementSchema constructor would otherwise use library
            // defaults instead of the codecs in the chunk header.
            const std::string measurement_name =
                timeseries_index->get_measurement_name().to_std_string();
            common::TSEncoding encoding = common::get_value_encoder(dt);
            common::CompressionType compression =
                common::get_default_compressor();

            // Aligned indexes keep chunk metadata on the value index; the
            // base get_chunk_meta_list() is intentionally null for them.
            auto* chunk_meta_list = timeseries_index->get_chunk_meta_list();
            if (chunk_meta_list == nullptr) {
                chunk_meta_list = timeseries_index->get_value_chunk_meta_list();
            }
            if (chunk_meta_list != nullptr && !chunk_meta_list->empty() &&
                chunk_meta_list->back() != nullptr) {
                common::TSEncoding stored_encoding = common::INVALID_ENCODING;
                common::CompressionType stored_compression =
                    common::INVALID_COMPRESSION;
                ret = read_chunk_header_codec(
                    tsfile_executor_->get_tsfile_io_reader()->get_read_file(),
                    chunk_meta_list->back()->offset_of_chunk_header_,
                    measurement_name.size(), stored_encoding,
                    stored_compression);
                if (RET_FAIL(ret)) {
                    break;
                }
                encoding = stored_encoding;
                compression = stored_compression;
            }

            result.emplace_back(measurement_name, dt, encoding, compression);
        }
    }
    return ret;
}

int TsFileReader::get_timeseries_metadata_impl(
    std::shared_ptr<IDeviceID> device_id,
    std::vector<std::shared_ptr<ITimeseriesIndex>>& result) {
    int ret = E_OK;
    std::vector<ITimeseriesIndex*> timeseries_indexs;
    auto noop_deleter = [](ITimeseriesIndex*) {};
    if (RET_FAIL(
            tsfile_executor_->get_tsfile_io_reader()
                ->get_device_timeseries_meta_without_chunk_meta(
                    device_id, timeseries_indexs, tsfile_reader_meta_pa_))) {
    } else {
        for (auto timeseries_index : timeseries_indexs) {
            result.emplace_back(std::shared_ptr<ITimeseriesIndex>(
                timeseries_index, noop_deleter));
        }
    }
    return ret;
}

DeviceTimeseriesMetadataMap TsFileReader::get_timeseries_metadata(
    const std::vector<std::shared_ptr<IDeviceID>>& device_ids) {
    // Reset the shared meta arena up front: every call writes fresh
    // timeseries-index metadata into it via _impl(), and the previous
    // implementation only ever appended.  A long-lived reader that repeats
    // this query would grow tsfile_reader_meta_pa_ without bound (each call
    // duplicates the per-device payload).  Callers that need to retain prior
    // results past this call must copy them out before invoking again — the
    // shared_ptrs handed back use a noop deleter pointing into this arena.
    tsfile_reader_meta_pa_.destroy();
    tsfile_reader_meta_pa_.init(512, MOD_TSFILE_READER);
    DeviceTimeseriesMetadataMap result;
    for (const auto& device_id : device_ids) {
        std::vector<std::shared_ptr<ITimeseriesIndex>> list;
        if (get_timeseries_metadata_impl(device_id, list) == E_OK) {
            result.insert(std::make_pair(device_id, std::move(list)));
        }
        // Skip non-existent devices (not inserted)
    }
    return result;
}

DeviceTimeseriesMetadataMap TsFileReader::get_timeseries_metadata() {
    DeviceTimeseriesMetadataMap result;
    TsFileMeta* tsfile_meta = tsfile_executor_->get_tsfile_meta();
    if (tsfile_meta == nullptr) {
        return result;
    }

    // Same arena-reset rationale as the device_ids overload above.
    tsfile_reader_meta_pa_.destroy();
    tsfile_reader_meta_pa_.init(512, MOD_TSFILE_READER);

    PageArena pa;
    pa.init(512, MOD_TSFILE_READER);
    std::vector<DeviceMetaEntry> entries;
    for (auto& table_entry : tsfile_meta->table_metadata_index_node_map_) {
        if (get_all_device_entries(entries, table_entry.second,
                                   read_file_.get(), pa) != E_OK) {
            return result;
        }
    }

    auto noop_deleter = [](ITimeseriesIndex*) {};
    for (auto& device_entry : entries) {
        std::vector<ITimeseriesIndex*> raw_ts_indexes;
        if (tsfile_executor_->get_tsfile_io_reader()
                ->get_device_timeseries_meta_by_offset(
                    device_entry.start_offset, device_entry.end_offset,
                    raw_ts_indexes, tsfile_reader_meta_pa_) == E_OK) {
            std::vector<std::shared_ptr<ITimeseriesIndex>> list;
            for (auto ts_idx : raw_ts_indexes) {
                list.emplace_back(
                    std::shared_ptr<ITimeseriesIndex>(ts_idx, noop_deleter));
            }
            result.insert(
                std::make_pair(device_entry.device_id, std::move(list)));
        }
    }
    return result;
}

TsFileProperties TsFileReader::get_tsfile_properties() {
    if (tsfile_executor_ == nullptr) {
        return TsFileProperties();
    }
    TsFileMeta* file_metadata = tsfile_executor_->get_tsfile_meta();
    return file_metadata == nullptr ? TsFileProperties()
                                    : file_metadata->tsfile_properties_;
}

ResultSet* TsFileReader::read_timeseries(
    const std::shared_ptr<IDeviceID>& device_id,
    const std::vector<std::string>& measurement_name) {
    return nullptr;
}

std::shared_ptr<TableSchema> TsFileReader::get_table_schema(
    const std::string& table_name) {
    TsFileMeta* file_metadata = tsfile_executor_->get_tsfile_meta();
    std::shared_ptr<TableSchema> table_schema;
    // A schema-only table has no device-level metadata index.  Schema lookup
    // must therefore be independent of the presence of data pages; callers
    // can still construct an empty result set from the returned schema.
    if (file_metadata == nullptr) return table_schema;
    file_metadata->get_table_schema(to_lower(table_name), table_schema);
    return table_schema;
}

std::vector<std::shared_ptr<TableSchema>>
TsFileReader::get_all_table_schemas() {
    std::vector<std::shared_ptr<TableSchema>> table_schemas;
    get_all_table_schemas(table_schemas);
    return table_schemas;
}

int TsFileReader::get_all_table_schemas(
    std::vector<std::shared_ptr<TableSchema>>& table_schemas) {
    table_schemas.clear();
    if (tsfile_executor_ == nullptr) {
        return E_INVALID_ARG;
    }
    TsFileMeta* file_metadata = nullptr;
    const int ret = tsfile_executor_->get_tsfile_meta(file_metadata);
    if (ret != E_OK) {
        return ret;
    }
    for (const auto& table_schema : file_metadata->table_schemas_) {
        table_schemas.push_back(table_schema.second);
    }
    return E_OK;
}

}  // namespace storage