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

#include <fcntl.h>
#include <file/write_file.h>
#include <reader/qds_without_timegenerator.h>
#include <writer/tsfile_table_writer.h>

#ifdef _WIN32
#include <io.h>
#else
#include <unistd.h>
#endif

#include <cstring>
#include <limits>
#include <memory>
#include <new>
#include <set>
#include <vector>

#include "common/device_id.h"
#include "common/global.h"
#include "common/statistic.h"
#include "common/tablet.h"
#include "common/tsfile_common.h"
#include "reader/filter/tag_filter.h"
#include "reader/result_set.h"
#include "reader/table_result_set.h"
#include "reader/tsfile_reader.h"
#include "utils/injection.h"
#include "writer/tsfile_writer.h"

// Forward declarations for arrow namespace functions (defined in arrow_c.cc)
namespace arrow {
int TsBlockToArrowStruct(common::TsBlock& tsblock, ArrowArray* out_array,
                         ArrowSchema* out_schema);
int ArrowStructToTablet(const char* table_name, const ArrowArray* in_array,
                        const ArrowSchema* in_schema,
                        const storage::TableSchema* reg_schema,
                        storage::Tablet** out_tablet, int time_col_index);
}  // namespace arrow

namespace {

bool is_supported_table_type(TSDataType type) {
    switch (type) {
        case TS_DATATYPE_BOOLEAN:
        case TS_DATATYPE_INT32:
        case TS_DATATYPE_INT64:
        case TS_DATATYPE_FLOAT:
        case TS_DATATYPE_DOUBLE:
        case TS_DATATYPE_TEXT:
        case TS_DATATYPE_TIMESTAMP:
        case TS_DATATYPE_DATE:
        case TS_DATATYPE_BLOB:
        case TS_DATATYPE_STRING:
            return true;
        default:
            return false;
    }
}

ERRNO validate_table_schema(const TableSchema* schema) {
    if (schema == nullptr || schema->table_name == nullptr) {
        return common::E_INVALID_ARG;
    }
    if (schema->table_name[0] == '\0' || schema->column_num <= 0) {
        return common::E_INVALID_SCHEMA;
    }
    if (schema->column_schemas == nullptr) {
        return common::E_INVALID_ARG;
    }
    std::set<std::string> names;
    for (int i = 0; i < schema->column_num; ++i) {
        const ColumnSchema& column = schema->column_schemas[i];
        if (column.column_name == nullptr) {
            return common::E_INVALID_ARG;
        }
        if (column.column_name[0] == '\0' ||
            !names.insert(storage::to_lower(column.column_name)).second ||
            !is_supported_table_type(column.data_type) ||
            (column.column_category != TAG &&
             column.column_category != FIELD) ||
            (column.column_category == TAG &&
             column.data_type != TS_DATATYPE_STRING)) {
            return common::E_INVALID_SCHEMA;
        }
    }
    return common::E_OK;
}

}  // namespace

#ifdef __cplusplus
extern "C" {
#endif

void init_tsfile_config() { storage::libtsfile_init(); }

uint8_t get_global_time_encoding() {
    return common::get_global_time_encoding();
}

uint8_t get_global_time_compression() {
    return common::get_global_time_compression();
}

uint8_t get_datatype_encoding(uint8_t data_type) {
    return common::get_datatype_encoding(data_type);
}

uint8_t get_global_compression() { return common::get_global_compression(); }

int set_global_time_encoding(uint8_t encoding) {
    return common::set_global_time_encoding(encoding);
}

int set_global_time_compression(uint8_t compression) {
    return common::set_global_time_compression(compression);
}

int set_datatype_encoding(uint8_t data_type, uint8_t encoding) {
    return common::set_datatype_encoding(data_type, encoding);
}

int set_global_compression(uint8_t compression) {
    return common::set_global_compression(compression);
}

ERRNO tsfile_set_file_read_backend(int32_t backend) {
    switch (backend) {
        case TSFILE_READ_BACKEND_AUTO:
            return common::set_file_read_backend(common::FileReadBackend::AUTO);
        case TSFILE_READ_BACKEND_MMAP:
            return common::set_file_read_backend(common::FileReadBackend::MMAP);
        case TSFILE_READ_BACKEND_PREAD:
            return common::set_file_read_backend(
                common::FileReadBackend::PREAD);
        default:
            return common::E_INVALID_ARG;
    }
}

TsFileReadBackend tsfile_get_file_read_backend() {
    return static_cast<TsFileReadBackend>(common::get_file_read_backend());
}

WriteFile write_file_new(const char* pathname, ERRNO* err_code) {
    if (err_code == nullptr) {
        return nullptr;
    }
    if (pathname == nullptr || pathname[0] == '\0') {
        *err_code = common::E_INVALID_ARG;
        return nullptr;
    }
    int ret;
    init_tsfile_config();

    int flags = O_RDWR | O_CREAT | O_TRUNC;
#ifdef _WIN32
    flags |= O_BINARY;
#endif
    mode_t mode = 0666;
    storage::WriteFile* file = new (std::nothrow) storage::WriteFile;
    if (file == nullptr) {
        *err_code = common::E_OOM;
        return nullptr;
    }
    ret = file->create(pathname, flags, mode);
    if (ret != common::E_OK) {
        delete file;
        *err_code = ret;
        return nullptr;
    }
    *err_code = ret;
    return file;
}

TsFileWriter tsfile_writer_new(WriteFile file, TableSchema* schema,
                               ERRNO* err_code) {
    // C API: every public entry must defend against null callers — a null
    // schema or err_code would crash the host process the moment it's
    // dereferenced.  The tag-filter helpers already follow this pattern.
    if (err_code == nullptr) {
        return nullptr;
    }
    if (file == nullptr) {
        *err_code = common::E_INVALID_ARG;
        return nullptr;
    }
    *err_code = validate_table_schema(schema);
    if (*err_code != common::E_OK) {
        return nullptr;
    }

    try {
        init_tsfile_config();
        std::vector<common::ColumnSchema> column_schemas;
        for (int i = 0; i < schema->column_num; i++) {
            ColumnSchema cur_schema = schema->column_schemas[i];
            column_schemas.emplace_back(
                cur_schema.column_name,
                static_cast<common::TSDataType>(cur_schema.data_type),
                static_cast<common::ColumnCategory>(
                    cur_schema.column_category));
        }

        std::unique_ptr<storage::TableSchema> table_schema(
            new storage::TableSchema(schema->table_name, column_schemas));
        std::unique_ptr<storage::TsFileTableWriter> table_writer(
            new storage::TsFileTableWriter(
                static_cast<storage::WriteFile*>(file), table_schema.get()));
        *err_code = table_writer->get_error();
        if (*err_code != common::E_OK) {
            return nullptr;
        }
        return table_writer.release();
    } catch (const std::bad_alloc&) {
        *err_code = common::E_OOM;
        return nullptr;
    } catch (...) {
        *err_code = common::E_INVALID_SCHEMA;
        return nullptr;
    }
}

TsFileWriter tsfile_writer_new_with_memory_threshold(WriteFile file,
                                                     TableSchema* schema,
                                                     uint64_t memory_threshold,
                                                     ERRNO* err_code) {
    // See tsfile_writer_new() above for the null-guard rationale.
    if (err_code == nullptr) {
        return nullptr;
    }
    if (file == nullptr || memory_threshold == 0) {
        *err_code = common::E_INVALID_ARG;
        return nullptr;
    }
    *err_code = validate_table_schema(schema);
    if (*err_code != common::E_OK) {
        return nullptr;
    }
    try {
        init_tsfile_config();
        std::vector<common::ColumnSchema> column_schemas;
        for (int i = 0; i < schema->column_num; i++) {
            ColumnSchema cur_schema = schema->column_schemas[i];
            column_schemas.emplace_back(
                cur_schema.column_name,
                static_cast<common::TSDataType>(cur_schema.data_type),
                static_cast<common::ColumnCategory>(
                    cur_schema.column_category));
        }

        std::unique_ptr<storage::TableSchema> table_schema(
            new storage::TableSchema(schema->table_name, column_schemas));
        std::unique_ptr<storage::TsFileTableWriter> table_writer(
            new storage::TsFileTableWriter(
                static_cast<storage::WriteFile*>(file), table_schema.get(),
                memory_threshold));
        *err_code = table_writer->get_error();
        if (*err_code != common::E_OK) {
            return nullptr;
        }
        return table_writer.release();
    } catch (const std::bad_alloc&) {
        *err_code = common::E_OOM;
        return nullptr;
    } catch (...) {
        *err_code = common::E_INVALID_SCHEMA;
        return nullptr;
    }
}

TsFileReader tsfile_reader_new(const char* pathname, ERRNO* err_code) {
    if (err_code == nullptr) {
        return nullptr;
    }
    if (pathname == nullptr || pathname[0] == '\0') {
        *err_code = common::E_INVALID_ARG;
        return nullptr;
    }
    init_tsfile_config();
    auto reader = new (std::nothrow) storage::TsFileReader();
    if (reader == nullptr) {
        *err_code = common::E_OOM;
        return nullptr;
    }
    int ret = reader->open(pathname);
    if (ret != common::E_OK) {
        *err_code = ret;
        delete reader;
        return nullptr;
    }
    return reader;
}

ERRNO tsfile_writer_close(TsFileWriter writer) {
    if (writer == nullptr) {
        return common::E_OK;
    }
    auto* w = static_cast<storage::TsFileTableWriter*>(writer);
    int ret = w->close();
    if (ret != common::E_OK) {
        return ret;
    }
    delete w;
    return ret;
}

ERRNO tsfile_writer_add_tsfile_property(TsFileWriter writer, const char* key,
                                        uint32_t key_len, const uint8_t* value,
                                        uint32_t value_len) {
    if (writer == nullptr || key == nullptr ||
        (value == nullptr && value_len > 0)) {
        return common::E_INVALID_ARG;
    }
    if (key_len > static_cast<uint32_t>(std::numeric_limits<int32_t>::max()) ||
        value_len >
            static_cast<uint32_t>(std::numeric_limits<int32_t>::max())) {
        return common::E_OUT_OF_RANGE;
    }
    try {
        auto* w = static_cast<storage::TsFileTableWriter*>(writer);
        return w->add_tsfile_property(std::string(key, key_len), value,
                                      value_len);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    }
}

ERRNO tsfile_writer_flush(TsFileWriter writer) {
    if (writer == nullptr) {
        return common::E_INVALID_ARG;
    }
    return static_cast<storage::TsFileTableWriter*>(writer)->flush();
}

ERRNO tsfile_reader_close(TsFileReader reader) {
    auto* ts_reader = static_cast<storage::TsFileReader*>(reader);
    delete ts_reader;
    return common::E_OK;
}

Tablet tablet_new(char** column_name_list, TSDataType* data_types,
                  uint32_t column_num, uint32_t max_rows, ERRNO* err_code) {
    if (err_code == nullptr) {
        return nullptr;
    }
    *err_code = common::E_INVALID_ARG;
    if (column_num == 0 || max_rows == 0 || column_name_list == nullptr ||
        data_types == nullptr || max_rows >= (1u << 30)) {
        return nullptr;
    }
    try {
        std::vector<std::string> measurement_list;
        std::vector<common::TSDataType> data_type_list;
        std::set<std::string> names;
        for (uint32_t i = 0; i < column_num; i++) {
            if (column_name_list[i] == nullptr ||
                column_name_list[i][0] == '\0') {
                return nullptr;
            }
            if (!is_supported_table_type(data_types[i])) {
                *err_code = common::E_TYPE_NOT_SUPPORTED;
                return nullptr;
            }
            std::string name = storage::to_lower(column_name_list[i]);
            if (!names.insert(name).second) {
                *err_code = common::E_INVALID_SCHEMA;
                return nullptr;
            }
            measurement_list.emplace_back(std::move(name));
            data_type_list.push_back(
                static_cast<common::TSDataType>(*(data_types + i)));
        }
        auto* tablet = new (std::nothrow)
            storage::Tablet(measurement_list, data_type_list, max_rows);
        if (tablet == nullptr) {
            *err_code = common::E_OOM;
            return nullptr;
        }
        if (tablet->err_code_ != common::E_OK) {
            *err_code = tablet->err_code_;
            delete tablet;
            return nullptr;
        }
        *err_code = common::E_OK;
        return tablet;
    } catch (const std::bad_alloc&) {
        *err_code = common::E_OOM;
        return nullptr;
    } catch (...) {
        *err_code = common::E_INVALID_SCHEMA;
        return nullptr;
    }
}

uint32_t tablet_get_cur_row_size(Tablet tablet) {
    return static_cast<storage::Tablet*>(tablet)->get_cur_row_size();
}

ERRNO tablet_add_timestamp(Tablet tablet, uint32_t row_index,
                           Timestamp timestamp) {
    return static_cast<storage::Tablet*>(tablet)->add_timestamp(row_index,
                                                                timestamp);
}

#define TABLET_ADD_VALUE_BY_NAME_DEF(type)                                   \
    ERRNO tablet_add_value_by_name_##type(Tablet tablet, uint32_t row_index, \
                                          const char* column_name,           \
                                          const type value) {                \
        return static_cast<storage::Tablet*>(tablet)->add_value(             \
            row_index, storage::to_lower(column_name), value);               \
    }
