/*
 * 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 a
 *
 *     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 <gtest/gtest.h>
#ifdef _WIN32
#include <io.h>
#else
#include <unistd.h>
#endif
#include <utils/db_utils.h>

#include <cstdio>
#include <cstring>
#include <string>

extern "C" {
#include "cwrapper/errno_define_c.h"
#include "cwrapper/tsfile_cwrapper.h"
}

#include "common/tablet.h"
#include "utils/errno_define.h"
namespace CReleaseTest {
class CReleaseTest : public testing::Test {};

TEST_F(CReleaseTest, TestCreateFile) {
    ERRNO error_no = RET_OK;
    remove("create_file1.tsfile");
    // Create File and Get RET_OK
    WriteFile file = write_file_new("create_file1.tsfile", &error_no);
    ASSERT_EQ(RET_OK, error_no);
    free_write_file(&file);

    // Already exists
    file = write_file_new("create_file1.tsfile", &error_no);
    ASSERT_EQ(RET_ALREADY_EXIST, error_no);
    ASSERT_EQ(nullptr, file);

    // Folder: rejected either as an open error (POSIX) or as already-existing
    // (Windows / filesystems where the directory already exists).
    file = write_file_new("test/", &error_no);
    ASSERT_TRUE(error_no == RET_FILE_OPEN_ERR || error_no == RET_ALREADY_EXIST);

    remove("create_file1.tsfile");
    free_write_file(&file);
}

TEST_F(CReleaseTest, RejectCorruptedFileWithoutDoubleClosingDescriptor) {
    const char* file_name = "corrupted_empty_file.tsfile";
    remove(file_name);
    FILE* empty_file = fopen(file_name, "wb");
    ASSERT_NE(nullptr, empty_file);
    ASSERT_EQ(0, fclose(empty_file));

    ERRNO error_no = RET_OK;
    TsFileReader reader = tsfile_reader_new(file_name, &error_no);
    EXPECT_EQ(nullptr, reader);
    EXPECT_EQ(RET_TSFILE_CORRUPTED, error_no);

    remove(file_name);
}

TEST_F(CReleaseTest, TsFileWriterNew) {
    ERRNO error_code = RET_OK;

    TableSchema test_schema;
    test_schema.table_name = strdup("test_table");
    test_schema.column_num = 0;

    // Invalid schema
    WriteFile file = write_file_new("test_empty_schema.tsfile", &error_code);
    ASSERT_EQ(RET_OK, error_code);
    TsFileWriter writer = tsfile_writer_new(file, &test_schema, &error_code);
    ASSERT_EQ(RET_INVALID_SCHEMA, error_code);
    ASSERT_EQ(nullptr, writer);
    ASSERT_EQ(RET_OK, tsfile_writer_close(writer));
    free_write_file(&file);
    ASSERT_EQ(nullptr, file);
    remove("test_empty_schema.tsfile");

    // Invalid schema with memory threshold
    file = write_file_new("test_empty_schema_memory_threshold.tsfile",
                          &error_code);
    ASSERT_EQ(RET_OK, error_code);
    // Invalid schema
    writer = tsfile_writer_new_with_memory_threshold(file, &test_schema, 100,
                                                     &error_code);
    ASSERT_EQ(RET_INVALID_SCHEMA, error_code);
    ASSERT_EQ(nullptr, writer);
    ASSERT_EQ(RET_OK, tsfile_writer_close(writer));
    free_write_file(&file);
    ASSERT_EQ(nullptr, file);
    remove("test_empty_schema_memory_threshold.tsfile");

    // Normal schema
    file = write_file_new("test_empty_writer.tsfile", &error_code);
    ASSERT_EQ(RET_OK, error_code);

    TableSchema table_schema;
    table_schema.table_name = strdup("test_table");
    table_schema.column_num = 2;
    table_schema.column_schemas =
        static_cast<ColumnSchema*>(malloc(sizeof(ColumnSchema) * 2));
    table_schema.column_schemas[0] =
        ColumnSchema{strdup("col1"), TS_DATATYPE_STRING, TAG};
    table_schema.column_schemas[1] =
        ColumnSchema{strdup("col2"), TS_DATATYPE_INT32, FIELD};

    writer = tsfile_writer_new(file, &table_schema, &error_code);
    ASSERT_EQ(RET_OK, error_code);
    error_code = tsfile_writer_close(writer);
    ASSERT_EQ(RET_OK, error_code);
    free_write_file(&file);
    remove("test_empty_writer.tsfile");

    // Normal schema with memory threshold
    file = write_file_new("test_memory_threshold_writer.tsfile", &error_code);
    ASSERT_EQ(RET_OK, error_code);
    writer = tsfile_writer_new_with_memory_threshold(file, &table_schema, 100,
                                                     &error_code);
    ASSERT_NE(nullptr, writer);
    ASSERT_EQ(RET_OK, error_code);
    ASSERT_EQ(RET_OK, tsfile_writer_close(writer));
    free_write_file(&file);
    remove("test_memory_threshold_writer.tsfile");

    free_table_schema(table_schema);
    free_table_schema(test_schema);
}

TEST_F(CReleaseTest, TsFileWriterWriteDataAbnormalColumn) {
    remove("TsFileWriterWriteDataAbnormalColumn.tsfile");
    ERRNO error_code = RET_OK;
    WriteFile file = write_file_new(
        "TsFileWriterWriteDataAbnormalColumn.tsfile", &error_code);

    TableSchema abnormal_schema;
    abnormal_schema.table_name = strdup("!@#$%^*()_+-=");
    abnormal_schema.column_num = 3;
    abnormal_schema.column_schemas =
        static_cast<ColumnSchema*>(malloc(sizeof(ColumnSchema) * 4));
    abnormal_schema.column_schemas[0] =
        ColumnSchema{strdup("!@#$%^*()_+-="), TS_DATATYPE_STRING, TAG};

    // TAG's datatype is not correct
    abnormal_schema.column_schemas[1] =
        ColumnSchema{strdup("TAG2"), TS_DATATYPE_INT32, TAG};

    // same column name with column[0]
    abnormal_schema.column_schemas[2] =
        ColumnSchema{strdup("!@#$%^*()_+-="), TS_DATATYPE_DOUBLE, FIELD};

    // column name conflict
    TsFileWriter writer =
        tsfile_writer_new(file, &abnormal_schema, &error_code);
    ASSERT_EQ(RET_INVALID_SCHEMA, error_code);
    writer = tsfile_writer_new_with_memory_threshold(file, &abnormal_schema,
                                                     100, &error_code);
    ASSERT_EQ(nullptr, writer);
    ASSERT_EQ(RET_INVALID_SCHEMA, error_code);
    free(abnormal_schema.column_schemas[2].column_name);

    abnormal_schema.column_schemas[2] =
        ColumnSchema{strdup("!@#$%^*()_+-=1"), TS_DATATYPE_DOUBLE, FIELD};

    // datatype conflict
    writer = tsfile_writer_new(file, &abnormal_schema, &error_code);
    ASSERT_EQ(RET_INVALID_SCHEMA, error_code);
    writer = tsfile_writer_new_with_memory_threshold(file, &abnormal_schema,
                                                     100, &error_code);
    ASSERT_EQ(nullptr, writer);
    ASSERT_EQ(RET_INVALID_SCHEMA, error_code);

    free(abnormal_schema.column_schemas[1].column_name);
    abnormal_schema.column_schemas[1] =
        ColumnSchema{strdup("TAG2"), TS_DATATYPE_STRING, TAG};

    writer = tsfile_writer_new(file, &abnormal_schema, &error_code);
    ASSERT_EQ(RET_OK, error_code);

    char** column_list = static_cast<char**>(malloc(sizeof(char*) * 3));
    column_list[0] = strdup("!@#$%^*()_+-=");
    column_list[1] = strdup("TAG2");
    column_list[2] = strdup("!@#$%^*()_+-=1");
    TSDataType* type_list =
        static_cast<TSDataType*>(malloc(sizeof(TSDataType) * 3));
    type_list[0] = TS_DATATYPE_STRING;
    type_list[1] = TS_DATATYPE_STRING;
    type_list[2] = TS_DATATYPE_DOUBLE;
    Tablet tablet = tablet_new(column_list, type_list, 3, 100);
    for (int i = 0; i < 100; i++) {
        tablet_add_timestamp(tablet, i, static_cast<int64_t>(i));
        tablet_add_value_by_name_string_with_len(
            tablet, i, "!@#$%^*()_+-=", "device1", strlen("device1"));
        tablet_add_value_by_index_string_with_len(
            tablet, i, 1, std::string("sensor" + std::to_string(i)).c_str(),
            std::string("sensor" + std::to_string(i)).length());
        tablet_add_value_by_name_double(tablet, i, "!@#$%^*()_+-=1", i * 100.0);
    }
    ASSERT_EQ(RET_OK, tsfile_writer_write(writer, tablet));
    ASSERT_EQ(RET_OK, tsfile_writer_close(writer));
    free_write_file(&file);

    TsFileReader reader = tsfile_reader_new(
        "TsFileWriterWriteDataAbnormalColumn.tsfile", &error_code);
    ASSERT_EQ(RET_OK, error_code);
    int i = 0;
    ResultSet result_set = tsfile_query_table(
        reader, "!@#$%^*()_+-=", column_list, 3, 0, 100, &error_code);
    while (tsfile_result_set_next(result_set, &error_code) &&
           error_code == RET_OK) {
        Timestamp timestamp =
            tsfile_result_set_get_value_by_name_int64_t(result_set, "time");
        ASSERT_EQ(timestamp * 100.0, tsfile_result_set_get_value_by_name_double(
                                         result_set, "!@#$%^*()_+-=1"));
        char* value_str =
            tsfile_result_set_get_value_by_index_string(result_set, 2);
        ASSERT_EQ("device1", std::string(value_str));
        free(value_str);
        i++;
    }
    ASSERT_EQ(100, i);
    for (int i = 0; i < 3; i++) {
        free(column_list[i]);
    }
    free(column_list);
    free(type_list);
    free_write_file(&file);
    free_table_schema(abnormal_schema);
    free_tablet(&tablet);
    free_tsfile_result_set(&result_set);
    tsfile_reader_close(reader);
    remove("TsFileWriterWriteDataAbnormalColumn.tsfile");
}

TEST_F(CReleaseTest, TsFileWriterMultiDataType) {
    ERRNO error_code = RET_OK;
    remove("TsFileWriterMultiDataType.tsfile");
    WriteFile file =
        write_file_new("TsFileWriterMultiDataType.tsfile", &error_code);
    ASSERT_EQ(RET_OK, error_code);
    TableSchema all_type_schema;
    all_type_schema.table_name = strdup("All_Datatype");
    all_type_schema.column_num = 6;
    all_type_schema.column_schemas =
        static_cast<ColumnSchema*>(malloc(sizeof(ColumnSchema) * 6));
    all_type_schema.column_schemas[0] =
        ColumnSchema{strdup("TAG"), TS_DATATYPE_STRING, TAG};
    all_type_schema.column_schemas[1] =
        ColumnSchema{strdup("INT32"), TS_DATATYPE_INT32, FIELD};
    all_type_schema.column_schemas[2] =
        ColumnSchema{strdup("INT64"), TS_DATATYPE_INT64, FIELD};
    all_type_schema.column_schemas[3] =
        ColumnSchema{strdup("FLOAT"), TS_DATATYPE_FLOAT, FIELD};
    all_type_schema.column_schemas[4] =
        ColumnSchema{strdup("DOUBLE"), TS_DATATYPE_DOUBLE, FIELD};
    all_type_schema.column_schemas[5] =
        ColumnSchema{strdup("BOOLEAN"), TS_DATATYPE_BOOLEAN, FIELD};

    TsFileWriter writer =
        tsfile_writer_new(file, &all_type_schema, &error_code);
    ASSERT_EQ(RET_OK, error_code);

    free_table_schema(all_type_schema);
    char** column_list = static_cast<char**>(malloc(sizeof(char*) * 6));
    column_list[0] = strdup("TAG");
    column_list[1] = strdup("INT32");
    column_list[2] = strdup("INT64");
    column_list[3] = strdup("FLOAT");
    column_list[4] = strdup("DOUBLE");
    column_list[5] = strdup("BOOLEAN");
    TSDataType* type_list =
        static_cast<TSDataType*>(malloc(sizeof(TSDataType) * 6));
    type_list[0] = TS_DATATYPE_STRING;
    type_list[1] = TS_DATATYPE_INT32;
    type_list[2] = TS_DATATYPE_INT64;
    type_list[3] = TS_DATATYPE_FLOAT;
    type_list[4] = TS_DATATYPE_DOUBLE;
    type_list[5] = TS_DATATYPE_BOOLEAN;
    Tablet tablet = tablet_new(column_list, type_list, 6, 1000);
    for (int i = 0; i < 1000; i++) {
        // negative timestamp included
        tablet_add_timestamp(tablet, i, static_cast<int64_t>(i - 10));
        tablet_add_value_by_name_string_with_len(tablet, i, "TAG", "device1",
                                                 strlen("device1"));
        tablet_add_value_by_name_int32_t(tablet, i, "INT32", i);
        tablet_add_value_by_index_int64_t(tablet, i, 2, i * 100);
        tablet_add_value_by_index_float(tablet, i, 3, i * 100.0);
        if (i > 900) {
            continue;
        }
        // Null value
        tablet_add_value_by_index_double(tablet, i, 4, i * 100.0);
        tablet_add_value_by_index_bool(tablet, i, 5, i % 2 == 0);
    }
    ASSERT_EQ(RET_OK, tsfile_writer_write(writer, tablet));
    ASSERT_EQ(RET_OK, tsfile_writer_close(writer));
    free_write_file(&file);

    TsFileReader reader =
        tsfile_reader_new("TsFileWriterMultiDataType.tsfile", &error_code);
    ASSERT_EQ(RET_OK, error_code);
    ResultSet result_set = tsfile_query_table(
        reader, "all_datatype", column_list, 6, 0, 1000, &error_code);
    int row_num = 0;
    while (tsfile_result_set_next(result_set, &error_code) &&
           error_code == RET_OK) {
        Timestamp timestamp =
            tsfile_result_set_get_value_by_name_int64_t(result_set, "time");
        int64_t value = timestamp + 10;
        char* str_value =
            tsfile_result_set_get_value_by_name_string(result_set, "TAG");
        ASSERT_EQ("device1", std::string(str_value));
        free(str_value);
        ASSERT_EQ(value, tsfile_result_set_get_value_by_name_int32_t(result_set,
                                                                     "int32"));
        ASSERT_EQ(value * 100, tsfile_result_set_get_value_by_name_int64_t(
                                   result_set, "int64"));
        ASSERT_EQ(value * 100.0, tsfile_result_set_get_value_by_name_float(
                                     result_set, "FLOAT"));

        if (value <= 900) {
            ASSERT_EQ(value * 100.0, tsfile_result_set_get_value_by_name_double(
                                         result_set, "DOUBLE"));
            ASSERT_EQ(value % 2 == 0, tsfile_result_set_get_value_by_name_bool(
                                          result_set, "BOOLEAN"));
        } else {
            ASSERT_TRUE(
                tsfile_result_set_is_null_by_name(result_set, "DOUBLE"));
        }
        row_num++;
    }
    ASSERT_EQ(990, row_num);
    free_tsfile_result_set(&result_set);
    tsfile_reader_close(reader);
    for (int i = 0; i < 6; i++) {
        free(column_list[i]);
    }
    free_tablet(&tablet);
    free(column_list);
    free(type_list);
    remove("TsFileWriterMultiDataType.tsfile");
}

TEST_F(CReleaseTest, TsFileWriterConfTest) {
    ERRNO err_no = RET_OK;
    WriteFile file = write_file_new("plain_file.tsfile", &err_no);
    TableSchema plain_schema;
    plain_schema.table_name = strdup("plain_table");
    plain_schema.column_num = 2;

    plain_schema.column_schemas =
        (ColumnSchema*)malloc(sizeof(ColumnSchema) * 2);
    plain_schema.column_schemas[0].column_name = strdup("id");
    plain_schema.column_schemas[0].data_type = TS_DATATYPE_STRING;
    plain_schema.column_schemas[0].column_category = ColumnCategory::TAG;
    plain_schema.column_schemas[1].column_name = strdup("value");
    plain_schema.column_schemas[1].data_type = TS_DATATYPE_INT64;
    plain_schema.column_schemas[1].column_category = ColumnCategory::FIELD;

    set_global_compression(TS_COMPRESSION_UNCOMPRESSED);
    set_datatype_encoding(TS_DATATYPE_INT64, TS_ENCODING_PLAIN);
    TsFileWriter writer = tsfile_writer_new(file, &plain_schema, &err_no);
    free_table_schema(plain_schema);

    char** column_list = static_cast<char**>(malloc(sizeof(char*) * 2));
    column_list[0] = strdup("id");
    column_list[1] = strdup("value");

    TSDataType* type_list =
        static_cast<TSDataType*>(malloc(sizeof(TSDataType) * 2));
    type_list[0] = TS_DATATYPE_STRING;
    type_list[1] = TS_DATATYPE_INT64;

    Tablet tablet = tablet_new(column_list, type_list, 2, 10);
    for (int i = 0; i < 10; i++) {
        tablet_add_timestamp(tablet, i, static_cast<int64_t>(i));
        tablet_add_value_by_name_string_with_len(tablet, i, "id", "device1",
                                                 strlen("device1"));
        tablet_add_value_by_name_int32_t(tablet, i, "value", i);
    }

    ASSERT_EQ(RET_OK, tsfile_writer_write(writer, tablet));
    ASSERT_EQ(RET_OK, tsfile_writer_close(writer));
    free_write_file(&file);
    TsFileReader reader = tsfile_reader_new("plain_file.tsfile", &err_no);
    ASSERT_EQ(RET_OK, err_no);
    TableSchema schema = tsfile_reader_get_table_schema(reader, "plain_table");
    ASSERT_EQ(schema.column_num, 2);
    uint32_t size = 0;
    DeviceSchema* device_schema =
        tsfile_reader_get_all_timeseries_schemas(reader, &size);
    ASSERT_EQ(1, size);
    ASSERT_EQ(1, device_schema->timeseries_num);
    ASSERT_EQ(device_schema->timeseries_schema[0].encoding, TS_ENCODING_PLAIN);
    ASSERT_EQ(device_schema->timeseries_schema[0].compression,
              TS_COMPRESSION_UNCOMPRESSED);
    tsfile_reader_close(reader);
    free_table_schema(schema);
    free_device_schema(*device_schema);
    free(device_schema);
    free(column_list[0]);
    free(column_list[1]);
    free(column_list);
    free(type_list);
    free_tablet(&tablet);

    remove("plain_file.tsfile");
}

}  // namespace CReleaseTest