jt2594838 commented on code in PR #387: URL: https://github.com/apache/tsfile/pull/387#discussion_r1929448663
########## cpp/src/cwrapper/tsfile_cwrapper.h: ########## @@ -0,0 +1,243 @@ +/* + * 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. + */ + +#ifndef SRC_CWRAPPER_TSFILE_CWRAPPER_H_ +#define SRC_CWRAPPER_TSFILE_CWRAPPER_H_ +#include <iostream> + +typedef enum { + TS_DATATYPE_BOOLEAN = 0, + TS_DATATYPE_INT32 = 1, + TS_DATATYPE_INT64 = 2, + TS_DATATYPE_FLOAT = 3, + TS_DATATYPE_DOUBLE = 4, + TS_DATATYPE_TEXT = 5, + TS_DATATYPE_VECTOR = 6, + TS_DATATYPE_NULL_TYPE = 254, + TS_DATATYPE_INVALID = 255 +} TSDataType; + +typedef enum { + TS_ENCODING_PLAIN = 0, + TS_ENCODING_DICTIONARY = 1, + TS_ENCODING_RLE = 2, + TS_ENCODING_DIFF = 3, + TS_ENCODING_TS_2DIFF = 4, + TS_ENCODING_BITMAP = 5, + TS_ENCODING_GORILLA_V1 = 6, + TS_ENCODING_REGULAR = 7, + TS_ENCODING_GORILLA = 8, + TS_ENCODING_ZIGZAG = 9, + TS_ENCODING_FREQ = 10, + TS_ENCODING_INVALID = 255 +} TSEncoding; + +typedef enum { + TS_COMPRESSION_UNCOMPRESSED = 0, + TS_COMPRESSION_SNAPPY = 1, + TS_COMPRESSION_GZIP = 2, + TS_COMPRESSION_LZO = 3, + TS_COMPRESSION_SDT = 4, + TS_COMPRESSION_PAA = 5, + TS_COMPRESSION_PLA = 6, + TS_COMPRESSION_LZ4 = 7, + TS_COMPRESSION_INVALID = 255 +} CompressionType; + +typedef enum column_category { TAG = 0, FIELD = 1 } ColumnCategory; + +typedef struct column_schema { + char* column_name; + TSDataType data_type; + ColumnCategory column_category; +} ColumnSchema; + +typedef struct table_schema { + char* table_name; + ColumnSchema* column_schemas; + int column_num; +} TableSchema; + +typedef struct timeseries_schema { + char* timeseries_name; + TSDataType data_type; + TSEncoding encoding; + CompressionType compression; +} TimeseriesSchema; + +typedef struct device_schema { + char* device_name; + TimeseriesSchema* timeseries_schema; + int timeseries_num; +} DeviceSchema; + +typedef struct { + char** column_names; + TSDataType* data_types; + int column_num; +} ResultSetMetaData; + +typedef struct tsfile_conf { + int mem_threshold_kb; +} TsFileConf; + +typedef void* TsFileReader; +typedef void* TsFileWriter; + +// just resue Tablet from c++ +typedef void* Tablet; +typedef void* TsRecord; + +typedef void* ResultSet; + +typedef int32_t ERRNO; +typedef int64_t timestamp; + +#ifdef __cplusplus +extern "C" { +#endif + +/*--------------------------Tablet API------------------------ */ +Tablet tablet_new_with_device(const char* device_id, char** column_name_list, + TSDataType* data_types, int column_num, + int max_rows); + +Tablet tablet_new(const char** column_name_list, TSDataType* data_types, + int column_num); + +uint32_t tablet_get_cur_row_size(Tablet tablet); + +ERRNO tablet_add_timestamp(Tablet tablet, uint32_t row_index, + timestamp timestamp); + +#define tablet_add_value_by_name_dec(type) \ + ERRNO tablet_add_value_by_name_##type(Tablet* tablet, uint32_t row_index, \ + char* column_name, type value); + +tablet_add_value_by_name_dec(int32_t); +tablet_add_value_by_name_dec(int64_t); +tablet_add_value_by_name_dec(float); +tablet_add_value_by_name_dec(double); +tablet_add_value_by_name_dec(bool); + +#define table_add_value_by_index_dec(type) \ + ERRNO tablet_add_value_by_index_##type(Tablet tablet, uint32_t row_index, \ + uint32_t column_index, type value); + +table_add_value_by_index_dec(int32_t); +table_add_value_by_index_dec(int64_t); +table_add_value_by_index_dec(float); +table_add_value_by_index_dec(double); +table_add_value_by_index_dec(bool); + +void* tablet_get_value(Tablet tablet, uint32_t row_index, uint32_t schema_index, + TSDataType& type); + +/*--------------------------TsRecord API------------------------ */ +TsRecord ts_record_new(const char* device_name, timestamp timestamp, + int timeseries_num); + +#define insert_data_into_ts_record_by_name_(type) \ + ERRNO insert_data_into_ts_record_by_name##type( \ + TsRecord data, char* measurement_name, type value); + +insert_data_into_ts_record_by_name_(int32_t); +insert_data_into_ts_record_by_name_(int64_t); +insert_data_into_ts_record_by_name_(bool); +insert_data_into_ts_record_by_name_(float); +insert_data_into_ts_record_by_name_(double); + +/*--------------------------TsFile Reader and Writer------------------------ */ +TsFileReader tsfile_reader_new(const char* pathname, ERRNO* err_code); +TsFileWriter tsfile_writer_new(const char* pathname, ERRNO* err_code); +TsFileWriter tsfile_writer_open_conf(const char* pathname, mode_t flag, + ERRNO* err_code, TsFileConf* conf); + +ERRNO tsfile_writer_close(TsFileWriter writer); +ERRNO tsfile_reader_close(TsFileReader reader); + +/*--------------------------TsFile Writer Register------------------------ */ +void tsfile_writer_register_table(TsFileWriter writer, TableSchema* schema); +ERRNO tsfile_writer_register_timeseries(TsFileWriter writer, + const char* device_name, Review Comment: device_name -> device_id ########## cpp/src/cwrapper/tsfile_cwrapper.h: ########## @@ -0,0 +1,243 @@ +/* + * 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. + */ + +#ifndef SRC_CWRAPPER_TSFILE_CWRAPPER_H_ +#define SRC_CWRAPPER_TSFILE_CWRAPPER_H_ +#include <iostream> + +typedef enum { + TS_DATATYPE_BOOLEAN = 0, + TS_DATATYPE_INT32 = 1, + TS_DATATYPE_INT64 = 2, + TS_DATATYPE_FLOAT = 3, + TS_DATATYPE_DOUBLE = 4, + TS_DATATYPE_TEXT = 5, + TS_DATATYPE_VECTOR = 6, + TS_DATATYPE_NULL_TYPE = 254, + TS_DATATYPE_INVALID = 255 +} TSDataType; + +typedef enum { + TS_ENCODING_PLAIN = 0, + TS_ENCODING_DICTIONARY = 1, + TS_ENCODING_RLE = 2, + TS_ENCODING_DIFF = 3, + TS_ENCODING_TS_2DIFF = 4, + TS_ENCODING_BITMAP = 5, + TS_ENCODING_GORILLA_V1 = 6, + TS_ENCODING_REGULAR = 7, + TS_ENCODING_GORILLA = 8, + TS_ENCODING_ZIGZAG = 9, + TS_ENCODING_FREQ = 10, + TS_ENCODING_INVALID = 255 +} TSEncoding; + +typedef enum { + TS_COMPRESSION_UNCOMPRESSED = 0, + TS_COMPRESSION_SNAPPY = 1, + TS_COMPRESSION_GZIP = 2, + TS_COMPRESSION_LZO = 3, + TS_COMPRESSION_SDT = 4, + TS_COMPRESSION_PAA = 5, + TS_COMPRESSION_PLA = 6, + TS_COMPRESSION_LZ4 = 7, + TS_COMPRESSION_INVALID = 255 +} CompressionType; + +typedef enum column_category { TAG = 0, FIELD = 1 } ColumnCategory; + +typedef struct column_schema { + char* column_name; + TSDataType data_type; + ColumnCategory column_category; +} ColumnSchema; + +typedef struct table_schema { + char* table_name; + ColumnSchema* column_schemas; + int column_num; +} TableSchema; + +typedef struct timeseries_schema { + char* timeseries_name; + TSDataType data_type; + TSEncoding encoding; + CompressionType compression; +} TimeseriesSchema; + +typedef struct device_schema { + char* device_name; + TimeseriesSchema* timeseries_schema; + int timeseries_num; +} DeviceSchema; + +typedef struct { + char** column_names; + TSDataType* data_types; + int column_num; +} ResultSetMetaData; + +typedef struct tsfile_conf { + int mem_threshold_kb; +} TsFileConf; + +typedef void* TsFileReader; +typedef void* TsFileWriter; + +// just resue Tablet from c++ +typedef void* Tablet; +typedef void* TsRecord; + +typedef void* ResultSet; + +typedef int32_t ERRNO; +typedef int64_t timestamp; + +#ifdef __cplusplus +extern "C" { +#endif + +/*--------------------------Tablet API------------------------ */ +Tablet tablet_new_with_device(const char* device_id, char** column_name_list, + TSDataType* data_types, int column_num, + int max_rows); + +Tablet tablet_new(const char** column_name_list, TSDataType* data_types, + int column_num); + +uint32_t tablet_get_cur_row_size(Tablet tablet); + +ERRNO tablet_add_timestamp(Tablet tablet, uint32_t row_index, + timestamp timestamp); + +#define tablet_add_value_by_name_dec(type) \ + ERRNO tablet_add_value_by_name_##type(Tablet* tablet, uint32_t row_index, \ + char* column_name, type value); + +tablet_add_value_by_name_dec(int32_t); +tablet_add_value_by_name_dec(int64_t); +tablet_add_value_by_name_dec(float); +tablet_add_value_by_name_dec(double); +tablet_add_value_by_name_dec(bool); + +#define table_add_value_by_index_dec(type) \ + ERRNO tablet_add_value_by_index_##type(Tablet tablet, uint32_t row_index, \ + uint32_t column_index, type value); + +table_add_value_by_index_dec(int32_t); +table_add_value_by_index_dec(int64_t); +table_add_value_by_index_dec(float); +table_add_value_by_index_dec(double); +table_add_value_by_index_dec(bool); + +void* tablet_get_value(Tablet tablet, uint32_t row_index, uint32_t schema_index, + TSDataType& type); + +/*--------------------------TsRecord API------------------------ */ +TsRecord ts_record_new(const char* device_name, timestamp timestamp, + int timeseries_num); + +#define insert_data_into_ts_record_by_name_(type) \ + ERRNO insert_data_into_ts_record_by_name##type( \ + TsRecord data, char* measurement_name, type value); + +insert_data_into_ts_record_by_name_(int32_t); +insert_data_into_ts_record_by_name_(int64_t); +insert_data_into_ts_record_by_name_(bool); +insert_data_into_ts_record_by_name_(float); +insert_data_into_ts_record_by_name_(double); + +/*--------------------------TsFile Reader and Writer------------------------ */ +TsFileReader tsfile_reader_new(const char* pathname, ERRNO* err_code); +TsFileWriter tsfile_writer_new(const char* pathname, ERRNO* err_code); +TsFileWriter tsfile_writer_open_conf(const char* pathname, mode_t flag, Review Comment: tsfile_writer_new_with_conf? ########## cpp/src/cwrapper/tsfile_cwrapper_expression.h: ########## @@ -0,0 +1,103 @@ +/* + * 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. + */ + +#ifndef CWRAPPER_TSFILE_CWRAPPER_H +#define CWRAPPER_TSFILE_CWRAPPER_H + +#include <fcntl.h> +#include <stdbool.h> +#include <stddef.h> +#include <stdint.h> +#ifdef _WIN32 +#include <sys/stat.h> +#endif + +#include "tsfile_cwrapper.h" + +typedef void* TimeFilterExpression; + +#define MAX_COLUMN_FILTER_NUM 10 +typedef enum operator_type { + LT, + LE, + EQ, + GT, + GE, + NOTEQ, +} OperatorType; + +typedef enum expression_type { + OR, + AND, + GLOBALTIME, +} ExpressionType; + +typedef struct constant { + int64_t value_condition; + int type; +} Constant; + +typedef struct expression { + const char* column_name; + Constant const_condition; + ExpressionType expression_type; + OperatorType operate_type; + struct expression* children[MAX_COLUMN_FILTER_NUM]; + int children_length; +} Expression; + +typedef Tablet DataResult; Review Comment: What is this for? ########## cpp/src/cwrapper/tsfile_cwrapper.h: ########## @@ -0,0 +1,243 @@ +/* + * 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. + */ + +#ifndef SRC_CWRAPPER_TSFILE_CWRAPPER_H_ +#define SRC_CWRAPPER_TSFILE_CWRAPPER_H_ +#include <iostream> + +typedef enum { + TS_DATATYPE_BOOLEAN = 0, + TS_DATATYPE_INT32 = 1, + TS_DATATYPE_INT64 = 2, + TS_DATATYPE_FLOAT = 3, + TS_DATATYPE_DOUBLE = 4, + TS_DATATYPE_TEXT = 5, + TS_DATATYPE_VECTOR = 6, + TS_DATATYPE_NULL_TYPE = 254, + TS_DATATYPE_INVALID = 255 +} TSDataType; + +typedef enum { + TS_ENCODING_PLAIN = 0, + TS_ENCODING_DICTIONARY = 1, + TS_ENCODING_RLE = 2, + TS_ENCODING_DIFF = 3, + TS_ENCODING_TS_2DIFF = 4, + TS_ENCODING_BITMAP = 5, + TS_ENCODING_GORILLA_V1 = 6, + TS_ENCODING_REGULAR = 7, + TS_ENCODING_GORILLA = 8, + TS_ENCODING_ZIGZAG = 9, + TS_ENCODING_FREQ = 10, + TS_ENCODING_INVALID = 255 +} TSEncoding; + +typedef enum { + TS_COMPRESSION_UNCOMPRESSED = 0, + TS_COMPRESSION_SNAPPY = 1, + TS_COMPRESSION_GZIP = 2, + TS_COMPRESSION_LZO = 3, + TS_COMPRESSION_SDT = 4, + TS_COMPRESSION_PAA = 5, + TS_COMPRESSION_PLA = 6, + TS_COMPRESSION_LZ4 = 7, + TS_COMPRESSION_INVALID = 255 +} CompressionType; + +typedef enum column_category { TAG = 0, FIELD = 1 } ColumnCategory; + +typedef struct column_schema { + char* column_name; + TSDataType data_type; + ColumnCategory column_category; +} ColumnSchema; + +typedef struct table_schema { + char* table_name; + ColumnSchema* column_schemas; + int column_num; +} TableSchema; + +typedef struct timeseries_schema { + char* timeseries_name; + TSDataType data_type; + TSEncoding encoding; + CompressionType compression; +} TimeseriesSchema; + +typedef struct device_schema { + char* device_name; + TimeseriesSchema* timeseries_schema; + int timeseries_num; +} DeviceSchema; + +typedef struct { + char** column_names; + TSDataType* data_types; + int column_num; +} ResultSetMetaData; + +typedef struct tsfile_conf { + int mem_threshold_kb; +} TsFileConf; + +typedef void* TsFileReader; +typedef void* TsFileWriter; + +// just resue Tablet from c++ +typedef void* Tablet; +typedef void* TsRecord; + +typedef void* ResultSet; + +typedef int32_t ERRNO; +typedef int64_t timestamp; + +#ifdef __cplusplus +extern "C" { +#endif + +/*--------------------------Tablet API------------------------ */ +Tablet tablet_new_with_device(const char* device_id, char** column_name_list, + TSDataType* data_types, int column_num, + int max_rows); + +Tablet tablet_new(const char** column_name_list, TSDataType* data_types, + int column_num); + +uint32_t tablet_get_cur_row_size(Tablet tablet); + +ERRNO tablet_add_timestamp(Tablet tablet, uint32_t row_index, + timestamp timestamp); + +#define tablet_add_value_by_name_dec(type) \ + ERRNO tablet_add_value_by_name_##type(Tablet* tablet, uint32_t row_index, \ + char* column_name, type value); + +tablet_add_value_by_name_dec(int32_t); +tablet_add_value_by_name_dec(int64_t); +tablet_add_value_by_name_dec(float); +tablet_add_value_by_name_dec(double); +tablet_add_value_by_name_dec(bool); + +#define table_add_value_by_index_dec(type) \ + ERRNO tablet_add_value_by_index_##type(Tablet tablet, uint32_t row_index, \ + uint32_t column_index, type value); + +table_add_value_by_index_dec(int32_t); +table_add_value_by_index_dec(int64_t); +table_add_value_by_index_dec(float); +table_add_value_by_index_dec(double); +table_add_value_by_index_dec(bool); + +void* tablet_get_value(Tablet tablet, uint32_t row_index, uint32_t schema_index, + TSDataType& type); + +/*--------------------------TsRecord API------------------------ */ +TsRecord ts_record_new(const char* device_name, timestamp timestamp, + int timeseries_num); + +#define insert_data_into_ts_record_by_name_(type) \ + ERRNO insert_data_into_ts_record_by_name##type( \ + TsRecord data, char* measurement_name, type value); + +insert_data_into_ts_record_by_name_(int32_t); +insert_data_into_ts_record_by_name_(int64_t); +insert_data_into_ts_record_by_name_(bool); +insert_data_into_ts_record_by_name_(float); +insert_data_into_ts_record_by_name_(double); + +/*--------------------------TsFile Reader and Writer------------------------ */ +TsFileReader tsfile_reader_new(const char* pathname, ERRNO* err_code); +TsFileWriter tsfile_writer_new(const char* pathname, ERRNO* err_code); +TsFileWriter tsfile_writer_open_conf(const char* pathname, mode_t flag, + ERRNO* err_code, TsFileConf* conf); + +ERRNO tsfile_writer_close(TsFileWriter writer); +ERRNO tsfile_reader_close(TsFileReader reader); + +/*--------------------------TsFile Writer Register------------------------ */ +void tsfile_writer_register_table(TsFileWriter writer, TableSchema* schema); +ERRNO tsfile_writer_register_timeseries(TsFileWriter writer, + const char* device_name, + TimeseriesSchema* schema); +ERRNO tsfile_writer_register_device(TsFileWriter writer, + const DeviceSchema* device_schema); + +/*-------------------TsFile Writer write and flush data------------------ */ +ERRNO tsfile_writer_write_tablet(TsFileWriter writer, Tablet tablet); +ERRNO tsfile_writer_write_ts_record(TsFileWriter writer, TsRecord record); +ERRNO tsfile_writer_flush_data(TsFileWriter writer); + +/*-------------------TsFile reader query data------------------ */ +ResultSet tsfile_reader_query_table(TsFileReader reader, char* table_name, + char** columns, uint32_t column_num, + timestamp start_time, timestamp end_time); +ResultSet tsfile_reader_query_path(TsFileReader reader, char** path_list, + uint32_t path_num, timestamp start_time, + timestamp end_time); +bool tsfile_result_set_has_next(ResultSet result_set); + +#define tsfile_result_set_get_value_by_name_(type) \ + type tsfile_result_set_get_value_by_name_##type(ResultSet result_set, \ + const char* column_name) +tsfile_result_set_get_value_by_name_(bool); +tsfile_result_set_get_value_by_name_(int32_t); +tsfile_result_set_get_value_by_name_(int64_t); +tsfile_result_set_get_value_by_name_(float); +tsfile_result_set_get_value_by_name_(double); + +#define tsfile_result_set_get_value_by_index_(type) \ + type tsfile_result_set_get_value_by_index_##type(ResultSet result_set, \ + uint32_t column_index); + +tsfile_result_set_get_value_by_index_(int32_t); +tsfile_result_set_get_value_by_index_(int64_t); +tsfile_result_set_get_value_by_index_(float); +tsfile_result_set_get_value_by_index_(double); +tsfile_result_set_get_value_by_index_(bool); + +bool tsfile_result_set_is_null_by_name(ResultSet result_set, + char* column_name); + +bool tsfile_result_set_is_null_by_index(ResultSet result_set, + uint32_t column_index); + +ResultSetMetaData tsfile_result_set_get_metadata(ResultSet result_set); + + +// Desc Table Schema + +TableSchema tsfile_reader_get_table_schema(TsFileReader reader, + const char* table_name); +DeviceSchema tsfile_reader_get_timeseries_schema(TsFileReader reader, const char* device_id); + Review Comment: TableSchema* tsfile_reader_get_all_table_schemas(TsFileReader reader, uint32_t* size) ########## cpp/src/cwrapper/tsfile_cwrapper.h: ########## @@ -0,0 +1,243 @@ +/* + * 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. + */ + +#ifndef SRC_CWRAPPER_TSFILE_CWRAPPER_H_ +#define SRC_CWRAPPER_TSFILE_CWRAPPER_H_ +#include <iostream> + +typedef enum { + TS_DATATYPE_BOOLEAN = 0, + TS_DATATYPE_INT32 = 1, + TS_DATATYPE_INT64 = 2, + TS_DATATYPE_FLOAT = 3, + TS_DATATYPE_DOUBLE = 4, + TS_DATATYPE_TEXT = 5, + TS_DATATYPE_VECTOR = 6, + TS_DATATYPE_NULL_TYPE = 254, + TS_DATATYPE_INVALID = 255 +} TSDataType; + +typedef enum { + TS_ENCODING_PLAIN = 0, + TS_ENCODING_DICTIONARY = 1, + TS_ENCODING_RLE = 2, + TS_ENCODING_DIFF = 3, + TS_ENCODING_TS_2DIFF = 4, + TS_ENCODING_BITMAP = 5, + TS_ENCODING_GORILLA_V1 = 6, + TS_ENCODING_REGULAR = 7, + TS_ENCODING_GORILLA = 8, + TS_ENCODING_ZIGZAG = 9, + TS_ENCODING_FREQ = 10, + TS_ENCODING_INVALID = 255 +} TSEncoding; + +typedef enum { + TS_COMPRESSION_UNCOMPRESSED = 0, + TS_COMPRESSION_SNAPPY = 1, + TS_COMPRESSION_GZIP = 2, + TS_COMPRESSION_LZO = 3, + TS_COMPRESSION_SDT = 4, + TS_COMPRESSION_PAA = 5, + TS_COMPRESSION_PLA = 6, + TS_COMPRESSION_LZ4 = 7, + TS_COMPRESSION_INVALID = 255 +} CompressionType; + +typedef enum column_category { TAG = 0, FIELD = 1 } ColumnCategory; + +typedef struct column_schema { + char* column_name; + TSDataType data_type; + ColumnCategory column_category; +} ColumnSchema; + +typedef struct table_schema { + char* table_name; + ColumnSchema* column_schemas; + int column_num; +} TableSchema; + +typedef struct timeseries_schema { + char* timeseries_name; + TSDataType data_type; + TSEncoding encoding; + CompressionType compression; +} TimeseriesSchema; + +typedef struct device_schema { + char* device_name; + TimeseriesSchema* timeseries_schema; + int timeseries_num; +} DeviceSchema; + +typedef struct { + char** column_names; + TSDataType* data_types; + int column_num; +} ResultSetMetaData; + +typedef struct tsfile_conf { + int mem_threshold_kb; +} TsFileConf; + +typedef void* TsFileReader; +typedef void* TsFileWriter; + +// just resue Tablet from c++ +typedef void* Tablet; +typedef void* TsRecord; + +typedef void* ResultSet; + +typedef int32_t ERRNO; +typedef int64_t timestamp; + +#ifdef __cplusplus +extern "C" { +#endif + +/*--------------------------Tablet API------------------------ */ +Tablet tablet_new_with_device(const char* device_id, char** column_name_list, + TSDataType* data_types, int column_num, + int max_rows); + +Tablet tablet_new(const char** column_name_list, TSDataType* data_types, + int column_num); + +uint32_t tablet_get_cur_row_size(Tablet tablet); + +ERRNO tablet_add_timestamp(Tablet tablet, uint32_t row_index, + timestamp timestamp); + +#define tablet_add_value_by_name_dec(type) \ + ERRNO tablet_add_value_by_name_##type(Tablet* tablet, uint32_t row_index, \ + char* column_name, type value); + +tablet_add_value_by_name_dec(int32_t); +tablet_add_value_by_name_dec(int64_t); +tablet_add_value_by_name_dec(float); +tablet_add_value_by_name_dec(double); +tablet_add_value_by_name_dec(bool); + +#define table_add_value_by_index_dec(type) \ + ERRNO tablet_add_value_by_index_##type(Tablet tablet, uint32_t row_index, \ + uint32_t column_index, type value); + +table_add_value_by_index_dec(int32_t); +table_add_value_by_index_dec(int64_t); +table_add_value_by_index_dec(float); +table_add_value_by_index_dec(double); +table_add_value_by_index_dec(bool); + +void* tablet_get_value(Tablet tablet, uint32_t row_index, uint32_t schema_index, + TSDataType& type); + +/*--------------------------TsRecord API------------------------ */ +TsRecord ts_record_new(const char* device_name, timestamp timestamp, + int timeseries_num); + +#define insert_data_into_ts_record_by_name_(type) \ + ERRNO insert_data_into_ts_record_by_name##type( \ + TsRecord data, char* measurement_name, type value); + +insert_data_into_ts_record_by_name_(int32_t); +insert_data_into_ts_record_by_name_(int64_t); +insert_data_into_ts_record_by_name_(bool); +insert_data_into_ts_record_by_name_(float); +insert_data_into_ts_record_by_name_(double); + +/*--------------------------TsFile Reader and Writer------------------------ */ +TsFileReader tsfile_reader_new(const char* pathname, ERRNO* err_code); +TsFileWriter tsfile_writer_new(const char* pathname, ERRNO* err_code); +TsFileWriter tsfile_writer_open_conf(const char* pathname, mode_t flag, + ERRNO* err_code, TsFileConf* conf); + +ERRNO tsfile_writer_close(TsFileWriter writer); +ERRNO tsfile_reader_close(TsFileReader reader); + +/*--------------------------TsFile Writer Register------------------------ */ +void tsfile_writer_register_table(TsFileWriter writer, TableSchema* schema); +ERRNO tsfile_writer_register_timeseries(TsFileWriter writer, + const char* device_name, + TimeseriesSchema* schema); +ERRNO tsfile_writer_register_device(TsFileWriter writer, + const DeviceSchema* device_schema); + +/*-------------------TsFile Writer write and flush data------------------ */ +ERRNO tsfile_writer_write_tablet(TsFileWriter writer, Tablet tablet); +ERRNO tsfile_writer_write_ts_record(TsFileWriter writer, TsRecord record); +ERRNO tsfile_writer_flush_data(TsFileWriter writer); + +/*-------------------TsFile reader query data------------------ */ +ResultSet tsfile_reader_query_table(TsFileReader reader, char* table_name, + char** columns, uint32_t column_num, + timestamp start_time, timestamp end_time); +ResultSet tsfile_reader_query_path(TsFileReader reader, char** path_list, + uint32_t path_num, timestamp start_time, + timestamp end_time); +bool tsfile_result_set_has_next(ResultSet result_set); + +#define tsfile_result_set_get_value_by_name_(type) \ + type tsfile_result_set_get_value_by_name_##type(ResultSet result_set, \ + const char* column_name) +tsfile_result_set_get_value_by_name_(bool); +tsfile_result_set_get_value_by_name_(int32_t); +tsfile_result_set_get_value_by_name_(int64_t); +tsfile_result_set_get_value_by_name_(float); +tsfile_result_set_get_value_by_name_(double); + +#define tsfile_result_set_get_value_by_index_(type) \ + type tsfile_result_set_get_value_by_index_##type(ResultSet result_set, \ + uint32_t column_index); + +tsfile_result_set_get_value_by_index_(int32_t); +tsfile_result_set_get_value_by_index_(int64_t); +tsfile_result_set_get_value_by_index_(float); +tsfile_result_set_get_value_by_index_(double); +tsfile_result_set_get_value_by_index_(bool); + +bool tsfile_result_set_is_null_by_name(ResultSet result_set, + char* column_name); + +bool tsfile_result_set_is_null_by_index(ResultSet result_set, + uint32_t column_index); + +ResultSetMetaData tsfile_result_set_get_metadata(ResultSet result_set); + + +// Desc Table Schema + +TableSchema tsfile_reader_get_table_schema(TsFileReader reader, + const char* table_name); +DeviceSchema tsfile_reader_get_timeseries_schema(TsFileReader reader, const char* device_id); + +// destroy pointer +ERRNO destroy_tsfile_ts_record(TsRecord record); +ERRNO destroy_tablet(Tablet tablet); +void destroy_tsfile_result_set(ResultSet result_set); Review Comment: The operation name on ResultSet is more often "close" rather than "destroy." ########## cpp/src/cwrapper/tsfile_cwrapper_expression.h: ########## @@ -0,0 +1,103 @@ +/* + * 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. + */ + +#ifndef CWRAPPER_TSFILE_CWRAPPER_H +#define CWRAPPER_TSFILE_CWRAPPER_H + +#include <fcntl.h> +#include <stdbool.h> +#include <stddef.h> +#include <stdint.h> +#ifdef _WIN32 +#include <sys/stat.h> +#endif + +#include "tsfile_cwrapper.h" + +typedef void* TimeFilterExpression; + +#define MAX_COLUMN_FILTER_NUM 10 +typedef enum operator_type { + LT, + LE, + EQ, + GT, + GE, + NOTEQ, +} OperatorType; + +typedef enum expression_type { + OR, + AND, + GLOBALTIME, +} ExpressionType; + +typedef struct constant { + int64_t value_condition; Review Comment: How about using a union? ########## cpp/src/cwrapper/tsfile_cwrapper.cc: ########## @@ -0,0 +1,478 @@ +/* + * 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 <reader/qds_without_timegenerator.h> + +#include "common/global.h" +#include "common/tablet.h" +#include "reader/result_set.h" +#include "reader/tsfile_reader.h" +#include "writer/tsfile_writer.h" + +static bool is_init = false; + +Tablet tablet_new_with_device(const char *device_id, char **column_name_list, + TSDataType *data_types, uint32_t 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))); + } + auto *tablet = + new storage::Tablet(device_id, &measurement_list, &data_type_list, max_rows); + tablet->init(); + return tablet; +} + +Tablet tablet_new(const char **column_name_list, TSDataType *data_types, + uint32_t column_num) { + 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))); + } + auto *tablet = new storage::Tablet("", &measurement_list, &data_type_list); + return tablet; +} + +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, \ + type value) { \ + return static_cast<storage::Tablet *>(tablet)->add_value( \ + row_index, 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); + +#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, \ + type value) { \ + return static_cast<storage::Tablet *>(tablet)->add_value( \ + row_index, column_index, value); \ + } + +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); + +void *tablet_get_value(Tablet tablet, uint32_t row_index, uint32_t schema_index, + TSDataType *type) { + common::TSDataType data_type; + void *result = static_cast<storage::Tablet *>(tablet)->get_value( + row_index, schema_index, data_type); + *type = static_cast<TSDataType>(data_type); + return result; +} + +// TsRecord API +TsRecord ts_record_new(const char *device_name, int64_t timestamp, + int timeseries_num) { + auto *record = + new storage::TsRecord(timestamp, device_name, 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; \ + } + +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); + +void init_tsfile_config() { + if (!is_init) { + common::init_config_value(); + is_init = true; + } +} + +TsFileReader tsfile_reader_new(const char *pathname, ERRNO *err_code) { + init_tsfile_config(); + auto reader = new storage::TsFileReader(); + int ret = reader->open(pathname); + if (ret != common::E_OK) { + std::cout << "open file failed" << std::endl; + *err_code = ret; + delete reader; + return nullptr; + } + return reader; +} + +TsFileWriter tsfile_writer_new(const char *pathname, 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); + if (ret != common::E_OK) { + delete writer; + *err_code = ret; + return nullptr; + } + return writer; +} + +TsFileWriter tsfile_writer_new_flag(const char *pathname, mode_t flag, + ERRNO *err_code) { + init_tsfile_config(); + auto *writer = new storage::TsFileWriter(); + 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_close(TsFileWriter writer) { + auto *w = static_cast<storage::TsFileWriter *>(writer); + int ret = w->close(); + delete w; + return ret; +} + +ERRNO tsfile_reader_close(TsFileReader reader) { + auto *ts_reader = (storage::TsFileReader *)reader; + delete ts_reader; + return common::E_OK; +} + +void tsfile_writer_register_table(TsFileWriter writer, TableSchema *schema) { + std::vector<storage::MeasurementSchema *> measurement_schemas; + std::vector<storage::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<storage::ColumnCategory>(cur_schema->column_category)); + } + auto tsfile_writer = static_cast<storage::TsFileWriter *>(writer); + 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_name, + TimeseriesSchema *schema) { + auto *w = (storage::TsFileWriter *)writer; + + int ret = w->register_timeseries( + device_name, storage::MeasurementSchema( + schema->name, (common::TSDataType)schema->data_type, + (common::TSEncoding)schema->encoding, + (common::CompressionType)schema->compression)); + return ret; +} Review Comment: It is weird that registering timeseries returns an error code while registering table does not. Please double-check both sides and make them consistent. ########## cpp/src/cwrapper/tsfile_cwrapper.cc: ########## @@ -0,0 +1,478 @@ +/* + * 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 <reader/qds_without_timegenerator.h> + +#include "common/global.h" +#include "common/tablet.h" +#include "reader/result_set.h" +#include "reader/tsfile_reader.h" +#include "writer/tsfile_writer.h" + +static bool is_init = false; + +Tablet tablet_new_with_device(const char *device_id, char **column_name_list, + TSDataType *data_types, uint32_t 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))); + } + auto *tablet = + new storage::Tablet(device_id, &measurement_list, &data_type_list, max_rows); + tablet->init(); + return tablet; +} + +Tablet tablet_new(const char **column_name_list, TSDataType *data_types, + uint32_t column_num) { + 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))); + } + auto *tablet = new storage::Tablet("", &measurement_list, &data_type_list); + return tablet; +} + +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, \ + type value) { \ + return static_cast<storage::Tablet *>(tablet)->add_value( \ + row_index, 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); + +#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, \ + type value) { \ + return static_cast<storage::Tablet *>(tablet)->add_value( \ + row_index, column_index, value); \ + } + +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); + +void *tablet_get_value(Tablet tablet, uint32_t row_index, uint32_t schema_index, + TSDataType *type) { + common::TSDataType data_type; + void *result = static_cast<storage::Tablet *>(tablet)->get_value( + row_index, schema_index, data_type); + *type = static_cast<TSDataType>(data_type); + return result; +} + +// TsRecord API +TsRecord ts_record_new(const char *device_name, int64_t timestamp, + int timeseries_num) { + auto *record = + new storage::TsRecord(timestamp, device_name, 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; \ + } + +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); + +void init_tsfile_config() { + if (!is_init) { + common::init_config_value(); + is_init = true; + } +} + +TsFileReader tsfile_reader_new(const char *pathname, ERRNO *err_code) { + init_tsfile_config(); + auto reader = new storage::TsFileReader(); + int ret = reader->open(pathname); + if (ret != common::E_OK) { + std::cout << "open file failed" << std::endl; + *err_code = ret; + delete reader; + return nullptr; + } + return reader; +} + +TsFileWriter tsfile_writer_new(const char *pathname, 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); + if (ret != common::E_OK) { + delete writer; + *err_code = ret; + return nullptr; + } + return writer; +} + +TsFileWriter tsfile_writer_new_flag(const char *pathname, mode_t flag, + ERRNO *err_code) { + init_tsfile_config(); + auto *writer = new storage::TsFileWriter(); + 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_close(TsFileWriter writer) { + auto *w = static_cast<storage::TsFileWriter *>(writer); + int ret = w->close(); + delete w; + return ret; +} + +ERRNO tsfile_reader_close(TsFileReader reader) { + auto *ts_reader = (storage::TsFileReader *)reader; + delete ts_reader; + return common::E_OK; +} + +void tsfile_writer_register_table(TsFileWriter writer, TableSchema *schema) { + std::vector<storage::MeasurementSchema *> measurement_schemas; + std::vector<storage::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<storage::ColumnCategory>(cur_schema->column_category)); + } + auto tsfile_writer = static_cast<storage::TsFileWriter *>(writer); + 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_name, + TimeseriesSchema *schema) { + auto *w = (storage::TsFileWriter *)writer; + + int ret = w->register_timeseries( + device_name, storage::MeasurementSchema( + schema->name, (common::TSDataType)schema->data_type, + (common::TSEncoding)schema->encoding, + (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.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_ts_record(TsFileWriter writer, TsRecord data) { + auto *w = static_cast<storage::TsFileWriter *>(writer); + const auto *record = static_cast<storage::TsRecord *>(data); + const int ret = w->write_record(*record); + if (ret == common::E_OK) { + delete record; + } + return ret; +} + +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_flush_data(TsFileWriter writer) { + auto *w = static_cast<storage::TsFileWriter *>(writer); + return w->flush(); +} + +// Query + +ResultSet tsfile_reader_query_table(TsFileReader reader, char *table_name, + char **columns, uint32_t column_num, + timestamp start_time, timestamp end_time) { + auto *r = static_cast<storage::TsFileReader *>(reader); + std::string table_name_str(table_name); + std::vector<std::string> selected_paths; + for (int i = 0; i < column_num; i++) { + std::string column_name(columns[i]); + selected_paths.push_back(table_name_str + "." + column_name); + } + storage::ResultSet *qds = nullptr; + r->query(selected_paths, start_time, end_time, qds); + return qds; +} + +ResultSet tsfile_reader_query_path(TsFileReader reader, char **path_list, + uint32_t path_num, timestamp start_time, + timestamp end_time) { + auto *r = static_cast<storage::TsFileReader *>(reader); + std::vector<std::string> selected_paths; + for (int i = 0; i < path_num; i++) { + selected_paths.push_back(path_list[i]); + } + storage::ResultSet *qds = nullptr; + r->query(selected_paths, start_time, end_time, qds); + return qds; +} + +bool tsfile_result_set_has_next(ResultSet result_set) { + auto *r = static_cast<storage::QDSWithoutTimeGenerator *>(result_set); + return r->next(); +} + +#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); \ + 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); + +#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); + +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(ResultSet result_set, + 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::QDSWithoutTimeGenerator *>(result_set); + ResultSetMetaData meta_data; + 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).c_str()); + meta_data.data_types[i] = + static_cast<TSDataType>(result_set_metadata->get_column_type(i)); + } + return meta_data; +} + +char *tsfile_result_set_meta_get_column_name(ResultSet result_set, + uint32_t column_index) { + auto *r = static_cast<storage::ResultSet *>(result_set); + return strdup(r->get_metadata()->get_column_name(column_index).c_str()); +} + +TSDataType tsfile_result_set_meta_get_data_type(ResultSet result_set, + uint32_t column_index) { + auto *r = static_cast<storage::ResultSet *>(result_set); + return static_cast<TSDataType>( + r->get_metadata()->get_column_type(column_index)); +} + +uint32_t tsfile_result_set_meta_get_column_num(ResultSet result_set) { + auto *r = static_cast<storage::ResultSet *>(result_set); + return r->get_metadata()->get_column_count(); +} + +TableSchema tsfile_reader_get_table_schema(TsFileReader reader, + const char *table_name) { + auto *r = static_cast<storage::TsFileReader *>(reader); + std::vector<storage::MeasurementSchema> schemas; + r->get_timeseries_schema( + std::make_shared<storage::StringArrayDeviceID>(table_name), schemas); + TableSchema schema; + schema.table_name = strdup(table_name); + schema.column_num = schemas.size(); + schema.column_schemas = static_cast<ColumnSchema *>( + malloc(sizeof(ColumnSchema) * schema.column_num)); + + for (uint32_t i = 0; i < schemas.size(); i++) { + schema.column_schemas[i].column_category = FIELD; + schema.column_schemas[i].column_name = + strdup(schemas[i].measurement_name_.c_str()); + schema.column_schemas[i].data_type = + static_cast<TSDataType>(schemas[i].data_type_); + } + return schema; +} Review Comment: Better to use an empty implementation. ########## cpp/src/cwrapper/tsfile_cwrapper.cc: ########## @@ -0,0 +1,478 @@ +/* + * 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 <reader/qds_without_timegenerator.h> + +#include "common/global.h" +#include "common/tablet.h" +#include "reader/result_set.h" +#include "reader/tsfile_reader.h" +#include "writer/tsfile_writer.h" + +static bool is_init = false; + +Tablet tablet_new_with_device(const char *device_id, char **column_name_list, + TSDataType *data_types, uint32_t 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))); + } + auto *tablet = + new storage::Tablet(device_id, &measurement_list, &data_type_list, max_rows); + tablet->init(); + return tablet; +} + +Tablet tablet_new(const char **column_name_list, TSDataType *data_types, + uint32_t column_num) { + 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))); + } + auto *tablet = new storage::Tablet("", &measurement_list, &data_type_list); + return tablet; +} + +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, \ + type value) { \ + return static_cast<storage::Tablet *>(tablet)->add_value( \ + row_index, 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); + +#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, \ + type value) { \ + return static_cast<storage::Tablet *>(tablet)->add_value( \ + row_index, column_index, value); \ + } + +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); + +void *tablet_get_value(Tablet tablet, uint32_t row_index, uint32_t schema_index, + TSDataType *type) { + common::TSDataType data_type; + void *result = static_cast<storage::Tablet *>(tablet)->get_value( + row_index, schema_index, data_type); + *type = static_cast<TSDataType>(data_type); + return result; +} + +// TsRecord API +TsRecord ts_record_new(const char *device_name, int64_t timestamp, + int timeseries_num) { + auto *record = + new storage::TsRecord(timestamp, device_name, 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; \ + } + +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); + +void init_tsfile_config() { + if (!is_init) { + common::init_config_value(); + is_init = true; + } +} + +TsFileReader tsfile_reader_new(const char *pathname, ERRNO *err_code) { + init_tsfile_config(); + auto reader = new storage::TsFileReader(); + int ret = reader->open(pathname); + if (ret != common::E_OK) { + std::cout << "open file failed" << std::endl; Review Comment: Remove or replace with LOG and add in other functions. ########## cpp/src/cwrapper/tsfile_cwrapper_expression.cc: ########## @@ -0,0 +1,207 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * License); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "tsfile_cwrapper_expression.h" + +#include "common/global.h" +#include "reader/expression.h" +#include "reader/filter/filter.h" +#include "reader/filter/time_filter.h" +#include "reader/filter/time_operator.h" +#include "reader/result_set.h" +#include "reader/tsfile_reader.h" +#include "utils/errno_define.h" +#include "writer/tsfile_writer.h" + +#define E_OK common::E_OK + +#define CONSTRUCT_EXP_INTERNAL(exp, column_name) \ + do { \ + exp.column_name = column_name; \ + exp.operate_type = oper; \ + exp.children_length = 0; \ + } while (0) Review Comment: Maybe 'oper' should be added as a param? It looks weird. ########## cpp/src/cwrapper/tsfile_cwrapper_expression.h: ########## @@ -0,0 +1,103 @@ +/* + * 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. + */ + +#ifndef CWRAPPER_TSFILE_CWRAPPER_H +#define CWRAPPER_TSFILE_CWRAPPER_H + +#include <fcntl.h> +#include <stdbool.h> +#include <stddef.h> +#include <stdint.h> +#ifdef _WIN32 +#include <sys/stat.h> +#endif + +#include "tsfile_cwrapper.h" + +typedef void* TimeFilterExpression; + +#define MAX_COLUMN_FILTER_NUM 10 +typedef enum operator_type { + LT, + LE, + EQ, + GT, + GE, + NOTEQ, +} OperatorType; + +typedef enum expression_type { + OR, + AND, + GLOBALTIME, +} ExpressionType; + +typedef struct constant { + int64_t value_condition; + int type; +} Constant; + +typedef struct expression { + const char* column_name; + Constant const_condition; + ExpressionType expression_type; + OperatorType operate_type; + struct expression* children[MAX_COLUMN_FILTER_NUM]; + int children_length; +} Expression; + +typedef Tablet DataResult; + +typedef void* QueryDataRetINTERNAL; +typedef struct query_data_ret { + char** column_names; + int column_num; + QueryDataRetINTERNAL data; +}* QueryDataRet; Review Comment: QueryDataRet -> ResultSet ########## cpp/src/cwrapper/tsfile_cwrapper.cc: ########## @@ -0,0 +1,478 @@ +/* + * 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 <reader/qds_without_timegenerator.h> + +#include "common/global.h" +#include "common/tablet.h" +#include "reader/result_set.h" +#include "reader/tsfile_reader.h" +#include "writer/tsfile_writer.h" + +static bool is_init = false; + +Tablet tablet_new_with_device(const char *device_id, char **column_name_list, + TSDataType *data_types, uint32_t 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))); + } + auto *tablet = + new storage::Tablet(device_id, &measurement_list, &data_type_list, max_rows); + tablet->init(); + return tablet; +} + +Tablet tablet_new(const char **column_name_list, TSDataType *data_types, + uint32_t column_num) { + 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))); + } + auto *tablet = new storage::Tablet("", &measurement_list, &data_type_list); + return tablet; +} + +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, \ + type value) { \ + return static_cast<storage::Tablet *>(tablet)->add_value( \ + row_index, 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); + +#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, \ + type value) { \ + return static_cast<storage::Tablet *>(tablet)->add_value( \ + row_index, column_index, value); \ + } + +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); + +void *tablet_get_value(Tablet tablet, uint32_t row_index, uint32_t schema_index, + TSDataType *type) { + common::TSDataType data_type; + void *result = static_cast<storage::Tablet *>(tablet)->get_value( + row_index, schema_index, data_type); + *type = static_cast<TSDataType>(data_type); + return result; +} + +// TsRecord API +TsRecord ts_record_new(const char *device_name, int64_t timestamp, + int timeseries_num) { + auto *record = + new storage::TsRecord(timestamp, device_name, 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; \ + } + +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); + +void init_tsfile_config() { + if (!is_init) { + common::init_config_value(); + is_init = true; + } +} + +TsFileReader tsfile_reader_new(const char *pathname, ERRNO *err_code) { + init_tsfile_config(); + auto reader = new storage::TsFileReader(); + int ret = reader->open(pathname); + if (ret != common::E_OK) { + std::cout << "open file failed" << std::endl; + *err_code = ret; + delete reader; + return nullptr; + } + return reader; +} + +TsFileWriter tsfile_writer_new(const char *pathname, 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); + if (ret != common::E_OK) { + delete writer; + *err_code = ret; + return nullptr; + } + return writer; +} + +TsFileWriter tsfile_writer_new_flag(const char *pathname, mode_t flag, + ERRNO *err_code) { + init_tsfile_config(); + auto *writer = new storage::TsFileWriter(); + 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_close(TsFileWriter writer) { + auto *w = static_cast<storage::TsFileWriter *>(writer); + int ret = w->close(); + delete w; + return ret; +} + +ERRNO tsfile_reader_close(TsFileReader reader) { + auto *ts_reader = (storage::TsFileReader *)reader; + delete ts_reader; + return common::E_OK; +} + +void tsfile_writer_register_table(TsFileWriter writer, TableSchema *schema) { + std::vector<storage::MeasurementSchema *> measurement_schemas; + std::vector<storage::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<storage::ColumnCategory>(cur_schema->column_category)); + } + auto tsfile_writer = static_cast<storage::TsFileWriter *>(writer); + 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_name, + TimeseriesSchema *schema) { + auto *w = (storage::TsFileWriter *)writer; + + int ret = w->register_timeseries( + device_name, storage::MeasurementSchema( + schema->name, (common::TSDataType)schema->data_type, + (common::TSEncoding)schema->encoding, + (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.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_ts_record(TsFileWriter writer, TsRecord data) { + auto *w = static_cast<storage::TsFileWriter *>(writer); + const auto *record = static_cast<storage::TsRecord *>(data); + const int ret = w->write_record(*record); + if (ret == common::E_OK) { + delete record; + } + return ret; +} + +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_flush_data(TsFileWriter writer) { + auto *w = static_cast<storage::TsFileWriter *>(writer); + return w->flush(); +} + +// Query + +ResultSet tsfile_reader_query_table(TsFileReader reader, char *table_name, + char **columns, uint32_t column_num, + timestamp start_time, timestamp end_time) { + auto *r = static_cast<storage::TsFileReader *>(reader); + std::string table_name_str(table_name); + std::vector<std::string> selected_paths; + for (int i = 0; i < column_num; i++) { + std::string column_name(columns[i]); + selected_paths.push_back(table_name_str + "." + column_name); + } + storage::ResultSet *qds = nullptr; + r->query(selected_paths, start_time, end_time, qds); + return qds; +} + +ResultSet tsfile_reader_query_path(TsFileReader reader, char **path_list, + uint32_t path_num, timestamp start_time, + timestamp end_time) { + auto *r = static_cast<storage::TsFileReader *>(reader); + std::vector<std::string> selected_paths; + for (int i = 0; i < path_num; i++) { + selected_paths.push_back(path_list[i]); + } + storage::ResultSet *qds = nullptr; + r->query(selected_paths, start_time, end_time, qds); + return qds; +} + +bool tsfile_result_set_has_next(ResultSet result_set) { + auto *r = static_cast<storage::QDSWithoutTimeGenerator *>(result_set); + return r->next(); +} + +#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); \ + 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); + +#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); + +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(ResultSet result_set, + 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::QDSWithoutTimeGenerator *>(result_set); + ResultSetMetaData meta_data; + 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).c_str()); + meta_data.data_types[i] = + static_cast<TSDataType>(result_set_metadata->get_column_type(i)); + } + return meta_data; +} + +char *tsfile_result_set_meta_get_column_name(ResultSet result_set, + uint32_t column_index) { + auto *r = static_cast<storage::ResultSet *>(result_set); + return strdup(r->get_metadata()->get_column_name(column_index).c_str()); +} + +TSDataType tsfile_result_set_meta_get_data_type(ResultSet result_set, + uint32_t column_index) { + auto *r = static_cast<storage::ResultSet *>(result_set); + return static_cast<TSDataType>( + r->get_metadata()->get_column_type(column_index)); +} + +uint32_t tsfile_result_set_meta_get_column_num(ResultSet result_set) { + auto *r = static_cast<storage::ResultSet *>(result_set); + return r->get_metadata()->get_column_count(); +} + +TableSchema tsfile_reader_get_table_schema(TsFileReader reader, + const char *table_name) { + auto *r = static_cast<storage::TsFileReader *>(reader); + std::vector<storage::MeasurementSchema> schemas; + r->get_timeseries_schema( + std::make_shared<storage::StringArrayDeviceID>(table_name), schemas); + TableSchema schema; + schema.table_name = strdup(table_name); + schema.column_num = schemas.size(); + schema.column_schemas = static_cast<ColumnSchema *>( + malloc(sizeof(ColumnSchema) * schema.column_num)); + + for (uint32_t i = 0; i < schemas.size(); i++) { + schema.column_schemas[i].column_category = FIELD; + schema.column_schemas[i].column_name = + strdup(schemas[i].measurement_name_.c_str()); + schema.column_schemas[i].data_type = + static_cast<TSDataType>(schemas[i].data_type_); + } + return schema; +} + +DeviceSchema tsfile_reader_get_timeseries_schema(TsFileReader reader, + const char *device_id) { + auto *r = static_cast<storage::TsFileReader *>(reader); + std::vector<storage::MeasurementSchema> measurement_schemas; + r->get_timeseries_schema(std::make_shared<storage::StringArrayDeviceID>(device_id), measurement_schemas); + DeviceSchema schema; + schema.device_name = strdup(device_id); + schema.timeseries_num = measurement_schemas.size(); + schema.timeseries_schema = static_cast<TimeseriesSchema *>(malloc(sizeof(TimeseriesSchema)* schema.timeseries_num)); + for (uint32_t i = 0; i < schema.timeseries_num; i++) { + schema.timeseries_schema[i].name = strdup(measurement_schemas[i].measurement_name_.c_str()); + schema.timeseries_schema[i].data_type = static_cast<TSDataType>(measurement_schemas[i].data_type_); + schema.timeseries_schema[i].compression = static_cast<CompressionType>(measurement_schemas[i].compression_type_); + schema.timeseries_schema[i].encoding = static_cast<TSEncoding>(measurement_schemas[i].encoding_); + } + return schema; +} + + +TableSchema *tsfile_reader_get_all_table_schemas(TsFileReader reader, + const char *table_name, + uint32_t *num) { + auto *r = static_cast<storage::TsFileReader *>(reader); + std::vector<std::shared_ptr<storage::IDeviceID>> devices = + r->get_all_devices(table_name); + *num = devices.size(); + TableSchema *schemas = static_cast<TableSchema *>( + malloc(sizeof(TableSchema) * devices.size())); + std::vector<storage::MeasurementSchema> measurement_schemas; + for (uint32_t i = 0; i < devices.size(); i++) { + r->get_timeseries_schema(devices[i], measurement_schemas); + schemas[i].table_name = + strdup(devices[i].get()->get_table_name().c_str()); + schemas[i].column_num = measurement_schemas.size(); + for (int j = 0; j < measurement_schemas.size(); j++) { + schemas[i].column_schemas[j].column_category = FIELD; + schemas[i].column_schemas[j].column_name = + strdup(measurement_schemas[j].measurement_name_.c_str()); + schemas[i].column_schemas[j].data_type = + static_cast<TSDataType>(measurement_schemas[j].data_type_); + } + } + return schemas; Review Comment: Why not get it from FileMetadata? ########## cpp/src/cwrapper/tsfile_cwrapper.cc: ########## @@ -0,0 +1,478 @@ +/* + * 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 <reader/qds_without_timegenerator.h> + +#include "common/global.h" +#include "common/tablet.h" +#include "reader/result_set.h" +#include "reader/tsfile_reader.h" +#include "writer/tsfile_writer.h" + +static bool is_init = false; + +Tablet tablet_new_with_device(const char *device_id, char **column_name_list, + TSDataType *data_types, uint32_t 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))); + } + auto *tablet = + new storage::Tablet(device_id, &measurement_list, &data_type_list, max_rows); + tablet->init(); + return tablet; +} + +Tablet tablet_new(const char **column_name_list, TSDataType *data_types, + uint32_t column_num) { + 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))); + } + auto *tablet = new storage::Tablet("", &measurement_list, &data_type_list); + return tablet; +} + +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, \ + type value) { \ + return static_cast<storage::Tablet *>(tablet)->add_value( \ + row_index, 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); + +#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, \ + type value) { \ + return static_cast<storage::Tablet *>(tablet)->add_value( \ + row_index, column_index, value); \ + } + +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); + +void *tablet_get_value(Tablet tablet, uint32_t row_index, uint32_t schema_index, + TSDataType *type) { + common::TSDataType data_type; + void *result = static_cast<storage::Tablet *>(tablet)->get_value( + row_index, schema_index, data_type); + *type = static_cast<TSDataType>(data_type); + return result; +} + +// TsRecord API +TsRecord ts_record_new(const char *device_name, int64_t timestamp, + int timeseries_num) { + auto *record = + new storage::TsRecord(timestamp, device_name, 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; \ + } + +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); + +void init_tsfile_config() { + if (!is_init) { + common::init_config_value(); + is_init = true; + } +} + +TsFileReader tsfile_reader_new(const char *pathname, ERRNO *err_code) { + init_tsfile_config(); + auto reader = new storage::TsFileReader(); + int ret = reader->open(pathname); + if (ret != common::E_OK) { + std::cout << "open file failed" << std::endl; + *err_code = ret; + delete reader; + return nullptr; + } + return reader; +} + +TsFileWriter tsfile_writer_new(const char *pathname, 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); + if (ret != common::E_OK) { + delete writer; + *err_code = ret; + return nullptr; + } + return writer; +} + +TsFileWriter tsfile_writer_new_flag(const char *pathname, mode_t flag, + ERRNO *err_code) { + init_tsfile_config(); + auto *writer = new storage::TsFileWriter(); + 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_close(TsFileWriter writer) { + auto *w = static_cast<storage::TsFileWriter *>(writer); + int ret = w->close(); + delete w; + return ret; +} + +ERRNO tsfile_reader_close(TsFileReader reader) { + auto *ts_reader = (storage::TsFileReader *)reader; + delete ts_reader; + return common::E_OK; +} + +void tsfile_writer_register_table(TsFileWriter writer, TableSchema *schema) { + std::vector<storage::MeasurementSchema *> measurement_schemas; + std::vector<storage::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<storage::ColumnCategory>(cur_schema->column_category)); + } + auto tsfile_writer = static_cast<storage::TsFileWriter *>(writer); + 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_name, + TimeseriesSchema *schema) { + auto *w = (storage::TsFileWriter *)writer; + + int ret = w->register_timeseries( + device_name, storage::MeasurementSchema( + schema->name, (common::TSDataType)schema->data_type, + (common::TSEncoding)schema->encoding, + (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.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_ts_record(TsFileWriter writer, TsRecord data) { + auto *w = static_cast<storage::TsFileWriter *>(writer); + const auto *record = static_cast<storage::TsRecord *>(data); + const int ret = w->write_record(*record); + if (ret == common::E_OK) { + delete record; + } + return ret; +} + +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_flush_data(TsFileWriter writer) { + auto *w = static_cast<storage::TsFileWriter *>(writer); + return w->flush(); +} + +// Query + +ResultSet tsfile_reader_query_table(TsFileReader reader, char *table_name, + char **columns, uint32_t column_num, + timestamp start_time, timestamp end_time) { + auto *r = static_cast<storage::TsFileReader *>(reader); + std::string table_name_str(table_name); + std::vector<std::string> selected_paths; + for (int i = 0; i < column_num; i++) { + std::string column_name(columns[i]); + selected_paths.push_back(table_name_str + "." + column_name); + } + storage::ResultSet *qds = nullptr; + r->query(selected_paths, start_time, end_time, qds); + return qds; +} Review Comment: Better to leave a TODO here or just use an empty implementation. ########## cpp/src/cwrapper/tsfile_cwrapper_expression.cc: ########## @@ -0,0 +1,207 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * License); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "tsfile_cwrapper_expression.h" + +#include "common/global.h" +#include "reader/expression.h" +#include "reader/filter/filter.h" +#include "reader/filter/time_filter.h" +#include "reader/filter/time_operator.h" +#include "reader/result_set.h" +#include "reader/tsfile_reader.h" +#include "utils/errno_define.h" +#include "writer/tsfile_writer.h" + +#define E_OK common::E_OK + +#define CONSTRUCT_EXP_INTERNAL(exp, column_name) \ + do { \ + exp.column_name = column_name; \ + exp.operate_type = oper; \ + exp.children_length = 0; \ + } while (0) + +Expression create_column_filter(const char* column_name, OperatorType oper, + int32_t int32_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = int32_value; + exp.const_condition.type = TSDataType::TS_DATATYPE_INT32; + return exp; +} + +Expression create_column_filter(const char* column_name, OperatorType oper, + int64_t int64_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = int64_value; + exp.const_condition.type = TSDataType::TS_DATATYPE_INT64; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + bool bool_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = bool_value ? 1 : 0; + exp.const_condition.type = TSDataType::TS_DATATYPE_BOOLEAN; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + float float_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + memcpy(&exp.const_condition.value_condition, &float_value, sizeof(float)); + exp.const_condition.type = TSDataType::TS_DATATYPE_FLOAT; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + double double_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = double_value; + exp.const_condition.type = TSDataType::TS_DATATYPE_DOUBLE; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + const char* char_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = reinterpret_cast<int64_t>(char_value); + exp.const_condition.type = TSDataType::TS_DATATYPE_TEXT; + return exp; +} + +TimeFilterExpression* create_andquery_timefilter() { + storage::Expression* exp = new storage::Expression(storage::AND_EXPR); + return (TimeFilterExpression*)exp; +} + +TimeFilterExpression* create_time_filter(const char* table_name, + const char* column_name, + OperatorType oper, int64_t timestamp) { + std::string table_name_str(table_name); + std::string column_name_str(column_name); + storage::Path path(table_name_str, column_name_str); + storage::Filter* filter; + switch (oper) { + case GT: + filter = storage::TimeFilter::gt(timestamp); + break; + case LT: + filter = storage::TimeFilter::lt(timestamp); + break; + case EQ: + filter = storage::TimeFilter::eq(timestamp); + break; + case NOTEQ: + filter = storage::TimeFilter::not_eqt(timestamp); + break; + case GE: + filter = storage::TimeFilter::gt_eq(timestamp); + break; + case LE: + filter = storage::TimeFilter::lt_eq(timestamp); + break; + default: + filter = nullptr; + break; + } + storage::Expression* exp = + new storage::Expression(storage::SERIES_EXPR, path, filter); + return (TimeFilterExpression*)exp; +} + +TimeFilterExpression* add_time_filter_to_and_query( + TimeFilterExpression* exp_and, TimeFilterExpression* exp) { + storage::Expression* and_exp = (storage::Expression*)exp_and; + storage::Expression* time_exp = (storage::Expression*)exp; + if (and_exp->left_ == nullptr) { + and_exp->left_ = time_exp; + } else if (and_exp->right_ == nullptr) { + and_exp->right_ = time_exp; + } else { + storage::Expression* new_exp = + new storage::Expression(storage::AND_EXPR); + new_exp->left_ = and_exp->right_; + and_exp->right_ = new_exp; + add_time_filter_to_and_query((TimeFilterExpression*)new_exp, exp); + } + return exp_and; +} + +void destory_time_filter_query(TimeFilterExpression* expression) { + if (expression == nullptr) { + return; + } + + destory_time_filter_query( + (TimeFilterExpression*)((storage::Expression*)expression)->left_); + destory_time_filter_query( + (TimeFilterExpression*)((storage::Expression*)expression)->right_); + storage::Expression* exp = (storage::Expression*)expression; + if (exp->type_ == storage::ExpressionType::SERIES_EXPR) { + delete exp->filter_; + } else { + delete exp; + } +} + +Expression create_global_time_expression(OperatorType oper, int64_t timestamp) { + Expression exp; + exp.operate_type = oper; + exp.expression_type = GLOBALTIME; + exp.const_condition.value_condition = timestamp; + exp.const_condition.type = TSDataType::TS_DATATYPE_INT64; + return exp; +} + +Expression* and_filter_to_and_query(Expression* exp_and, Expression* exp) { + if (exp_and->children_length >= MAX_COLUMN_FILTER_NUM - 1) { + return nullptr; + } + exp_and->children[exp_and->children_length++] = exp; + return exp_and; +} Review Comment: and_filter_to_and_query -> add_filter_to_and_filter ########## cpp/src/cwrapper/tsfile_cwrapper_expression.cc: ########## @@ -0,0 +1,207 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * License); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "tsfile_cwrapper_expression.h" + +#include "common/global.h" +#include "reader/expression.h" +#include "reader/filter/filter.h" +#include "reader/filter/time_filter.h" +#include "reader/filter/time_operator.h" +#include "reader/result_set.h" +#include "reader/tsfile_reader.h" +#include "utils/errno_define.h" +#include "writer/tsfile_writer.h" + +#define E_OK common::E_OK + +#define CONSTRUCT_EXP_INTERNAL(exp, column_name) \ + do { \ + exp.column_name = column_name; \ + exp.operate_type = oper; \ + exp.children_length = 0; \ + } while (0) + +Expression create_column_filter(const char* column_name, OperatorType oper, + int32_t int32_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = int32_value; + exp.const_condition.type = TSDataType::TS_DATATYPE_INT32; + return exp; +} + +Expression create_column_filter(const char* column_name, OperatorType oper, + int64_t int64_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = int64_value; + exp.const_condition.type = TSDataType::TS_DATATYPE_INT64; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + bool bool_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = bool_value ? 1 : 0; + exp.const_condition.type = TSDataType::TS_DATATYPE_BOOLEAN; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + float float_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + memcpy(&exp.const_condition.value_condition, &float_value, sizeof(float)); + exp.const_condition.type = TSDataType::TS_DATATYPE_FLOAT; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + double double_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = double_value; + exp.const_condition.type = TSDataType::TS_DATATYPE_DOUBLE; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + const char* char_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = reinterpret_cast<int64_t>(char_value); + exp.const_condition.type = TSDataType::TS_DATATYPE_TEXT; + return exp; +} + +TimeFilterExpression* create_andquery_timefilter() { + storage::Expression* exp = new storage::Expression(storage::AND_EXPR); + return (TimeFilterExpression*)exp; +} + +TimeFilterExpression* create_time_filter(const char* table_name, + const char* column_name, + OperatorType oper, int64_t timestamp) { + std::string table_name_str(table_name); + std::string column_name_str(column_name); + storage::Path path(table_name_str, column_name_str); Review Comment: Concatenating the table name with the column name does not result in a correct path because the tags are missing. ########## cpp/test/cwrapper/cwrapper_test.cc: ########## @@ -18,30 +18,112 @@ */ #include <gtest/gtest.h> #include <unistd.h> +extern "C" { +#include "cwrapper/tsfile_cwrapper.h" +} -#include "cwrapper/TsFile-cwrapper.h" #include "utils/errno_define.h" using namespace common; namespace cwrapper { #define TSFILE_NAME "cwrapper.tsfile" -class CWrapperTest : public testing::Test {}; +class CWrapperTest : public testing::Test { + protected: + void TearDown() override { remove(TSFILE_NAME); } +}; -TEST_F(CWrapperTest, write_tsfile) { - ErrorCode code = 0; - CTsFileWriter writer = ts_writer_open(TSFILE_NAME, &code); +TEST_F(CWrapperTest, RegisterTimeSeries) { + ERRNO code = 0; + char* temperature = strdup("temperature"); + TimeseriesSchema ts_schema{temperature, TS_DATATYPE_INT32, + TS_ENCODING_PLAIN, TS_COMPRESSION_UNCOMPRESSED}; + TsFileWriter writer = tsfile_writer_new(TSFILE_NAME, &code); + ASSERT_EQ(code, 0); + code = tsfile_writer_register_timeseries(writer, "device1", &ts_schema); ASSERT_EQ(code, 0); - ASSERT_NE(writer, nullptr); - // open again - writer = ts_writer_open(TSFILE_NAME, &code); - ASSERT_EQ(code, E_ALREADY_EXIST); - ts_writer_close(writer); - ASSERT_EQ(writer, nullptr); - ASSERT_EQ(access(TSFILE_NAME, F_OK), 0); - - writer = ts_writer_open(TSFILE_NAME, &code); - ASSERT_EQ(code, E_ALREADY_EXIST); - ASSERT_EQ(writer, nullptr); + free(temperature); } + +TEST_F(CWrapperTest, WriterFlushTabletAndReadData) { + ERRNO code = 0; + const int device_num = 50; + const int measurement_num = 50; + DeviceSchema device_schema[50]; + TsFileWriter writer = tsfile_writer_new(TSFILE_NAME, &code); + ASSERT_EQ(code, 0); + for (int i = 0; i < device_num; i++) { + char* device_name = strdup(("device" + std::to_string(i)).c_str()); + device_schema[i].device_name = device_name; + device_schema[i].timeseries_num = measurement_num; + device_schema[i].timeseries_schema = (TimeseriesSchema*)malloc( + sizeof(TimeseriesSchema) * measurement_num); + for (int j = 0; j < measurement_num; j++) { + TimeseriesSchema* schema = device_schema[i].timeseries_schema + j; + schema->name = strdup(("measurement" + std::to_string(j)).c_str()); + schema->compression = TS_COMPRESSION_UNCOMPRESSED; + schema->data_type = TS_DATATYPE_INT64; + schema->encoding = TS_ENCODING_PLAIN; + } + code = tsfile_writer_register_device(writer, &device_schema[i]); + ASSERT_EQ(code, 0); + } + int max_rows = 100; + for (int i = 0; i < device_num; i++) { + char* device_name = strdup(("device" + std::to_string(i)).c_str()); + char** measurements_name = static_cast<char**>(malloc(measurement_num * sizeof(char*))); + TSDataType *data_types = static_cast<TSDataType*>( + malloc(sizeof(TSDataType) * measurement_num)); + for (int j = 0; j < measurement_num; j++) { + measurements_name[j] = strdup(("measurement" + std::to_string(j)).c_str()); + data_types[j] = TS_DATATYPE_INT64; + } + Tablet tablet = tablet_new_with_device(device_name, measurements_name, data_types, measurement_num, max_rows); + free(device_name); + free(data_types); + for (int j = 0; j < measurement_num; j++) { + free(measurements_name[j]); + } + free(measurements_name); + for (int j = 0; j < measurement_num; j++) { + for (int row = 0; row < max_rows; row++) { + tablet_add_timestamp(tablet, row, 16225600 + row); + } + for (int row = 0; row < max_rows; row++) { + tablet_add_value_by_index_int64_t(tablet, row, j, static_cast<int64_t>(row + j)); + } + } + code = tsfile_writer_write_tablet(writer, tablet); + ASSERT_EQ(code, 0); + code = destroy_tablet(tablet); + } + ASSERT_EQ(tsfile_writer_flush_data(writer), 0); + ASSERT_EQ(tsfile_writer_close(writer), 0); + + TsFileReader reader = tsfile_reader_new(TSFILE_NAME, &code); + ASSERT_EQ(code, 0); + + char** select_list = static_cast<char**>(malloc(measurement_num * sizeof(char*))); + for (int i = 0; i < measurement_num; i++) { + select_list[i] = strdup(("device" + std::to_string(i) + ".measurement" + std::to_string(i)).c_str()); + } + ResultSet result_set = tsfile_reader_query_path(reader, select_list, measurement_num, 16225600, 16225600 + max_rows -1); + + ResultSetMetaData metadata = tsfile_result_set_get_metadata(result_set); + ASSERT_EQ(metadata.column_num, measurement_num); + ASSERT_EQ(std::string(metadata.column_names[4]), std::string("device4.measurement4")); + ASSERT_EQ(metadata.data_types[9], TS_DATATYPE_INT64); + for (int i = 0; i < measurement_num - 1; i++) { + ASSERT_TRUE(tsfile_result_set_has_next(result_set)); + ASSERT_FALSE(tsfile_result_set_is_null_by_index(result_set,i)); + ASSERT_EQ(tsfile_result_set_get_value_by_index_int64_t(result_set, i), i*2); + ASSERT_EQ(tsfile_result_set_get_value_by_name_int64_t(result_set, std::string("measurement" + std::to_string(i)).c_str()), i*2); + } Review Comment: Better to check the whole data set ########## cpp/src/cwrapper/tsfile_cwrapper_expression.cc: ########## @@ -0,0 +1,207 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * License); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "tsfile_cwrapper_expression.h" + +#include "common/global.h" +#include "reader/expression.h" +#include "reader/filter/filter.h" +#include "reader/filter/time_filter.h" +#include "reader/filter/time_operator.h" +#include "reader/result_set.h" +#include "reader/tsfile_reader.h" +#include "utils/errno_define.h" +#include "writer/tsfile_writer.h" + +#define E_OK common::E_OK + +#define CONSTRUCT_EXP_INTERNAL(exp, column_name) \ + do { \ + exp.column_name = column_name; \ + exp.operate_type = oper; \ + exp.children_length = 0; \ + } while (0) + +Expression create_column_filter(const char* column_name, OperatorType oper, + int32_t int32_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = int32_value; + exp.const_condition.type = TSDataType::TS_DATATYPE_INT32; + return exp; +} + +Expression create_column_filter(const char* column_name, OperatorType oper, + int64_t int64_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = int64_value; + exp.const_condition.type = TSDataType::TS_DATATYPE_INT64; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + bool bool_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = bool_value ? 1 : 0; + exp.const_condition.type = TSDataType::TS_DATATYPE_BOOLEAN; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + float float_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + memcpy(&exp.const_condition.value_condition, &float_value, sizeof(float)); + exp.const_condition.type = TSDataType::TS_DATATYPE_FLOAT; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + double double_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = double_value; + exp.const_condition.type = TSDataType::TS_DATATYPE_DOUBLE; + return exp; +} +Expression create_column_filter(const char* column_name, OperatorType oper, + const char* char_value) { + Expression exp; + CONSTRUCT_EXP_INTERNAL(exp, column_name); + exp.const_condition.value_condition = reinterpret_cast<int64_t>(char_value); + exp.const_condition.type = TSDataType::TS_DATATYPE_TEXT; + return exp; +} + +TimeFilterExpression* create_andquery_timefilter() { + storage::Expression* exp = new storage::Expression(storage::AND_EXPR); + return (TimeFilterExpression*)exp; +} + +TimeFilterExpression* create_time_filter(const char* table_name, + const char* column_name, + OperatorType oper, int64_t timestamp) { + std::string table_name_str(table_name); + std::string column_name_str(column_name); + storage::Path path(table_name_str, column_name_str); + storage::Filter* filter; + switch (oper) { + case GT: + filter = storage::TimeFilter::gt(timestamp); + break; + case LT: + filter = storage::TimeFilter::lt(timestamp); + break; + case EQ: + filter = storage::TimeFilter::eq(timestamp); + break; + case NOTEQ: + filter = storage::TimeFilter::not_eqt(timestamp); + break; + case GE: + filter = storage::TimeFilter::gt_eq(timestamp); + break; + case LE: + filter = storage::TimeFilter::lt_eq(timestamp); + break; + default: + filter = nullptr; + break; + } + storage::Expression* exp = + new storage::Expression(storage::SERIES_EXPR, path, filter); + return (TimeFilterExpression*)exp; +} + +TimeFilterExpression* add_time_filter_to_and_query( + TimeFilterExpression* exp_and, TimeFilterExpression* exp) { + storage::Expression* and_exp = (storage::Expression*)exp_and; + storage::Expression* time_exp = (storage::Expression*)exp; + if (and_exp->left_ == nullptr) { + and_exp->left_ = time_exp; + } else if (and_exp->right_ == nullptr) { + and_exp->right_ = time_exp; + } else { + storage::Expression* new_exp = + new storage::Expression(storage::AND_EXPR); + new_exp->left_ = and_exp->right_; + and_exp->right_ = new_exp; + add_time_filter_to_and_query((TimeFilterExpression*)new_exp, exp); + } + return exp_and; +} + +void destory_time_filter_query(TimeFilterExpression* expression) { + if (expression == nullptr) { + return; + } + + destory_time_filter_query( + (TimeFilterExpression*)((storage::Expression*)expression)->left_); + destory_time_filter_query( + (TimeFilterExpression*)((storage::Expression*)expression)->right_); + storage::Expression* exp = (storage::Expression*)expression; + if (exp->type_ == storage::ExpressionType::SERIES_EXPR) { + delete exp->filter_; + } else { + delete exp; + } Review Comment: 'exp' is not deleted in the first branch? Is it necessary to make this judge? Can you let it delete its 'filter_' in its destruction? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