TABLET_ADD_VALUE_BY_NAME_DEF(int32_t);
TABLET_ADD_VALUE_BY_NAME_DEF(int64_t);
TABLET_ADD_VALUE_BY_NAME_DEF(float);
TABLET_ADD_VALUE_BY_NAME_DEF(double);
TABLET_ADD_VALUE_BY_NAME_DEF(bool);

ERRNO tablet_add_value_by_name_string_with_len(Tablet tablet,
                                               uint32_t row_index,
                                               const char* column_name,
                                               const char* value,
                                               int value_len) {
    return static_cast<storage::Tablet*>(tablet)->add_value(
        row_index, storage::to_lower(column_name),
        common::String(value, value_len));
}

#define TABLE_ADD_VALUE_BY_INDEX_DEF(type)                                    \
    ERRNO tablet_add_value_by_index_##type(Tablet tablet, uint32_t row_index, \
                                           uint32_t column_index,             \
                                           const type value) {                \
        return static_cast<storage::Tablet*>(tablet)->add_value(              \
            row_index, column_index, value);                                  \
    }

ERRNO tablet_add_value_by_index_string_with_len(Tablet tablet,
                                                uint32_t row_index,
                                                uint32_t column_index,
                                                const char* value,
                                                int value_len) {
    return static_cast<storage::Tablet*>(tablet)->add_value(
        row_index, column_index, common::String(value, value_len));
}

TABLE_ADD_VALUE_BY_INDEX_DEF(int32_t);
TABLE_ADD_VALUE_BY_INDEX_DEF(int64_t);
TABLE_ADD_VALUE_BY_INDEX_DEF(float);
TABLE_ADD_VALUE_BY_INDEX_DEF(double);
TABLE_ADD_VALUE_BY_INDEX_DEF(bool);

// TsRecord API
TsRecord _ts_record_new(const char* device_id, Timestamp timestamp,
                        int timeseries_num) {
    auto* record = new storage::TsRecord(timestamp, device_id, timeseries_num);
    return record;
}

#define INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(type)                 \
    ERRNO _insert_data_into_ts_record_by_name_##type(                \
        TsRecord data, const char* measurement_name, type value) {   \
        auto* record = (storage::TsRecord*)data;                     \
        storage::DataPoint point(measurement_name, value);           \
        if (record->points_.size() + 1 > record->points_.capacity()) \
            return common::E_BUF_NOT_ENOUGH;                         \
        record->points_.push_back(point);                            \
        return common::E_OK;                                         \
    }

ERRNO _insert_data_into_ts_record_by_name_string_with_len(
    TsRecord data, const char* measurement_name, const char* value,
    const uint32_t value_len) {
    auto* record = (storage::TsRecord*)data;
    if (record->points_.size() + 1 > record->points_.capacity())
        return common::E_BUF_NOT_ENOUGH;
    common::String str_value;
    str_value.dup_from(value, value_len, record->pa);
    record->add_point(measurement_name, str_value);
    return common::E_OK;
}

INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(int32_t);
INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(int64_t);
INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(bool);
INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(float);
INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(double);
/*
TsFileWriter tsfile_writer_new_with_conf(const char *pathname,
                                     const mode_t flag, ERRNO *err_code,
                                     TsFileConf *conf) {
init_tsfile_config();
auto *writer = new storage::TsFileWriter();
const int ret = writer->open(pathname, O_CREAT | O_RDWR, flag);
if (ret != common::E_OK) {
    delete writer;
    *err_code = ret;
    return nullptr;
}
return writer;
}

*/
ERRNO tsfile_writer_write(TsFileWriter writer, Tablet tablet) {
    if (writer == nullptr || tablet == nullptr) {
        return common::E_INVALID_ARG;
    }
    auto* w = static_cast<storage::TsFileTableWriter*>(writer);
    auto* tbl = static_cast<storage::Tablet*>(tablet);
    return w->write_table(*tbl);
}

ERRNO tsfile_writer_write_arrow(TsFileWriter writer, ArrowArray* array,
                                ArrowSchema* schema, int time_col_index) {
    if (writer == nullptr || array == nullptr || schema == nullptr) {
        return common::E_INVALID_ARG;
    }
    try {
        auto* w = static_cast<storage::TsFileTableWriter*>(writer);
        std::shared_ptr<storage::TableSchema> registered_schema =
            w->get_table_schema();
        if (!registered_schema) {
            return common::E_INVALID_SCHEMA;
        }
        storage::Tablet* raw_tablet = nullptr;
        int ret = arrow::ArrowStructToTablet(w->get_table_name().c_str(), array,
                                             schema, registered_schema.get(),
                                             &raw_tablet, time_col_index);
        std::unique_ptr<storage::Tablet> tablet(raw_tablet);
        if (ret != common::E_OK) {
            return ret;
        }
        return w->write_table(*tablet);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    } catch (...) {
        return common::E_INVALID_ARG;
    }
}

// ERRNO tsfile_writer_flush_data(TsFileWriter writer) {
//     auto *w = static_cast<storage::TsFileWriter *>(writer);
//     return w->flush();
// }

// Query

PreparedSeriesHandle tsfile_reader_prepare_series(
    TsFileReader reader, const TsFilePreparedLocator* locator, bool trust_index,
    ERRNO* err_code) {
    if (err_code == nullptr) {
        return nullptr;
    }
    *err_code = common::E_INVALID_ARG;
    if (reader == nullptr || locator == nullptr) {
        return nullptr;
    }
    storage::FileGeneration generation;
    generation.mapped_index_identity = locator->mapped_index_identity;
    generation.file_id = locator->file_id;
    generation.file_size = locator->file_size;
    generation.file_fingerprint = locator->file_fingerprint;
    generation.trust_index = trust_index;
    storage::PreparedLocator native_locator;
    native_locator.locator_id = locator->locator_id;
    native_locator.layout = locator->layout;
    native_locator.flags = locator->flags;
    native_locator.value_metadata_offset = locator->value_metadata_offset;
    native_locator.value_metadata_length = locator->value_metadata_length;
    native_locator.time_metadata_offset = locator->time_metadata_offset;
    native_locator.time_metadata_length = locator->time_metadata_length;
    std::shared_ptr<storage::PreparedSeries> prepared;
    *err_code = static_cast<storage::TsFileReader*>(reader)->prepare_series(
        generation, native_locator, prepared);
    if (*err_code != common::E_OK) {
        return nullptr;
    }
    auto* handle = new (std::nothrow)
        std::shared_ptr<storage::PreparedSeries>(std::move(prepared));
    if (handle == nullptr) {
        *err_code = common::E_OOM;
    }
    return handle;
}

PreparedSeriesHandle tsfile_reader_prepare_series_with_time_owner(
    TsFileReader reader, const TsFilePreparedLocator* locator,
    PreparedSeriesHandle aligned_time_owner, bool trust_index,
    ERRNO* err_code) {
    if (err_code == nullptr) {
        return nullptr;
    }
    *err_code = common::E_INVALID_ARG;
    if (reader == nullptr || locator == nullptr ||
        aligned_time_owner == nullptr) {
        return nullptr;
    }

    storage::FileGeneration generation;
    generation.mapped_index_identity = locator->mapped_index_identity;
    generation.file_id = locator->file_id;
    generation.file_size = locator->file_size;
    generation.file_fingerprint = locator->file_fingerprint;
    generation.trust_index = trust_index;
    storage::PreparedLocator native_locator;
    native_locator.locator_id = locator->locator_id;
    native_locator.layout = locator->layout;
    native_locator.flags = locator->flags;
    native_locator.value_metadata_offset = locator->value_metadata_offset;
    native_locator.value_metadata_length = locator->value_metadata_length;
    native_locator.time_metadata_offset = locator->time_metadata_offset;
    native_locator.time_metadata_length = locator->time_metadata_length;

    auto* owner = static_cast<std::shared_ptr<storage::PreparedSeries>*>(
        aligned_time_owner);
    std::shared_ptr<storage::PreparedSeries> prepared;
    *err_code = static_cast<storage::TsFileReader*>(reader)->prepare_series(
        generation, native_locator, *owner, prepared);
    if (*err_code != common::E_OK) {
        return nullptr;
    }
    auto* handle = new (std::nothrow)
        std::shared_ptr<storage::PreparedSeries>(std::move(prepared));
    if (handle == nullptr) {
        *err_code = common::E_OOM;
    }
    return handle;
}

void tsfile_prepared_series_free(PreparedSeriesHandle prepared) {
    delete static_cast<std::shared_ptr<storage::PreparedSeries>*>(prepared);
}

ResultSet tsfile_reader_query_prepared(TsFileReader reader,
                                       PreparedSeriesHandle prepared,
                                       Timestamp start_time, Timestamp end_time,
                                       int offset, int limit, ERRNO* err_code) {
    if (err_code == nullptr) {
        return nullptr;
    }
    *err_code = common::E_INVALID_ARG;
    if (reader == nullptr || prepared == nullptr) {
        return nullptr;
    }
    auto* handle =
        static_cast<std::shared_ptr<storage::PreparedSeries>*>(prepared);
    storage::ResultSet* result = nullptr;
    *err_code = static_cast<storage::TsFileReader*>(reader)->query_prepared(
        *handle, start_time, end_time, offset, limit, result);
    return result;
}

ResultSet tsfile_reader_query_prepared_multi(
    TsFileReader reader, const PreparedSeriesHandle* prepared,
    uint32_t prepared_count, Timestamp start_time, Timestamp end_time,
    int offset, int limit, ERRNO* err_code) {
    if (err_code == nullptr) {
        return nullptr;
    }
    *err_code = common::E_INVALID_ARG;
    if (reader == nullptr || prepared == nullptr || prepared_count == 0) {
        return nullptr;
    }

    std::vector<std::shared_ptr<storage::PreparedSeries>> native_prepared;
    native_prepared.reserve(prepared_count);
    for (uint32_t i = 0; i < prepared_count; i++) {
        if (prepared[i] == nullptr) {
            return nullptr;
        }
        auto* handle =
            static_cast<std::shared_ptr<storage::PreparedSeries>*>(prepared[i]);
        native_prepared.push_back(*handle);
    }

    storage::ResultSet* result = nullptr;
    *err_code =
        static_cast<storage::TsFileReader*>(reader)->query_prepared_multi(
            native_prepared, start_time, end_time, offset, limit, result);
    return result;
}

ResultSet tsfile_query_table(TsFileReader reader, const char* table_name,
                             char** columns, uint32_t column_num,
                             Timestamp start_time, Timestamp end_time,
                             ERRNO* err_code) {
    auto* r = static_cast<storage::TsFileReader*>(reader);
    storage::ResultSet* table_result_set = nullptr;
    std::vector<std::string> column_names;
    for (uint32_t i = 0; i < column_num; i++) {
        column_names.emplace_back(columns[i]);
    }
    *err_code = r->query(table_name, column_names, start_time, end_time,
                         table_result_set);
    return table_result_set;
}

ResultSet tsfile_query_table_on_tree(TsFileReader reader, char** columns,
                                     uint32_t column_num, Timestamp start_time,
                                     Timestamp end_time, ERRNO* err_code) {
    auto* r = static_cast<storage::TsFileReader*>(reader);
    storage::ResultSet* table_result_set = nullptr;
    std::vector<std::string> column_names;
    for (uint32_t i = 0; i < column_num; i++) {
        column_names.emplace_back(columns[i]);
    }
    *err_code = r->query_table_on_tree(column_names, start_time, end_time,
                                       table_result_set);
    return table_result_set;
}

ResultSet tsfile_reader_query_tree(TsFileReader reader, char** paths,
                                   uint32_t path_num, Timestamp start_time,
                                   Timestamp end_time, ERRNO* err_code) {
    if (err_code == nullptr) {
        return nullptr;
    }
    *err_code = common::E_INVALID_ARG;
    if (reader == nullptr || paths == nullptr || path_num == 0) {
        return nullptr;
    }
    try {
        std::vector<std::string> selected_paths;
        selected_paths.reserve(path_num);
        for (uint32_t i = 0; i < path_num; i++) {
            if (paths[i] == nullptr) {
                return nullptr;
            }
            selected_paths.emplace_back(paths[i]);
        }
        storage::ResultSet* result_set = nullptr;
        *err_code = static_cast<storage::TsFileReader*>(reader)->query(
            selected_paths, start_time, end_time, result_set);
        return result_set;
    } catch (const std::bad_alloc&) {
        *err_code = common::E_OOM;
        return nullptr;
    }
}

ResultSet tsfile_reader_query_tree_by_row(TsFileReader reader,
                                          char** device_ids, int device_ids_len,
                                          char** measurement_names,
                                          int measurement_names_len, int offset,
                                          int limit, ERRNO* err_code) {
    auto* r = static_cast<storage::TsFileReader*>(reader);
    storage::ResultSet* result_set = nullptr;

    std::vector<std::string> path_list;
    if (device_ids_len > 0 && measurement_names_len > 0) {
        path_list.reserve(static_cast<size_t>(device_ids_len) *
                          static_cast<size_t>(measurement_names_len));
    }

    for (int i = 0; i < device_ids_len; i++) {
        const char* device_id = device_ids[i];
        if (device_id == nullptr) {
            continue;
        }
        for (int j = 0; j < measurement_names_len; j++) {
            const char* measurement_name = measurement_names[j];
            if (measurement_name == nullptr) {
                continue;
            }
            path_list.emplace_back(std::string(device_id) + "." +
                                   std::string(measurement_name));
        }
    }

    *err_code = r->queryByRow(path_list, offset, limit, result_set);
    return result_set;
}

ResultSet tsfile_reader_query_table_by_row(
    TsFileReader reader, const char* table_name, char** column_names,
    int column_names_len, int offset, int limit, TagFilterHandle tag_filter,
    int batch_size, ERRNO* err_code) {
    auto* r = static_cast<storage::TsFileReader*>(reader);
    storage::ResultSet* result_set = nullptr;

    std::vector<std::string> columns;
    if (column_names_len > 0) {
        columns.reserve(static_cast<size_t>(column_names_len));
    }
    for (int i = 0; i < column_names_len; i++) {
        const char* name = column_names[i];
        columns.emplace_back(name == nullptr ? "" : std::string(name));
    }

    *err_code = r->queryByRow(
        table_name == nullptr ? "" : table_name, columns, offset, limit,
        result_set, static_cast<storage::Filter*>(tag_filter), batch_size);
    return result_set;
}

ResultSet tsfile_reader_query_table(TsFileReader reader, const char* table_name,
                                    char** column_names,
                                    uint32_t column_names_len,
                                    Timestamp start_time, Timestamp end_time,
                                    int offset, int limit,
                                    TagFilterHandle tag_filter, int batch_size,
                                    ERRNO* err_code) {
    if (err_code == nullptr) {
        return nullptr;
    }
    *err_code = common::E_INVALID_ARG;
    if (reader == nullptr || table_name == nullptr || column_names == nullptr ||
        column_names_len == 0 || offset < 0 || batch_size < 0) {
        return nullptr;
    }
    try {
        std::vector<std::string> columns;
        columns.reserve(column_names_len);
        for (uint32_t i = 0; i < column_names_len; ++i) {
            if (column_names[i] == nullptr) {
                return nullptr;
            }
            columns.emplace_back(column_names[i]);
        }
        storage::ResultSet* result_set = nullptr;
        const int normalized_limit = limit < 0 ? -1 : limit;
        *err_code = static_cast<storage::TsFileReader*>(reader)->query(
            table_name, columns, start_time, end_time, offset, normalized_limit,
            result_set, static_cast<storage::Filter*>(tag_filter), batch_size);
        return result_set;
    } catch (const std::bad_alloc&) {
        *err_code = common::E_OOM;
        return nullptr;
    } catch (...) {
        *err_code = common::E_FILE_READ_ERR;
        return nullptr;
    }
}

ResultSet tsfile_query_table_batch(TsFileReader reader, const char* table_name,
                                   char** columns, uint32_t column_num,
                                   Timestamp start_time, Timestamp end_time,
                                   TagFilterHandle tag_filter, int batch_size,
                                   ERRNO* err_code) {
    auto* r = static_cast<storage::TsFileReader*>(reader);
    storage::ResultSet* table_result_set = nullptr;
    std::vector<std::string> column_names;
    for (uint32_t i = 0; i < column_num; i++) {
        column_names.emplace_back(columns[i]);
    }
    *err_code = r->query(table_name, column_names, start_time, end_time,
                         table_result_set,
                         static_cast<storage::Filter*>(tag_filter), batch_size);
    return table_result_set;
}

bool tsfile_result_set_next(ResultSet result_set, ERRNO* err_code) {
    if (err_code == nullptr) {
        return false;
    }
    if (result_set == nullptr) {
        *err_code = common::E_INVALID_ARG;
        return false;
    }
    auto* r = static_cast<storage::ResultSet*>(result_set);
    bool has_next = true;
    int ret = common::E_OK;
    ret = r->next(has_next);
    *err_code = ret;
    if (ret != common::E_OK) {
        return false;
    }
    return has_next;
}

ERRNO tsfile_result_set_get_next_tsblock_as_arrow(ResultSet result_set,
                                                  ArrowArray* out_array,
                                                  ArrowSchema* out_schema) {
    if (result_set == nullptr || out_array == nullptr ||
        out_schema == nullptr) {
        return common::E_INVALID_ARG;
    }
    std::memset(out_array, 0, sizeof(*out_array));
    std::memset(out_schema, 0, sizeof(*out_schema));

    auto* r = static_cast<storage::ResultSet*>(result_set);
    auto* table_result_set = dynamic_cast<storage::TableResultSet*>(r);
    if (table_result_set == nullptr) {
        return common::E_INVALID_ARG;
    }

    common::TsBlock* tsblock = nullptr;
    int ret = table_result_set->get_next_tsblock(tsblock);
    if (ret != common::E_OK) {
        return ret;
    }

    if (tsblock == nullptr) {
        return common::E_NO_MORE_DATA;
    }

    ret = arrow::TsBlockToArrowStruct(*tsblock, out_array, out_schema);
    return ret;
}

#define TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(type)                          \
    type tsfile_result_set_get_value_by_name_##type(ResultSet result_set,      \
                                                    const char* column_name) { \
        auto* r = static_cast<storage::ResultSet*>(result_set);                \
        std::string column_name_(column_name);                                 \
        return r->get_value<type>(column_name_);                               \
    }

TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(bool);
TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(int32_t);
TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(int64_t);
TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(float);
TSFILE_RESULT_SET_GET_VALUE_BY_NAME_DEF(double);
char* tsfile_result_set_get_value_by_name_string(ResultSet result_set,
                                                 const char* column_name) {
    auto* r = static_cast<storage::ResultSet*>(result_set);
    std::string column_name_(column_name);
    common::String* ret = r->get_value<common::String*>(column_name_);
    if (ret == nullptr) {
        return nullptr;
    }
    // Caller should free return's char* 's space.
    char* dup = (char*)malloc(ret->len_ + 1);
    if (dup) {
        memcpy(dup, ret->buf_, ret->len_);
        dup[ret->len_] = '\0';
    }
    return dup;
}

#define TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(type)                        \
    type tsfile_result_set_get_value_by_index_##type(ResultSet result_set,    \
                                                     uint32_t column_index) { \
        auto* r = static_cast<storage::ResultSet*>(result_set);               \
        return r->get_value<type>(column_index);                              \
    }

TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(int32_t);
TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(int64_t);
TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(float);
TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(double);
TSFILE_RESULT_SET_GET_VALUE_BY_INDEX_DEF(bool);

char* tsfile_result_set_get_value_by_index_string(ResultSet result_set,
                                                  uint32_t column_index) {
    auto* r = static_cast<storage::ResultSet*>(result_set);
    common::String* ret = r->get_value<common::String*>(column_index);
    if (ret == nullptr) {
        return nullptr;
    }
    // Caller should free return's char* 's space.
    char* dup = (char*)malloc(ret->len_ + 1);
    if (dup) {
        memcpy(dup, ret->buf_, ret->len_);
        dup[ret->len_] = '\0';
    }
    return dup;
}

ERRNO tsfile_result_set_get_value_by_index_binary(ResultSet result_set,
                                                  uint32_t column_index,
                                                  uint8_t** out_value,
                                                  uint32_t* out_length) {
    if (result_set == nullptr || column_index == 0 || out_value == nullptr ||
        out_length == nullptr) {
        return common::E_INVALID_ARG;
    }
    *out_value = nullptr;
    *out_length = 0;
    auto* r = static_cast<storage::ResultSet*>(result_set);
    common::String* value = r->get_value<common::String*>(column_index);
    if (value == nullptr) {
        return common::E_INVALID_ARG;
    }
    if (value->len_ == 0) {
        return common::E_OK;
    }
    uint8_t* copied = static_cast<uint8_t*>(malloc(value->len_));
    if (copied == nullptr) {
        return common::E_OOM;
    }
    memcpy(copied, value->buf_, value->len_);
    *out_value = copied;
    *out_length = value->len_;
    return common::E_OK;
}

bool tsfile_result_set_is_null_by_name(ResultSet result_set,
                                       const char* column_name) {
    auto* r = static_cast<storage::ResultSet*>(result_set);
    return r->is_null(column_name);
}

bool tsfile_result_set_is_null_by_index(const ResultSet result_set,
                                        const uint32_t column_index) {
    auto* r = static_cast<storage::ResultSet*>(result_set);
    return r->is_null(column_index);
}

ResultSetMetaData tsfile_result_set_get_metadata(ResultSet result_set) {
    auto* r = static_cast<storage::ResultSet*>(result_set);
    if (result_set == NULL) {
        return ResultSetMetaData();
    }

    ResultSetMetaData meta_data;
    std::shared_ptr<storage::ResultSetMetadata> result_set_metadata =
        r->get_metadata();
    meta_data.column_num = result_set_metadata->get_column_count();
    meta_data.column_names =
        static_cast<char**>(malloc(meta_data.column_num * sizeof(char*)));
    meta_data.data_types = static_cast<TSDataType*>(
        malloc(meta_data.column_num * sizeof(TSDataType)));
    for (int i = 0; i < meta_data.column_num; i++) {
        meta_data.column_names[i] =
            strdup(result_set_metadata->get_column_name(i + 1).c_str());
        meta_data.data_types[i] = static_cast<TSDataType>(
            result_set_metadata->get_column_type(i + 1));
    }
    return meta_data;
}

char* tsfile_result_set_metadata_get_column_name(ResultSetMetaData result_set,
                                                 uint32_t column_index) {
    if (column_index < 1 || column_index > (uint32_t)result_set.column_num) {
        return nullptr;
    }
    return result_set.column_names[column_index - 1];
}

TSDataType tsfile_result_set_metadata_get_data_type(
    ResultSetMetaData result_set, uint32_t column_index) {
    if (column_index < 1 ||
        column_index > static_cast<uint32_t>(result_set.column_num)) {
        return TS_DATATYPE_INVALID;
    }
    return result_set.data_types[column_index - 1];
}

int tsfile_result_set_metadata_get_column_num(ResultSetMetaData result_set) {
    return result_set.column_num;
}

TableSchema tsfile_reader_get_table_schema(TsFileReader reader,
                                           const char* table_name) {
    TableSchema ret_schema{};
    if (tsfile_reader_get_table_schema_checked(reader, table_name,
                                               &ret_schema) != common::E_OK) {
        return TableSchema{};
    }
    return ret_schema;
}

static ColumnSchema* allocate_table_schema_columns(size_t count) {
#ifdef ENABLE_TEST
    using common::g_all_inject_points;
    using common::InjectPoint;
    DBUG_EXECUTE_IF("table_schema_column_alloc_fail", return nullptr;);
#endif
    return static_cast<ColumnSchema*>(calloc(count, sizeof(ColumnSchema)));
}

// MSVC /EHsc assumes C-linkage helpers do not let exceptions escape.
static ERRNO copy_table_schema(const std::shared_ptr<storage::TableSchema>& src,
                               TableSchema* out_schema) {
    if (out_schema == nullptr) {
        return common::E_INVALID_ARG;
    }
    *out_schema = TableSchema{};
    if (!src) {
        return common::E_TABLE_NOT_EXIST;
    }
    try {
        out_schema->table_name = strdup(src->get_table_name().c_str());
        if (out_schema->table_name == nullptr) {
            return common::E_OOM;
        }
        const int column_num = src->get_columns_num();
        if (column_num == 0) {
            return common::E_OK;
        }
        out_schema->column_schemas =
            allocate_table_schema_columns(static_cast<size_t>(column_num));
        if (out_schema->column_schemas == nullptr) {
            free_table_schema(*out_schema);
            *out_schema = TableSchema{};
            return common::E_OOM;
        }
        out_schema->column_num = column_num;
#ifdef ENABLE_TEST
        using common::g_all_inject_points;
        using common::InjectPoint;
        DBUG_EXECUTE_IF("table_schema_copy_oom", throw std::bad_alloc(););
        DBUG_EXECUTE_IF("table_schema_copy_error",
                        throw common::E_FILE_READ_ERR;);
#endif
        const auto& measurements = src->get_measurement_schemas();
        const auto& categories = src->get_column_categories();
        if (measurements.size() < static_cast<size_t>(out_schema->column_num) ||
            categories.size() < static_cast<size_t>(out_schema->column_num)) {
            free_table_schema(*out_schema);
            *out_schema = TableSchema{};
            return common::E_INVALID_SCHEMA;
        }
        for (int i = 0; i < out_schema->column_num; ++i) {
            if (!measurements[i]) {
                free_table_schema(*out_schema);
                *out_schema = TableSchema{};
                return common::E_INVALID_SCHEMA;
            }
            out_schema->column_schemas[i].column_name =
                strdup(measurements[i]->measurement_name_.c_str());
            if (out_schema->column_schemas[i].column_name == nullptr) {
                free_table_schema(*out_schema);
                *out_schema = TableSchema{};
                return common::E_OOM;
            }
            out_schema->column_schemas[i].data_type =
                static_cast<TSDataType>(measurements[i]->data_type_);
            out_schema->column_schemas[i].column_category =
                static_cast<ColumnCategory>(categories[i]);
        }
        return common::E_OK;
    } catch (const std::bad_alloc&) {
        free_table_schema(*out_schema);
        *out_schema = TableSchema{};
        return common::E_OOM;
    } catch (...) {
        free_table_schema(*out_schema);
        *out_schema = TableSchema{};
        return common::E_FILE_READ_ERR;
    }
}

static void free_table_schema_array(TableSchema* schemas, size_t size) {
    if (schemas == nullptr) {
        return;
    }
    for (size_t i = 0; i < size; ++i) {
        free_table_schema(schemas[i]);
    }
    free(schemas);
}

static void free_device_schema_array(DeviceSchema* schemas, size_t size) {
    if (schemas == nullptr) {
        return;
    }
    for (size_t i = 0; i < size; ++i) {
        free_device_schema(schemas[i]);
    }
    free(schemas);
}

ERRNO tsfile_reader_get_table_schema_checked(TsFileReader reader,
                                             const char* table_name,
                                             TableSchema* out_schema) {
    if (out_schema == nullptr) {
        return common::E_INVALID_ARG;
    }
    *out_schema = TableSchema{};
    if (reader == nullptr || table_name == nullptr) {
        return common::E_INVALID_ARG;
    }
    try {
        std::shared_ptr<storage::TableSchema> schema;
        const int ret =
            static_cast<storage::TsFileReader*>(reader)->get_table_schema(
                table_name, schema);
        if (ret != common::E_OK) {
            return ret;
        }
        return copy_table_schema(schema, out_schema);
    } catch (const std::bad_alloc&) {
        free_table_schema(*out_schema);
        *out_schema = TableSchema{};
        return common::E_OOM;
    } catch (...) {
        free_table_schema(*out_schema);
        *out_schema = TableSchema{};
        return common::E_FILE_READ_ERR;
    }
}

TableSchema* tsfile_reader_get_all_table_schemas(TsFileReader reader,
                                                 uint32_t* size) {
    ERRNO error_code = common::E_OK;
    return tsfile_reader_get_all_table_schemas_with_error(reader, size,
                                                          &error_code);
}

TableSchema* tsfile_reader_get_all_table_schemas_with_error(TsFileReader reader,
                                                            uint32_t* size,
                                                            ERRNO* error_code) {
    if (size != nullptr) {
        *size = 0;
    }
    if (error_code == nullptr) {
        return nullptr;
    }
    TableSchema* schemas = nullptr;
    *error_code =
        tsfile_reader_get_all_table_schemas_checked(reader, &schemas, size);
    return *error_code == common::E_OK ? schemas : nullptr;
}

ERRNO tsfile_reader_get_all_table_schemas_checked(TsFileReader reader,
                                                  TableSchema** out_schemas,
                                                  uint32_t* out_size) {
    if (out_schemas != nullptr) {
        *out_schemas = nullptr;
    }
    if (out_size != nullptr) {
        *out_size = 0;
    }
    if (reader == nullptr || out_schemas == nullptr || out_size == nullptr) {
        return common::E_INVALID_ARG;
    }
    TableSchema* copied = nullptr;
    size_t initialized = 0;
    try {
        auto* r = static_cast<storage::TsFileReader*>(reader);
        std::vector<std::shared_ptr<storage::TableSchema>> schemas;
        const ERRNO read_ret = r->get_all_table_schemas(schemas);
        if (read_ret != common::E_OK) {
            return read_ret;
        }
        if (schemas.empty()) {
            return common::E_OK;
        }
        const size_t table_num = schemas.size();
        if (table_num > std::numeric_limits<uint32_t>::max()) {
            return common::E_OVERFLOW;
        }
        copied = static_cast<TableSchema*>(
            calloc(schemas.size(), sizeof(TableSchema)));
        if (copied == nullptr) {
            return common::E_OOM;
        }
        for (size_t i = 0; i < schemas.size(); ++i) {
            initialized = i + 1;
            ERRNO ret = copy_table_schema(schemas[i], &copied[i]);
            if (ret != common::E_OK) {
                free_table_schema_array(copied, initialized);
                return ret;
            }
        }
        *out_schemas = copied;
        *out_size = static_cast<uint32_t>(schemas.size());
        return common::E_OK;
    } catch (const std::bad_alloc&) {
        free_table_schema_array(copied, initialized);
        return common::E_OOM;
    } catch (...) {
        free_table_schema_array(copied, initialized);
        return common::E_FILE_READ_ERR;
    }
}

ERRNO tsfile_reader_get_all_timeseries_schemas_checked(
    TsFileReader reader, DeviceSchema** out_schemas, uint32_t* out_size) {
    if (out_schemas != nullptr) {
        *out_schemas = nullptr;
    }
    if (out_size != nullptr) {
        *out_size = 0;
    }
    if (reader == nullptr || out_schemas == nullptr || out_size == nullptr) {
        return common::E_INVALID_ARG;
    }
    DeviceSchema* device_schema = nullptr;
    size_t initialized = 0;
    try {
        auto* r = static_cast<storage::TsFileReader*>(reader);
        std::vector<std::shared_ptr<storage::IDeviceID>> device_ids;
        const ERRNO read_ret = r->get_all_devices(device_ids);
        if (read_ret != common::E_OK) {
            return read_ret;
        }
        if (device_ids.empty()) {
            return common::E_OK;
        }
        const size_t device_count = device_ids.size();
        if (device_count > std::numeric_limits<uint32_t>::max()) {
            return common::E_OVERFLOW;
        }
        device_schema = static_cast<DeviceSchema*>(
            calloc(device_count, sizeof(DeviceSchema)));
        if (device_schema == nullptr) {
            return common::E_OOM;
        }

        size_t device_index = 0;
        for (const auto& device_id : device_ids) {
            initialized = device_index + 1;
            DeviceSchema& cur_schema = device_schema[device_index++];
            std::string device_name =
                device_id == nullptr ? "" : device_id->get_device_name();
            cur_schema.device_name = strdup(device_name.c_str());
            if (cur_schema.device_name == nullptr) {
                free_device_schema_array(device_schema, initialized);
                return common::E_OOM;
            }

            std::vector<storage::MeasurementSchema> schemas;
            int ret = r->get_timeseries_schema(device_id, schemas);
            if (ret != common::E_OK) {
                free_device_schema_array(device_schema, initialized);
                return ret;
            }
            if (schemas.empty()) {
                continue;
            }
            if (schemas.size() >
                static_cast<size_t>(std::numeric_limits<int>::max())) {
                free_device_schema_array(device_schema, initialized);
                return common::E_OVERFLOW;
            }

            cur_schema.timeseries_schema = static_cast<TimeseriesSchema*>(
                calloc(schemas.size(), sizeof(TimeseriesSchema)));
            if (cur_schema.timeseries_schema == nullptr) {
                free_device_schema_array(device_schema, initialized);
                return common::E_OOM;
            }
            cur_schema.timeseries_num = static_cast<int>(schemas.size());
            for (size_t i = 0; i < schemas.size(); ++i) {
                const auto& measurement_schema = schemas[i];
                cur_schema.timeseries_schema[i].timeseries_name =
                    strdup(measurement_schema.measurement_name_.c_str());
                if (cur_schema.timeseries_schema[i].timeseries_name ==
                    nullptr) {
                    free_device_schema_array(device_schema, initialized);
                    return common::E_OOM;
                }
                cur_schema.timeseries_schema[i].data_type =
                    static_cast<TSDataType>(measurement_schema.data_type_);
                cur_schema.timeseries_schema[i].encoding =
                    static_cast<TSEncoding>(measurement_schema.encoding_);
                cur_schema.timeseries_schema[i].compression =
                    static_cast<CompressionType>(
                        measurement_schema.compression_type_);
            }
        }
        *out_schemas = device_schema;
        *out_size = static_cast<uint32_t>(device_count);
        return common::E_OK;
    } catch (const std::bad_alloc&) {
        free_device_schema_array(device_schema, initialized);
        return common::E_OOM;
    } catch (...) {
        free_device_schema_array(device_schema, initialized);
        return common::E_FILE_READ_ERR;
    }
}

DeviceSchema* tsfile_reader_get_all_timeseries_schemas(TsFileReader reader,
                                                       uint32_t* size) {
    DeviceSchema* schemas = nullptr;
    const ERRNO ret = tsfile_reader_get_all_timeseries_schemas_checked(
        reader, &schemas, size);
    return ret == common::E_OK ? schemas : nullptr;
}

void tsfile_device_id_free_contents(DeviceID* d) {
    if (d == nullptr) {
        return;
    }
    free(d->path);
    d->path = nullptr;
    free(d->table_name);
    d->table_name = nullptr;
    if (d->segments != nullptr) {
        for (uint32_t k = 0; k < d->segment_count; k++) {
            free(d->segments[k]);
        }
        free(d->segments);
        d->segments = nullptr;
    }
    d->segment_count = 0;
}

namespace {

char* dup_common_string_to_cstr(const common::String& s) {
    if (s.buf_ == nullptr || s.len_ == 0) {
        return strdup("");
    }
    char* p = static_cast<char*>(malloc(static_cast<size_t>(s.len_) + 1U));
    if (p == nullptr) {
        return nullptr;
    }
    memcpy(p, s.buf_, static_cast<size_t>(s.len_));
    p[s.len_] = '\0';
    return p;
}

static TSDataType cpp_stat_type_to_c(common::TSDataType t) {
    return static_cast<TSDataType>(static_cast<uint8_t>(t));
}

void free_timeseries_statistic_heap(TimeseriesStatistic* s) {
    if (s == nullptr) {
        return;
    }
    TsFileStatisticBase* b = tsfile_statistic_base(s);
    if (!b->has_statistic) {
        return;
    }
    switch (b->type) {
        case TS_DATATYPE_STRING:
            free(s->u.string_s.str_min);
            s->u.string_s.str_min = nullptr;
            free(s->u.string_s.str_max);
            s->u.string_s.str_max = nullptr;
            free(s->u.string_s.str_first);
            s->u.string_s.str_first = nullptr;
            free(s->u.string_s.str_last);
            s->u.string_s.str_last = nullptr;
            break;
        case TS_DATATYPE_TEXT:
            free(s->u.text_s.str_first);
            s->u.text_s.str_first = nullptr;
            free(s->u.text_s.str_last);
            s->u.text_s.str_last = nullptr;
            break;
        default:
            break;
    }
}

void clear_timeseries_statistic(TimeseriesStatistic* s) {
    memset(s, 0, sizeof(*s));
    tsfile_statistic_base(s)->type = TS_DATATYPE_INVALID;
}

/**
 * Fills @p out from C++ Statistic. On allocation failure returns E_OOM and
 * clears/frees any partial string fields in @p out.
 */
int fill_timeseries_statistic(storage::Statistic* st,
                              TimeseriesStatistic* out) {
    clear_timeseries_statistic(out);
    if (st == nullptr) {
        return common::E_OK;
    }
    const common::TSDataType t = st->get_type();
    switch (t) {
        case common::BOOLEAN: {
            auto* bs = static_cast<storage::BooleanStatistic*>(st);
            TsFileBoolStatistic* p = &out->u.bool_s;
            p->base.has_statistic = true;
            p->base.type = cpp_stat_type_to_c(common::BOOLEAN);
            p->base.row_count = st->get_count();
            p->base.start_time = st->start_time_;
            p->base.end_time = st->get_end_time();
            p->sum = static_cast<double>(bs->sum_value_);
            p->first_bool = bs->first_value_;
            p->last_bool = bs->last_value_;
            break;
        }
        case common::INT32: {
            auto* is = static_cast<storage::Int32Statistic*>(st);
            TsFileIntStatistic* p = &out->u.int_s;
            p->base.has_statistic = true;
            p->base.type = cpp_stat_type_to_c(common::INT32);
            p->base.row_count = st->get_count();
            p->base.start_time = st->start_time_;
            p->base.end_time = st->get_end_time();
            p->sum = static_cast<double>(is->sum_value_);
            if (p->base.row_count > 0) {
                p->min_int64 = static_cast<int64_t>(is->min_value_);
                p->max_int64 = static_cast<int64_t>(is->max_value_);
                p->first_int64 = static_cast<int64_t>(is->first_value_);
                p->last_int64 = static_cast<int64_t>(is->last_value_);
            }
            break;
        }
        case common::DATE: {
            auto* is = static_cast<storage::Int32Statistic*>(st);
            TsFileIntStatistic* p = &out->u.int_s;
            p->base.has_statistic = true;
            p->base.type = cpp_stat_type_to_c(common::DATE);
            p->base.row_count = st->get_count();
            p->base.start_time = st->start_time_;
            p->base.end_time = st->get_end_time();
            p->sum = static_cast<double>(is->sum_value_);
            if (p->base.row_count > 0) {
                p->min_int64 = static_cast<int64_t>(is->min_value_);
                p->max_int64 = static_cast<int64_t>(is->max_value_);
                p->first_int64 = static_cast<int64_t>(is->first_value_);
                p->last_int64 = static_cast<int64_t>(is->last_value_);
            }
            break;
        }
        case common::INT64: {
            auto* ls = static_cast<storage::Int64Statistic*>(st);
            TsFileIntStatistic* p = &out->u.int_s;
            p->base.has_statistic = true;
            p->base.type = cpp_stat_type_to_c(common::INT64);
            p->base.row_count = st->get_count();
            p->base.start_time = st->start_time_;
            p->base.end_time = st->get_end_time();
            p->sum = ls->sum_value_;
            if (p->base.row_count > 0) {
                p->min_int64 = ls->min_value_;
                p->max_int64 = ls->max_value_;
                p->first_int64 = ls->first_value_;
                p->last_int64 = ls->last_value_;
            }
            break;
        }
        case common::TIMESTAMP: {
            auto* ls = static_cast<storage::Int64Statistic*>(st);
            TsFileIntStatistic* p = &out->u.int_s;
            p->base.has_statistic = true;
            p->base.type = cpp_stat_type_to_c(common::TIMESTAMP);
            p->base.row_count = st->get_count();
            p->base.start_time = st->start_time_;
            p->base.end_time = st->get_end_time();
            p->sum = ls->sum_value_;
            if (p->base.row_count > 0) {
                p->min_int64 = ls->min_value_;
                p->max_int64 = ls->max_value_;
                p->first_int64 = ls->first_value_;
                p->last_int64 = ls->last_value_;
            }
            break;
        }
        case common::FLOAT: {
            auto* fs = static_cast<storage::FloatStatistic*>(st);
            TsFileFloatStatistic* p = &out->u.float_s;
            p->base.has_statistic = true;
            p->base.type = cpp_stat_type_to_c(common::FLOAT);
            p->base.row_count = st->get_count();
            p->base.start_time = st->start_time_;
            p->base.end_time = st->get_end_time();
            p->sum = static_cast<double>(fs->sum_value_);
            if (p->base.row_count > 0) {
                p->min_float64 = static_cast<double>(fs->min_value_);
                p->max_float64 = static_cast<double>(fs->max_value_);
                p->first_float64 = static_cast<double>(fs->first_value_);
                p->last_float64 = static_cast<double>(fs->last_value_);
            }
            break;
        }
        case common::DOUBLE: {
            auto* ds = static_cast<storage::DoubleStatistic*>(st);
            TsFileFloatStatistic* p = &out->u.float_s;
            p->base.has_statistic = true;
            p->base.type = cpp_stat_type_to_c(common::DOUBLE);
            p->base.row_count = st->get_count();
            p->base.start_time = st->start_time_;
            p->base.end_time = st->get_end_time();
            p->sum = ds->sum_value_;
            if (p->base.row_count > 0) {
                p->min_float64 = ds->min_value_;
                p->max_float64 = ds->max_value_;
                p->first_float64 = ds->first_value_;
                p->last_float64 = ds->last_value_;
            }
            break;
        }
        case common::STRING: {
            auto* ss = static_cast<storage::StringStatistic*>(st);
            TsFileStringStatistic* p = &out->u.string_s;
            p->base.has_statistic = true;
            p->base.type = cpp_stat_type_to_c(common::STRING);
            p->base.row_count = st->get_count();
            p->base.start_time = st->start_time_;
            p->base.end_time = st->get_end_time();
            p->str_min = dup_common_string_to_cstr(ss->min_value_);
            if (p->str_min == nullptr) {
                free_timeseries_statistic_heap(out);
                clear_timeseries_statistic(out);
                return common::E_OOM;
            }
            p->str_max = dup_common_string_to_cstr(ss->max_value_);
            if (p->str_max == nullptr) {
                free_timeseries_statistic_heap(out);
                clear_timeseries_statistic(out);
                return common::E_OOM;
            }
            p->str_first = dup_common_string_to_cstr(ss->first_value_);
            if (p->str_first == nullptr) {
                free_timeseries_statistic_heap(out);
                clear_timeseries_statistic(out);
                return common::E_OOM;
            }
            p->str_last = dup_common_string_to_cstr(ss->last_value_);
            if (p->str_last == nullptr) {
                free_timeseries_statistic_heap(out);
                clear_timeseries_statistic(out);
                return common::E_OOM;
            }
            break;
        }
        case common::TEXT: {
            auto* ts = static_cast<storage::TextStatistic*>(st);
            TsFileTextStatistic* p = &out->u.text_s;
            p->base.has_statistic = true;
            p->base.type = cpp_stat_type_to_c(common::TEXT);
            p->base.row_count = st->get_count();
            p->base.start_time = st->start_time_;
            p->base.end_time = st->get_end_time();
            p->str_first = dup_common_string_to_cstr(ts->first_value_);
            if (p->str_first == nullptr) {
                free_timeseries_statistic_heap(out);
                clear_timeseries_statistic(out);
                return common::E_OOM;
            }
            p->str_last = dup_common_string_to_cstr(ts->last_value_);
            if (p->str_last == nullptr) {
                free_timeseries_statistic_heap(out);
                clear_timeseries_statistic(out);
                return common::E_OOM;
            }
            break;
        }
        default: {
            TsFileStatisticBase* b = tsfile_statistic_base(out);
            b->has_statistic = true;
            b->type = TS_DATATYPE_INVALID;
            b->row_count = st->get_count();
            b->start_time = st->start_time_;
            b->end_time = st->get_end_time();
            break;
        }
    }
    return common::E_OK;
}

int fill_timeline_statistic(storage::ITimeseriesIndex* idx,
                            TimeseriesStatistic* out) {
    clear_timeseries_statistic(out);
    if (idx == nullptr) {
        return common::E_OK;
    }

    auto* aligned_idx = dynamic_cast<storage::AlignedTimeseriesIndex*>(idx);
    if (aligned_idx != nullptr && aligned_idx->time_ts_idx_ != nullptr &&
        aligned_idx->time_ts_idx_->get_statistic() != nullptr) {
        auto* st = aligned_idx->time_ts_idx_->get_statistic();
        TsFileStatisticBase* b = tsfile_statistic_base(out);
        b->has_statistic = true;
        b->type = TS_DATATYPE_VECTOR;
        b->row_count = st->get_count();
        b->start_time = st->start_time_;
        b->end_time = st->get_end_time();
        return common::E_OK;
    }

    if (idx->get_statistic() != nullptr &&
        idx->get_time_chunk_meta_list() == nullptr) {
        auto* st = idx->get_statistic();
        TsFileStatisticBase* b = tsfile_statistic_base(out);
        b->has_statistic = true;
        b->type = TS_DATATYPE_VECTOR;
        b->row_count = st->get_count();
        b->start_time = st->start_time_;
        b->end_time = st->get_end_time();
        return common::E_OK;
    }

    auto* list = idx->get_time_chunk_meta_list();
    if (list == nullptr) {
        list = idx->get_chunk_meta_list();
    }
    if (list == nullptr) {
        return common::E_OK;
    }

    int64_t row_count = 0;
    int64_t start_time = 0;
    int64_t end_time = 0;
    bool has_statistic = false;
    for (auto it = list->begin(); it != list->end(); it++) {
        auto* chunk_meta = it.get();
        if (chunk_meta == nullptr || chunk_meta->statistic_ == nullptr ||
            chunk_meta->statistic_->count_ <= 0) {
            continue;
        }
        if (!has_statistic) {
            start_time = chunk_meta->statistic_->start_time_;
            end_time = chunk_meta->statistic_->end_time_;
            has_statistic = true;
        } else {
            start_time =
                std::min(start_time, chunk_meta->statistic_->start_time_);
            end_time = std::max(end_time, chunk_meta->statistic_->end_time_);
        }
        row_count += chunk_meta->statistic_->count_;
    }

    if (!has_statistic) {
        return common::E_OK;
    }

    TsFileStatisticBase* b = tsfile_statistic_base(out);
    b->has_statistic = true;
    b->type = TS_DATATYPE_VECTOR;
    b->row_count = row_count;
    b->start_time = start_time;
    b->end_time = end_time;
    return common::E_OK;
}

void free_device_timeseries_metadata_entries_partial(
    DeviceTimeseriesMetadataEntry* entries, size_t filled_count) {
    if (entries == nullptr) {
        return;
    }
    for (size_t i = 0; i < filled_count; i++) {
        tsfile_device_id_free_contents(&entries[i].device);
        if (entries[i].timeseries != nullptr) {
            for (uint32_t j = 0; j < entries[i].timeseries_count; j++) {
                free_timeseries_statistic_heap(
                    &entries[i].timeseries[j].statistic);
                free_timeseries_statistic_heap(
                    &entries[i].timeseries[j].timeline_statistic);
                free(entries[i].timeseries[j].measurement_name);
            }
            free(entries[i].timeseries);
            entries[i].timeseries = nullptr;
        }
    }
    free(entries);
}

/**
 * Copies path, table name, and segment strings from IDeviceID into heap
 * buffers. On failure, frees any partial allocations and returns an error code.
 */
int duplicate_ideviceid_to_device_fields(storage::IDeviceID* id,
                                         char** out_path, char** out_table_name,
                                         uint32_t* out_segment_count,
                                         char*** out_segments) {
    *out_path = nullptr;
    *out_table_name = nullptr;
    *out_segment_count = 0;
    *out_segments = nullptr;
    auto cleanup = [&]() {
        free(*out_path);
        free(*out_table_name);
        *out_path = nullptr;
        *out_table_name = nullptr;
    };
    try {
        if (id == nullptr) {
            *out_path = strdup("");
            *out_table_name = strdup("");
            if (*out_path == nullptr || *out_table_name == nullptr) {
                free(*out_path);
                free(*out_table_name);
                *out_path = nullptr;
                *out_table_name = nullptr;
                return common::E_OOM;
            }
            return common::E_OK;
        }
        const std::string dname = id->get_device_name();
        *out_path = strdup(dname.c_str());
        if (*out_path == nullptr) {
            return common::E_OOM;
        }
#ifdef ENABLE_TEST
        using common::g_all_inject_points;
        using common::InjectPoint;
        DBUG_EXECUTE_IF("device_id_copy_oom", throw std::bad_alloc(););
        DBUG_EXECUTE_IF("device_id_copy_error", throw common::E_FILE_READ_ERR;);
#endif
        const std::string tname = id->get_table_name();
        *out_table_name = strdup(tname.c_str());
        if (*out_table_name == nullptr) {
            free(*out_path);
            *out_path = nullptr;
            return common::E_OOM;
        }
        const int n = id->segment_num();
        if (n <= 0) {
            return common::E_OK;
        }
        const auto& segs = id->get_segments();
        auto* seg_arr =
            static_cast<char**>(malloc(sizeof(char*) * static_cast<size_t>(n)));
        if (seg_arr == nullptr) {
            free(*out_table_name);
            *out_table_name = nullptr;
            free(*out_path);
            *out_path = nullptr;
            return common::E_OOM;
        }
        memset(seg_arr, 0, sizeof(char*) * static_cast<size_t>(n));
        for (int i = 0; i < n; i++) {
            const std::string* ps =
                (static_cast<size_t>(i) < segs.size()) ? segs[i] : nullptr;
            // A null tag segment is exposed as a NULL pointer so callers can
            // distinguish a missing/null tag from the literal string "null".
            if (ps == nullptr) {
                seg_arr[i] = nullptr;
                continue;
            }
            seg_arr[i] = strdup(ps->c_str());
            if (seg_arr[i] == nullptr) {
                for (int j = 0; j < i; j++) {
                    free(seg_arr[j]);
                }
                free(seg_arr);
                free(*out_table_name);
                *out_table_name = nullptr;
                free(*out_path);
                *out_path = nullptr;
                return common::E_OOM;
            }
        }
        *out_segment_count = static_cast<uint32_t>(n);
        *out_segments = seg_arr;
        return common::E_OK;
    } catch (const std::bad_alloc&) {
        cleanup();
        return common::E_OOM;
    } catch (...) {
        cleanup();
        return common::E_FILE_READ_ERR;
    }
}

int fill_device_id_from_ideviceid(storage::IDeviceID* id, DeviceID* out) {
    memset(out, 0, sizeof(*out));
    return duplicate_ideviceid_to_device_fields(
        id, &out->path, &out->table_name, &out->segment_count, &out->segments);
}

ERRNO populate_c_metadata_map_from_cpp(
    storage::DeviceTimeseriesMetadataMap& cpp_map,
    DeviceTimeseriesMetadataMap* out_map) {
    try {
        if (cpp_map.empty()) {
            return common::E_OK;
        }
        const uint32_t dev_n = static_cast<uint32_t>(cpp_map.size());
        auto* entries = static_cast<DeviceTimeseriesMetadataEntry*>(
            malloc(sizeof(DeviceTimeseriesMetadataEntry) * dev_n));
        if (entries == nullptr) {
            return common::E_OOM;
        }
        memset(entries, 0, sizeof(DeviceTimeseriesMetadataEntry) * dev_n);
        auto cleanup = [dev_n](DeviceTimeseriesMetadataEntry* data) {
            free_device_timeseries_metadata_entries_partial(data, dev_n);
        };
        std::unique_ptr<DeviceTimeseriesMetadataEntry, decltype(cleanup)> guard(
            entries, cleanup);
        size_t di = 0;
        for (const auto& kv : cpp_map) {
            DeviceTimeseriesMetadataEntry& e = entries[di];
            const int dup_rc = fill_device_id_from_ideviceid(
                kv.first ? kv.first.get() : nullptr, &e.device);
            if (dup_rc != common::E_OK) {
                return dup_rc;
            }
            const auto& vec = kv.second;
            uint32_t n_ts = 0;
            for (const auto& idx_nz : vec) {
                if (idx_nz != nullptr) {
                    n_ts++;
                }
            }
            e.timeseries_count = n_ts;
            if (e.timeseries_count == 0) {
                e.timeseries = nullptr;
                di++;
                continue;
            }
            e.timeseries = static_cast<TimeseriesMetadata*>(
                malloc(sizeof(TimeseriesMetadata) * e.timeseries_count));
            if (e.timeseries == nullptr) {
                return common::E_OOM;
            }
            memset(e.timeseries, 0,
                   sizeof(TimeseriesMetadata) * e.timeseries_count);
            uint32_t slot = 0;
            for (const auto& idx : vec) {
                if (idx == nullptr) {
                    continue;
                }
                TimeseriesMetadata& m = e.timeseries[slot];
                common::String mn = idx->get_measurement_name();
                m.measurement_name = strdup(mn.to_std_string().c_str());
                if (m.measurement_name == nullptr) {
                    return common::E_OOM;
                }
#ifdef ENABLE_TEST
                using common::g_all_inject_points;
                using common::InjectPoint;
                DBUG_EXECUTE_IF("timeseries_metadata_copy_oom",
                                throw std::bad_alloc(););
                DBUG_EXECUTE_IF("timeseries_metadata_copy_error",
                                throw common::E_FILE_READ_ERR;);
#endif
                auto* aligned_idx =
                    dynamic_cast<storage::AlignedTimeseriesIndex*>(idx.get());
                if (aligned_idx != nullptr &&
                    aligned_idx->value_ts_idx_ != nullptr) {
                    m.data_type = static_cast<TSDataType>(
                        aligned_idx->value_ts_idx_->get_data_type());
                    const storage::TimeseriesIndex* value_idx =
                        aligned_idx->value_ts_idx_;
                    const storage::TimeseriesIndex* time_idx =
                        aligned_idx->time_ts_idx_;
                    if (value_idx->get_metadata_offset() >= 0) {
                        m.value_metadata_offset = static_cast<uint64_t>(
                            value_idx->get_metadata_offset());
                        m.value_metadata_length =
                            value_idx->get_metadata_length();
                    }
                    if (time_idx != nullptr &&
                        time_idx->get_metadata_offset() >= 0) {
                        m.time_metadata_offset = static_cast<uint64_t>(
                            time_idx->get_metadata_offset());
                        m.time_metadata_length =
                            time_idx->get_metadata_length();
                    }
                    m.layout = 1;
                } else {
                    m.data_type = static_cast<TSDataType>(idx->get_data_type());
                    const storage::TimeseriesIndex* value_idx =
                        dynamic_cast<const storage::TimeseriesIndex*>(
                            idx.get());
                    if (value_idx != nullptr &&
                        value_idx->get_metadata_offset() >= 0) {
                        m.value_metadata_offset = static_cast<uint64_t>(
                            value_idx->get_metadata_offset());
                        m.value_metadata_length =
                            value_idx->get_metadata_length();
                    }
                }
                storage::Statistic* st = idx->get_statistic();
                int32_t chunk_cnt = 0;
                auto* cl = aligned_idx != nullptr
                               ? idx->get_value_chunk_meta_list()
                               : idx->get_chunk_meta_list();
                if (cl != nullptr) {
                    chunk_cnt = static_cast<int32_t>(cl->size());
                }
                m.chunk_meta_count = chunk_cnt;
                if (chunk_cnt >= 0 && m.value_metadata_length > 0) {
                    m.locator_flags |= 1;
                }
                if (aligned_idx != nullptr) {
                    auto* time_chunks = idx->get_time_chunk_meta_list();
                    if (time_chunks != nullptr) {
                        m.time_chunk_meta_count =
                            static_cast<uint32_t>(time_chunks->size());
                    }
                    if (m.time_metadata_length == 0 ||
                        m.time_chunk_meta_count !=
                            static_cast<uint32_t>(chunk_cnt)) {
                        m.locator_flags &= ~static_cast<uint16_t>(1);
                    }
                }
                const int st_rc = fill_timeseries_statistic(st, &m.statistic);
                if (st_rc != common::E_OK) {
                    return st_rc;
                }
                const int timeline_st_rc =
                    fill_timeline_statistic(idx.get(), &m.timeline_statistic);
                if (timeline_st_rc != common::E_OK) {
                    return timeline_st_rc;
                }
                slot++;
            }
            di++;
        }
        out_map->entries = guard.release();
        out_map->device_count = dev_n;
        return common::E_OK;
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    } catch (...) {
        return common::E_FILE_READ_ERR;
    }
}

}  // namespace

void tsfile_free_device_id_array(DeviceID* devices, uint32_t length) {
    if (devices == nullptr) {
        return;
    }
    for (uint32_t i = 0; i < length; i++) {
        tsfile_device_id_free_contents(&devices[i]);
    }
    free(devices);
}

ERRNO tsfile_reader_get_all_devices(TsFileReader reader, DeviceID** out_devices,
                                    uint32_t* out_length) {
    if (reader == nullptr || out_devices == nullptr || out_length == nullptr) {
        return common::E_INVALID_ARG;
    }
    *out_devices = nullptr;
    *out_length = 0;
    auto* r = static_cast<storage::TsFileReader*>(reader);
    std::vector<std::shared_ptr<storage::IDeviceID>> ids;
    const int ret = r->get_all_devices(ids);
    if (ret != common::E_OK) {
        return ret;
    }
    if (ids.empty()) {
        return common::E_OK;
    }
    auto* arr = static_cast<DeviceID*>(malloc(sizeof(DeviceID) * ids.size()));
    if (arr == nullptr) {
        return common::E_OOM;
    }
    memset(arr, 0, sizeof(DeviceID) * ids.size());
    for (size_t i = 0; i < ids.size(); i++) {
        const int rc = fill_device_id_from_ideviceid(ids[i].get(), &arr[i]);
        if (rc != common::E_OK) {
            tsfile_free_device_id_array(arr, static_cast<uint32_t>(i));
            return rc;
        }
    }
    *out_devices = arr;
    *out_length = static_cast<uint32_t>(ids.size());
    return common::E_OK;
}

ERRNO tsfile_reader_get_timeseries_metadata_all(
    TsFileReader reader, DeviceTimeseriesMetadataMap* out_map) {
    if (out_map != nullptr) {
        out_map->entries = nullptr;
        out_map->device_count = 0;
    }
    if (reader == nullptr || out_map == nullptr) {
        return common::E_INVALID_ARG;
    }
    try {
        auto* r = static_cast<storage::TsFileReader*>(reader);
        storage::DeviceTimeseriesMetadataMap cpp_map;
        const int ret = r->get_timeseries_metadata(cpp_map);
        if (ret != common::E_OK) return ret;
        return populate_c_metadata_map_from_cpp(cpp_map, out_map);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    } catch (...) {
        return common::E_FILE_READ_ERR;
    }
}

ERRNO tsfile_reader_get_timeseries_metadata_for_devices(
    TsFileReader reader, const DeviceID* devices, uint32_t length,
    DeviceTimeseriesMetadataMap* out_map) {
    if (out_map != nullptr) {
        out_map->entries = nullptr;
        out_map->device_count = 0;
    }
    if (reader == nullptr || out_map == nullptr) {
        return common::E_INVALID_ARG;
    }
    if (length == 0) {
        return common::E_OK;
    }
    if (devices == nullptr) {
        return common::E_INVALID_ARG;
    }
    for (uint32_t i = 0; i < length; i++) {
        if (devices[i].segment_count > 0) {
            if (devices[i].segments == nullptr ||
                devices[i].segments[0] == nullptr) {
                return common::E_INVALID_ARG;
            }
        } else if (devices[i].path == nullptr) {
            return common::E_INVALID_ARG;
        }
    }
    try {
        auto* r = static_cast<storage::TsFileReader*>(reader);
        std::vector<std::shared_ptr<storage::IDeviceID>> query_ids;
        query_ids.reserve(length);
        for (uint32_t i = 0; i < length; i++) {
            const DeviceID& device = devices[i];
            if (device.segment_count == 0) {
                query_ids.push_back(
                    std::make_shared<storage::StringArrayDeviceID>(
                        std::string(device.path)));
                continue;
            }
            std::vector<std::string> values;
            values.reserve(device.segment_count);
            std::vector<std::string*> segments(device.segment_count, nullptr);
            for (uint32_t j = 0; j < device.segment_count; ++j) {
                if (device.segments[j] != nullptr) {
                    values.emplace_back(device.segments[j]);
                    segments[j] = &values.back();
                }
            }
            query_ids.push_back(
                std::make_shared<storage::StringArrayDeviceID>(segments));
        }
        storage::DeviceTimeseriesMetadataMap cpp_map;
        const int ret = r->get_timeseries_metadata(query_ids, cpp_map);
        if (ret != common::E_OK) return ret;
        return populate_c_metadata_map_from_cpp(cpp_map, out_map);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    } catch (...) {
        return common::E_FILE_READ_ERR;
    }
}

void tsfile_free_device_timeseries_metadata_map(
    DeviceTimeseriesMetadataMap* map) {
    if (map == nullptr) {
        return;
    }
    free_device_timeseries_metadata_entries_partial(map->entries,
                                                    map->device_count);
    map->entries = nullptr;
    map->device_count = 0;
}

void tsfile_free_tsfile_properties(TsFileProperty* properties,
                                   uint32_t length) {
    if (properties == nullptr) {
        return;
    }
    for (uint32_t i = 0; i < length; i++) {
        free(properties[i].key);
        properties[i].key = nullptr;
        free(properties[i].value);
        properties[i].value = nullptr;
        properties[i].key_len = 0;
        properties[i].value_len = 0;
        properties[i].is_null = false;
    }
    free(properties);
}

ERRNO tsfile_reader_get_tsfile_properties(TsFileReader reader,
                                          TsFileProperty** out_properties,
                                          uint32_t* out_length) {
    if (out_properties == nullptr || out_length == nullptr) {
        return common::E_INVALID_ARG;
    }
    *out_properties = nullptr;
    *out_length = 0;
    if (reader == nullptr) {
        return common::E_INVALID_ARG;
    }

    try {
        auto* r = static_cast<storage::TsFileReader*>(reader);
        storage::TsFileProperties cpp_properties = r->get_tsfile_properties();
        if (cpp_properties.size() >
                static_cast<size_t>(std::numeric_limits<uint32_t>::max()) ||
            cpp_properties.size() >
                std::numeric_limits<size_t>::max() / sizeof(TsFileProperty)) {
            return common::E_OUT_OF_RANGE;
        }
        if (cpp_properties.empty()) {
            return common::E_OK;
        }

        auto* properties = static_cast<TsFileProperty*>(
            malloc(sizeof(TsFileProperty) * cpp_properties.size()));
        if (properties == nullptr) {
            return common::E_OOM;
        }
        memset(properties, 0, sizeof(TsFileProperty) * cpp_properties.size());

        uint32_t property_index = 0;
        for (const auto& cpp_property : cpp_properties) {
            TsFileProperty& property = properties[property_index];
            if (cpp_property.first.size() >
                static_cast<size_t>(std::numeric_limits<uint32_t>::max())) {
                tsfile_free_tsfile_properties(properties, property_index);
                return common::E_OUT_OF_RANGE;
            }
            property.key_len = static_cast<uint32_t>(cpp_property.first.size());
            property.key = static_cast<char*>(
                malloc(static_cast<size_t>(property.key_len) + 1U));
            if (property.key == nullptr) {
                tsfile_free_tsfile_properties(properties, property_index + 1);
                return common::E_OOM;
            }
            if (property.key_len > 0) {
                memcpy(property.key, cpp_property.first.data(),
                       property.key_len);
            }
            property.key[property.key_len] = '\0';

            const storage::TsFilePropertyValue& cpp_value = cpp_property.second;
            property.is_null = cpp_value.is_null;
            if (!cpp_value.is_null) {
                if (cpp_value.value.size() >
                    static_cast<size_t>(std::numeric_limits<uint32_t>::max())) {
                    tsfile_free_tsfile_properties(properties,
                                                  property_index + 1);
                    return common::E_OUT_OF_RANGE;
                }
                property.value_len =
                    static_cast<uint32_t>(cpp_value.value.size());
                if (property.value_len > 0) {
                    property.value =
                        static_cast<uint8_t*>(malloc(property.value_len));
                    if (property.value == nullptr) {
                        tsfile_free_tsfile_properties(properties,
                                                      property_index + 1);
                        return common::E_OOM;
                    }
                    memcpy(property.value, cpp_value.value.data(),
                           property.value_len);
                }
            }
            property_index++;
        }

        *out_properties = properties;
        *out_length = static_cast<uint32_t>(cpp_properties.size());
        return common::E_OK;
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    }
}

// delete pointer
void _free_tsfile_ts_record(TsRecord* record) {
    if (*record != nullptr) {
        delete static_cast<storage::TsRecord*>(*record);
    }
    *record = nullptr;
}

void free_tablet(Tablet* tablet) {
    if (*tablet != nullptr) {
        delete static_cast<storage::Tablet*>(*tablet);
    }
    *tablet = nullptr;
}

void free_tsfile_result_set(ResultSet* result_set) {
    if (*result_set != nullptr) {
        delete static_cast<storage::ResultSet*>(*result_set);
    }
    *result_set = nullptr;
}

void free_result_set_meta_data(ResultSetMetaData result_set_meta_data) {
    for (int i = 0; i < result_set_meta_data.column_num; i++) {
        free(result_set_meta_data.column_names[i]);
    }
    free(result_set_meta_data.column_names);
    free(result_set_meta_data.data_types);
}

void free_device_schema(DeviceSchema schema) {
    free(schema.device_name);
    for (int i = 0; i < schema.timeseries_num; i++) {
        free_timeseries_schema(schema.timeseries_schema[i]);
    }
    free(schema.timeseries_schema);
}
void free_timeseries_schema(TimeseriesSchema schema) {
    free(schema.timeseries_name);
}
void free_table_schema(TableSchema schema) {
    free(schema.table_name);
    for (int i = 0; i < schema.column_num; i++) {
        free_column_schema(schema.column_schemas[i]);
    }
    if (schema.column_num > 0) {
        free(schema.column_schemas);
    }
}
void free_column_schema(ColumnSchema schema) { free(schema.column_name); }

void free_write_file(WriteFile* write_file) {
    auto f = static_cast<storage::WriteFile*>(*write_file);
    delete f;
    *write_file = nullptr;
}

// For Python API
TsFileWriter _tsfile_writer_new(const char* pathname, uint64_t memory_threshold,
                                ERRNO* err_code) {
    init_tsfile_config();
    auto writer = new storage::TsFileWriter();
    int flags = O_WRONLY | O_CREAT | O_TRUNC;
#ifdef _WIN32
    flags |= O_BINARY;
#endif
    int ret = writer->open(pathname, flags, 0644);
    common::g_config_value_.chunk_group_size_threshold_ = memory_threshold;
    if (ret != common::E_OK) {
        delete writer;
        *err_code = ret;
        return nullptr;
    }
    return writer;
}

Tablet _tablet_new_with_target_name(const char* device_id,
                                    char** column_name_list,
                                    TSDataType* data_types, int column_num,
                                    int max_rows) {
    std::vector<std::string> measurement_list;
    std::vector<common::TSDataType> data_type_list;
    for (int i = 0; i < column_num; i++) {
        measurement_list.emplace_back(column_name_list[i]);
        data_type_list.push_back(
            static_cast<common::TSDataType>(*(data_types + i)));
    }
    if (device_id != nullptr) {
        return new storage::Tablet(device_id, &measurement_list,
                                   &data_type_list, max_rows);
    } else {
        return new storage::Tablet(measurement_list, data_type_list, max_rows);
    }
}

ERRNO _tsfile_writer_register_table(TsFileWriter writer, TableSchema* schema) {
    if (writer == nullptr || schema == nullptr ||
        schema->column_schemas == nullptr || schema->table_name == nullptr) {
        return common::E_INVALID_ARG;
    }
    if (schema->column_num <= 0) {
        return common::E_INVALID_SCHEMA;
    }
    std::vector<storage::MeasurementSchema*> measurement_schemas;
    std::vector<common::ColumnCategory> column_categories;
    measurement_schemas.resize(schema->column_num);
    for (int i = 0; i < schema->column_num; i++) {
        ColumnSchema* cur_schema = schema->column_schemas + i;
        measurement_schemas[i] = new storage::MeasurementSchema(
            cur_schema->column_name,
            static_cast<common::TSDataType>(cur_schema->data_type));
        column_categories.push_back(
            static_cast<common::ColumnCategory>(cur_schema->column_category));
    }
    auto tsfile_writer = static_cast<storage::TsFileWriter*>(writer);
    return tsfile_writer->register_table(std::make_shared<storage::TableSchema>(
        schema->table_name, measurement_schemas, column_categories));
}

ERRNO _tsfile_writer_register_timeseries(TsFileWriter writer,
                                         const char* device_id,
                                         const TimeseriesSchema* schema) {
    auto* w = static_cast<storage::TsFileWriter*>(writer);

    int ret = w->register_timeseries(
        device_id,
        storage::MeasurementSchema(
            schema->timeseries_name,
            static_cast<common::TSDataType>(schema->data_type),
            static_cast<common::TSEncoding>(schema->encoding),
            static_cast<common::CompressionType>(schema->compression)));
    return ret;
}

ERRNO _tsfile_writer_register_device(TsFileWriter writer,
                                     const device_schema* device_schema) {
    auto* w = static_cast<storage::TsFileWriter*>(writer);
    for (int column_id = 0; column_id < device_schema->timeseries_num;
         column_id++) {
        TimeseriesSchema schema = device_schema->timeseries_schema[column_id];
        const ERRNO ret = w->register_timeseries(
            device_schema->device_name,
            storage::MeasurementSchema(
                schema.timeseries_name,
                static_cast<common::TSDataType>(schema.data_type),
                static_cast<common::TSEncoding>(schema.encoding),
                static_cast<common::CompressionType>(schema.compression)));
        if (ret != common::E_OK) {
            return ret;
        }
    }
    return common::E_OK;
}

ERRNO _tsfile_writer_write_tablet(TsFileWriter writer, Tablet tablet) {
    auto* w = static_cast<storage::TsFileWriter*>(writer);
    const auto* tbl = static_cast<storage::Tablet*>(tablet);
    return w->write_tablet(*tbl);
}

ERRNO _tsfile_writer_write_table(TsFileWriter writer, Tablet tablet) {
    auto* w = static_cast<storage::TsFileWriter*>(writer);
    auto* tbl = static_cast<storage::Tablet*>(tablet);
    return w->write_table(*tbl);
}

ERRNO _tsfile_writer_write_arrow_table(TsFileWriter writer,
                                       const char* table_name,
                                       ArrowArray* array, ArrowSchema* schema,
                                       int time_col_index) {
    auto* w = static_cast<storage::TsFileWriter*>(writer);
    std::shared_ptr<storage::TableSchema> reg_schema =
        w->get_table_schema(table_name ? std::string(table_name) : "");
    storage::Tablet* tablet = nullptr;
    int ret = arrow::ArrowStructToTablet(
        table_name, array, schema, reg_schema.get(), &tablet, time_col_index);
    if (ret != common::E_OK) return ret;
    ret = w->write_table(*tablet);
    delete tablet;
    return ret;
}

ERRNO _tsfile_writer_write_ts_record(TsFileWriter writer, TsRecord data) {
    auto* w = static_cast<storage::TsFileWriter*>(writer);
    const storage::TsRecord* record = static_cast<storage::TsRecord*>(data);
    const int ret = w->write_record(*record);
    return ret;
}

ERRNO _tsfile_writer_close(TsFileWriter writer) {
    auto* w = static_cast<storage::TsFileWriter*>(writer);
    int ret = w->close();
    if (ret != common::E_OK) {
        return ret;
    }
    delete w;
    return ret;
}

ERRNO _tsfile_writer_flush(TsFileWriter writer) {
    auto* w = static_cast<storage::TsFileWriter*>(writer);
    return w->flush();
}

ERRNO _tsfile_writer_add_tsfile_property(TsFileWriter writer, const char* key,
                                         uint32_t key_len, const uint8_t* value,
                                         uint32_t value_len) {
    if (writer == nullptr || key == nullptr ||
        (value == nullptr && value_len > 0)) {
        return common::E_INVALID_ARG;
    }
    if (key_len > static_cast<uint32_t>(std::numeric_limits<int32_t>::max()) ||
        value_len >
            static_cast<uint32_t>(std::numeric_limits<int32_t>::max())) {
        return common::E_OUT_OF_RANGE;
    }
    try {
        auto* w = static_cast<storage::TsFileWriter*>(writer);
        return w->add_tsfile_property(std::string(key, key_len), value,
                                      value_len);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    }
}

TsFileGenericWriter tsfile_generic_writer_new(const char* pathname,
                                              uint64_t memory_threshold,
                                              ERRNO* err_code) {
    // Public C entry point: guard null arguments before the private delegate
    // dereferences them, and keep std::bad_alloc from crossing the C ABI.
    // See tsfile_writer_new() above for the null-guard rationale.
    if (err_code == nullptr) {
        return nullptr;
    }
    *err_code = common::E_OK;
    if (pathname == nullptr) {
        *err_code = common::E_INVALID_ARG;
        return nullptr;
    }
    try {
        return _tsfile_writer_new(pathname, memory_threshold, err_code);
    } catch (const std::bad_alloc&) {
        *err_code = common::E_OOM;
        return nullptr;
    }
}

Tablet tablet_new_with_target_name(const char* target_name,
                                   char** column_name_list,
                                   TSDataType* data_types, int column_num,
                                   int max_rows) {
    // Public C entry point: reject arguments that would make the private
    // delegate read out-of-bounds memory (negative column_num, null lists or
    // null column names) before touching caller buffers. The native Tablet
    // constructor asserts 0 < max_rows < (1 << 30); reject values outside
    // that domain here instead of letting an assertion abort the host
    // process. Returns NULL on invalid arguments or allocation failure.
    if (column_num < 0 || max_rows <= 0 || max_rows >= (1 << 30)) {
        return nullptr;
    }
    if (column_num > 0) {
        if (column_name_list == nullptr || data_types == nullptr) {
            return nullptr;
        }
        for (int i = 0; i < column_num; i++) {
            if (column_name_list[i] == nullptr) {
                return nullptr;
            }
        }
    }
    try {
        return _tablet_new_with_target_name(target_name, column_name_list,
                                            data_types, column_num, max_rows);
    } catch (const std::bad_alloc&) {
        return nullptr;
    }
}

ERRNO tsfile_generic_writer_register_table(TsFileGenericWriter writer,
                                           TableSchema* schema) {
    if (writer == nullptr) {
        return common::E_INVALID_ARG;
    }
    try {
        return _tsfile_writer_register_table(writer, schema);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    }
}

ERRNO tsfile_generic_writer_register_timeseries(
    TsFileGenericWriter writer, const char* device_id,
    const TimeseriesSchema* schema) {
    if (writer == nullptr || device_id == nullptr || schema == nullptr ||
        schema->timeseries_name == nullptr) {
        return common::E_INVALID_ARG;
    }
    try {
        return _tsfile_writer_register_timeseries(writer, device_id, schema);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    }
}

ERRNO tsfile_generic_writer_register_device(TsFileGenericWriter writer,
                                            const DeviceSchema* device_schema) {
    if (writer == nullptr || device_schema == nullptr ||
        device_schema->device_name == nullptr ||
        device_schema->timeseries_num < 0) {
        return common::E_INVALID_ARG;
    }
    if (device_schema->timeseries_num > 0 &&
        device_schema->timeseries_schema == nullptr) {
        return common::E_INVALID_ARG;
    }
    for (int i = 0; i < device_schema->timeseries_num; i++) {
        if (device_schema->timeseries_schema[i].timeseries_name == nullptr) {
            return common::E_INVALID_ARG;
        }
    }
    try {
        return _tsfile_writer_register_device(writer, device_schema);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    }
}

ERRNO tsfile_generic_writer_write_tree_tablet(TsFileGenericWriter writer,
                                              Tablet tablet) {
    if (writer == nullptr || tablet == nullptr) {
        return common::E_INVALID_ARG;
    }
    try {
        return _tsfile_writer_write_tablet(writer, tablet);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    }
}

ERRNO tsfile_generic_writer_write_table_tablet(TsFileGenericWriter writer,
                                               Tablet tablet) {
    if (writer == nullptr || tablet == nullptr) {
        return common::E_INVALID_ARG;
    }
    try {
        return _tsfile_writer_write_table(writer, tablet);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    }
}

ERRNO tsfile_generic_writer_flush(TsFileGenericWriter writer) {
    if (writer == nullptr) {
        return common::E_INVALID_ARG;
    }
    try {
        return _tsfile_writer_flush(writer);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    }
}

ERRNO tsfile_generic_writer_add_tsfile_property(TsFileGenericWriter writer,
                                                const char* key,
                                                uint32_t key_len,
                                                const uint8_t* value,
                                                uint32_t value_len) {
    return _tsfile_writer_add_tsfile_property(writer, key, key_len, value,
                                              value_len);
}

ERRNO tsfile_generic_writer_close(TsFileGenericWriter writer) {
    // Matches the public tsfile_writer_close(): a null handle is treated as
    // already closed.
    if (writer == nullptr) {
        return common::E_OK;
    }
    try {
        return _tsfile_writer_close(writer);
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    }
}

ResultSet _tsfile_reader_query_device(TsFileReader reader,
                                      const char* device_name,
                                      char** sensor_name, uint32_t sensor_num,
                                      Timestamp start_time, Timestamp end_time,
                                      ERRNO* err_code) {
    auto* r = static_cast<storage::TsFileReader*>(reader);
    std::vector<std::string> selected_paths;
    selected_paths.reserve(sensor_num);
    for (uint32_t i = 0; i < sensor_num; i++) {
        selected_paths.push_back(std::string(device_name) + "." +
                                 std::string(sensor_name[i]));
    }
    storage::ResultSet* qds = nullptr;
    *err_code = r->query(selected_paths, start_time, end_time, qds);
    return qds;
}

// ============== Tag Filter API Implementation ==============

#define DEFINE_TAG_FILTER_FACTORY(name, op)                                   \
    TagFilterHandle tsfile_tag_filter_##name(                                 \
        TsFileReader reader, const char* table_name, const char* column_name, \
        const char* value) {                                                  \
        ERRNO error_code = common::E_OK;                                      \
        return tsfile_tag_filter_create(reader, table_name, column_name,      \
                                        value, TAG_FILTER_##op, &error_code); \
    }

DEFINE_TAG_FILTER_FACTORY(eq, EQ)
DEFINE_TAG_FILTER_FACTORY(neq, NEQ)
DEFINE_TAG_FILTER_FACTORY(lt, LT)
DEFINE_TAG_FILTER_FACTORY(lteq, LTEQ)
DEFINE_TAG_FILTER_FACTORY(gt, GT)
DEFINE_TAG_FILTER_FACTORY(gteq, GTEQ)

#undef DEFINE_TAG_FILTER_FACTORY

ERRNO tsfile_tag_filter_create_checked(TsFileReader reader,
                                       const char* table_name,
                                       const char* column_name,
                                       const char* value, TagFilterOp op,
                                       TagFilterHandle* out_filter) {
    if (out_filter != nullptr) {
        *out_filter = nullptr;
    }
    if (reader == nullptr || table_name == nullptr || column_name == nullptr ||
        out_filter == nullptr ||
        (value == nullptr && op != TAG_FILTER_IS_NULL &&
         op != TAG_FILTER_IS_NOT_NULL)) {
        return common::E_INVALID_ARG;
    }
    try {
        auto* r = static_cast<storage::TsFileReader*>(reader);
        std::shared_ptr<storage::TableSchema> schema;
        const int ret = r->get_table_schema(table_name, schema);
        if (ret != common::E_OK) {
            return ret;
        }
        storage::TagFilterBuilder builder(schema.get());
        storage::Filter* filter = nullptr;
        switch (op) {
            case TAG_FILTER_EQ:
                filter = builder.eq(column_name, value);
                break;
            case TAG_FILTER_NEQ:
                filter = builder.neq(column_name, value);
                break;
            case TAG_FILTER_LT:
                filter = builder.lt(column_name, value);
                break;
            case TAG_FILTER_LTEQ:
                filter = builder.lteq(column_name, value);
                break;
            case TAG_FILTER_GT:
                filter = builder.gt(column_name, value);
                break;
            case TAG_FILTER_GTEQ:
                filter = builder.gteq(column_name, value);
                break;
            case TAG_FILTER_REGEXP:
                filter = builder.reg_exp(column_name, value);
                break;
            case TAG_FILTER_NOT_REGEXP:
                filter = builder.not_reg_exp(column_name, value);
                break;
            case TAG_FILTER_IS_NULL:
                filter = builder.is_null(column_name);
                break;
            case TAG_FILTER_IS_NOT_NULL:
                filter = builder.is_not_null(column_name);
                break;
            default:
                return common::E_INVALID_ARG;
        }
        if (filter == nullptr) {
            return common::E_COLUMN_NOT_EXIST;
        }
        *out_filter = static_cast<void*>(filter);
        return common::E_OK;
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    } catch (...) {
        return common::E_FILE_READ_ERR;
    }
}

TagFilterHandle tsfile_tag_filter_create(TsFileReader reader,
                                         const char* table_name,
                                         const char* column_name,
                                         const char* value, TagFilterOp op,
                                         ERRNO* err_code) {
    if (err_code == nullptr) {
        return nullptr;
    }
    TagFilterHandle filter = nullptr;
    *err_code = tsfile_tag_filter_create_checked(
        reader, table_name, column_name, value, op, &filter);
    if (*err_code == common::E_TABLE_NOT_EXIST) {
        *err_code = common::E_INVALID_ARG;
    }
    return *err_code == common::E_OK ? filter : nullptr;
}

ERRNO tsfile_tag_filter_between_checked(TsFileReader reader,
                                        const char* table_name,
                                        const char* column_name,
                                        const char* lower, const char* upper,
                                        bool is_not,
                                        TagFilterHandle* out_filter) {
    if (out_filter != nullptr) {
        *out_filter = nullptr;
    }
    if (reader == nullptr || table_name == nullptr || column_name == nullptr ||
        lower == nullptr || upper == nullptr || out_filter == nullptr) {
        return common::E_INVALID_ARG;
    }
    try {
        auto* r = static_cast<storage::TsFileReader*>(reader);
        std::shared_ptr<storage::TableSchema> schema;
        const int ret = r->get_table_schema(table_name, schema);
        if (ret != common::E_OK) {
            return ret;
        }
        storage::TagFilterBuilder builder(schema.get());
        storage::Filter* filter =
            is_not ? builder.not_between_and(column_name, lower, upper)
                   : builder.between_and(column_name, lower, upper);
        if (filter == nullptr) {
            return common::E_COLUMN_NOT_EXIST;
        }
        *out_filter = static_cast<void*>(filter);
        return common::E_OK;
    } catch (const std::bad_alloc&) {
        return common::E_OOM;
    } catch (...) {
        return common::E_FILE_READ_ERR;
    }
}

TagFilterHandle tsfile_tag_filter_between(TsFileReader reader,
                                          const char* table_name,
                                          const char* column_name,
                                          const char* lower, const char* upper,
                                          bool is_not, ERRNO* err_code) {
    if (err_code == nullptr) {
        return nullptr;
    }
    TagFilterHandle filter = nullptr;
    *err_code = tsfile_tag_filter_between_checked(
        reader, table_name, column_name, lower, upper, is_not, &filter);
    if (*err_code == common::E_TABLE_NOT_EXIST) {
        *err_code = common::E_INVALID_ARG;
    }
    return *err_code == common::E_OK ? filter : nullptr;
}

TagFilterHandle tsfile_tag_filter_and(TagFilterHandle left,
                                      TagFilterHandle right) {
    if (!left || !right) return nullptr;
    try {
        return storage::TagFilterBuilder::and_filter(
            static_cast<storage::Filter*>(left),
            static_cast<storage::Filter*>(right));
    } catch (...) {
        return nullptr;
    }
}

TagFilterHandle tsfile_tag_filter_or(TagFilterHandle left,
                                     TagFilterHandle right) {
    if (!left || !right) return nullptr;
    try {
        return storage::TagFilterBuilder::or_filter(
            static_cast<storage::Filter*>(left),
            static_cast<storage::Filter*>(right));
    } catch (...) {
        return nullptr;
    }
}

TagFilterHandle tsfile_tag_filter_not(TagFilterHandle filter) {
    if (!filter) return nullptr;
    try {
        return storage::TagFilterBuilder::not_filter(
            static_cast<storage::Filter*>(filter));
    } catch (...) {
        return nullptr;
    }
}

void tsfile_tag_filter_free(TagFilterHandle filter) {
    if (filter) {
        delete static_cast<storage::Filter*>(filter);
    }
}

ResultSet tsfile_query_table_with_tag_filter(
    TsFileReader reader, const char* table_name, char** columns,
    uint32_t column_num, Timestamp start_time, Timestamp end_time,
    TagFilterHandle tag_filter, int batch_size, ERRNO* err_code) {
    auto* r = static_cast<storage::TsFileReader*>(reader);
    storage::ResultSet* table_result_set = nullptr;
    std::vector<std::string> column_names;
    for (uint32_t i = 0; i < column_num; i++) {
        column_names.emplace_back(columns[i]);
    }
    *err_code = r->query(table_name, column_names, start_time, end_time,
                         table_result_set,
                         static_cast<storage::Filter*>(tag_filter), batch_size);
    return table_result_set;
}

#ifdef __cplusplus
}
#endif