ColinLeeo commented on code in PR #418:
URL: https://github.com/apache/tsfile/pull/418#discussion_r1972982258
##########
cpp/src/cwrapper/tsfile_cwrapper.cc:
##########
@@ -19,292 +19,359 @@
#include "cwrapper/tsfile_cwrapper.h"
+#include <file/write_file.h>
#include <reader/qds_without_timegenerator.h>
+#include <unistd.h>
+#include <writer/tsfile_table_writer.h>
#include "common/tablet.h"
#include "reader/result_set.h"
#include "reader/tsfile_reader.h"
#include "writer/tsfile_writer.h"
+#ifdef __cplusplus
+extern "C" {
+#endif
+
static bool is_init = false;
-Tablet tablet_new_with_device(const char *device_id, char **column_name_list,
- TSDataType *data_types, int column_num,
- int max_rows) {
- std::vector<std::string> measurement_list;
- std::vector<common::TSDataType> data_type_list;
- for (int i = 0; i < column_num; i++) {
- measurement_list.emplace_back(column_name_list[i]);
- data_type_list.push_back(
- static_cast<common::TSDataType>(*(data_types + i)));
+void init_tsfile_config() {
+ if (!is_init) {
+ common::init_config_value();
+ is_init = true;
+ }
+}
+
+WriteFile write_file_new(const char *pathname, ERRNO *err_code) {
+ int ret;
+ init_tsfile_config();
+
+ if (access(pathname, F_OK) == 0) {
+ *err_code = common::E_ALREADY_EXIST;
+ return nullptr;
+ }
+
+ int flags = O_RDWR | O_CREAT | O_TRUNC;
+#ifdef _WIN32
+ flags |= O_BINARY;
+#endif
+ mode_t mode = 0666;
+ storage::WriteFile *file = new storage::WriteFile;
+ ret = file->create(pathname, flags, mode);
+ *err_code = ret;
+ return file;
+}
+
+TsFileWriter tsfile_writer_new(WriteFile file, TableSchema *schema,
+ ERRNO *err_code) {
+ init_tsfile_config();
+ std::vector<common::ColumnSchema> column_schemas;
+ for (int i = 0; i < schema->column_num; i++) {
+ ColumnSchema cur_schema = schema->column_schemas[i];
+ column_schemas.emplace_back(common::ColumnSchema(
+ cur_schema.column_name,
+ static_cast<common::TSDataType>(cur_schema.data_type),
+ static_cast<common::CompressionType>(cur_schema.compression),
+ static_cast<common::TSEncoding>(cur_schema.encoding),
+ static_cast<common::ColumnCategory>(cur_schema.column_category)));
+ }
+
+ // There is no need to free table_schema.
+ storage::TableSchema *table_schema =
+ new storage::TableSchema(schema->table_name, column_schemas);
+ *err_code = common::E_OK;
+ return new storage::TsFileTableWriter(
+ static_cast<storage::WriteFile *>(file), table_schema);
+}
+
+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) {
+ *err_code = ret;
+ delete reader;
+ return nullptr;
}
- auto *tablet = new storage::Tablet(device_id, &measurement_list,
- &data_type_list, max_rows);
- return tablet;
+ return reader;
+}
+
+ERRNO tsfile_writer_close(TsFileWriter writer) {
+ auto *w = static_cast<storage::TsFileTableWriter *>(writer);
+ int ret = w->flush();
+ if (ret != common::E_OK) {
+ return ret;
+ }
+ ret = w->close();
+ if (ret != common::E_OK) {
+ return ret;
+ }
+ delete w;
+ return ret;
}
-Tablet tablet_new(const char **column_name_list, TSDataType *data_types,
- uint32_t column_num) {
+ERRNO tsfile_reader_close(TsFileReader reader) {
+ auto *ts_reader = static_cast<storage::TsFileReader *>(reader);
+ delete ts_reader;
+ return common::E_OK;
+}
+
+Tablet tablet_new(char **column_name_list, TSDataType *data_types,
+ uint32_t column_num, uint32_t max_rows) {
std::vector<std::string> measurement_list;
std::vector<common::TSDataType> data_type_list;
for (uint32_t 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;
+ return new storage::Tablet(measurement_list, data_type_list, max_rows);
}
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) {
+ 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) { \
+ const 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);
+ERRNO tablet_add_value_by_name_string(Tablet tablet, uint32_t row_index,
+ const char *column_name,
+ const char *value) {
+ return static_cast<storage::Tablet *>(tablet)->add_value(
+ row_index, column_name, common::String(value));
+}
+
#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) { \
+ const type value) { \
return static_cast<storage::Tablet *>(tablet)->add_value( \
row_index, column_index, value); \
}
+ERRNO tablet_add_value_by_index_string(Tablet tablet, uint32_t row_index,
+ uint32_t column_index,
+ const char *value) {
+ return static_cast<storage::Tablet *>(tablet)->add_value(
+ row_index, column_index, common::String(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_id, timestamp timestamp,
- int timeseries_num) {
- auto *record = new storage::TsRecord(timestamp, device_id, timeseries_num);
- return record;
+TsRecord ts_record_new(const char *device_id, Timestamp timestamp,
+ int timeseries_num) {
+auto *record = new storage::TsRecord(timestamp, device_id,
+timeseries_num); return record;
}
#define INSERT_DATA_INTO_TS_RECORD_BY_NAME_DEF(type) \
- ERRNO insert_data_into_ts_record_by_name_##type( \
- TsRecord data, const char *measurement_name, type value) { \
- auto *record = (storage::TsRecord *)data; \
- storage::DataPoint point(measurement_name, value); \
- if (record->points_.size() + 1 > record->points_.capacity()) \
- return common::E_BUF_NOT_ENOUGH; \
- record->points_.push_back(point); \
- return common::E_OK; \
- }
+ERRNO insert_data_into_ts_record_by_name_##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) {
- *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_with_conf(const char *pathname,
- const mode_t flag, ERRNO *err_code,
- TsFileConf *conf) {
- init_tsfile_config();
- auto *writer = new storage::TsFileWriter();
- const int ret = writer->open(pathname, O_CREAT | O_RDWR, flag);
- if (ret != common::E_OK) {
- delete writer;
- *err_code = ret;
- return nullptr;
- }
- return writer;
+ const mode_t flag, ERRNO *err_code,
+ TsFileConf *conf) {
+init_tsfile_config();
+auto *writer = new storage::TsFileWriter();
+const int ret = writer->open(pathname, O_CREAT | O_RDWR, flag);
+if (ret != common::E_OK) {
+ delete writer;
+ *err_code = ret;
+ return nullptr;
}
-
-ERRNO tsfile_writer_close(TsFileWriter writer) {
- auto *w = static_cast<storage::TsFileWriter *>(writer);
- int ret = w->close();
- delete w;
- return ret;
+return writer;
}
-ERRNO tsfile_reader_close(TsFileReader reader) {
- auto *ts_reader = static_cast<storage::TsFileReader *>(reader);
- delete ts_reader;
- return common::E_OK;
+ERRNO tsfile_writer_register_table(TsFileWriter writer, TableSchema *schema)
+{
+std::vector<storage::MeasurementSchema *> measurement_schemas;
+std::vector<common::ColumnCategory> column_categories;
+measurement_schemas.resize(schema->column_num);
+for (int i = 0; i < schema->column_num; i++) {
+ ColumnSchema *cur_schema = schema->column_schemas + i;
+ measurement_schemas[i] = new storage::MeasurementSchema(
+ cur_schema->column_name,
+ static_cast<common::TSDataType>(cur_schema->data_type));
+ column_categories.push_back(
+ static_cast<common::ColumnCategory>(cur_schema->column_category));
}
-
-ERRNO tsfile_writer_register_table(TsFileWriter writer, TableSchema *schema) {
- std::vector<storage::MeasurementSchema *> measurement_schemas;
- std::vector<common::ColumnCategory> column_categories;
- measurement_schemas.resize(schema->column_num);
- for (int i = 0; i < schema->column_num; i++) {
- ColumnSchema *cur_schema = schema->column_schemas + i;
- measurement_schemas[i] = new storage::MeasurementSchema(
- cur_schema->column_name,
- static_cast<common::TSDataType>(cur_schema->data_type));
- column_categories.push_back(
- static_cast<common::ColumnCategory>(cur_schema->column_category));
- }
- auto tsfile_writer = static_cast<storage::TsFileWriter *>(writer);
- return
tsfile_writer->register_table(std::make_shared<storage::TableSchema>(
- schema->table_name, measurement_schemas, column_categories));
+auto tsfile_writer = static_cast<storage::TsFileWriter *>(writer);
+return
+tsfile_writer->register_table(std::make_shared<storage::TableSchema>(
+ schema->table_name, measurement_schemas, column_categories));
}
ERRNO tsfile_writer_register_timeseries(TsFileWriter writer,
- const char *device_id,
- const TimeseriesSchema *schema) {
- auto *w = static_cast<storage::TsFileWriter *>(writer);
+ const char *device_id,
+ const TimeseriesSchema *schema) {
+auto *w = static_cast<storage::TsFileWriter *>(writer);
- int ret = w->register_timeseries(
- device_id,
- storage::MeasurementSchema(
- schema->timeseries_name,
- static_cast<common::TSDataType>(schema->data_type),
- static_cast<common::TSEncoding>(schema->encoding),
- static_cast<common::CompressionType>(schema->compression)));
- return ret;
+int ret = w->register_timeseries(
+ device_id,
+ storage::MeasurementSchema(
+ schema->timeseries_name,
+ static_cast<common::TSDataType>(schema->data_type),
+ static_cast<common::TSEncoding>(schema->encoding),
+ static_cast<common::CompressionType>(schema->compression)));
+return ret;
}
ERRNO tsfile_writer_register_device(TsFileWriter writer,
- const device_schema *device_schema) {
- auto *w = static_cast<storage::TsFileWriter *>(writer);
- for (int column_id = 0; column_id < device_schema->timeseries_num;
- column_id++) {
- TimeseriesSchema schema = device_schema->timeseries_schema[column_id];
- const ERRNO ret = w->register_timeseries(
- device_schema->device_name,
- storage::MeasurementSchema(
- schema.timeseries_name,
- static_cast<common::TSDataType>(schema.data_type),
- static_cast<common::TSEncoding>(schema.encoding),
- static_cast<common::CompressionType>(schema.compression)));
- if (ret != common::E_OK) {
- return ret;
- }
+ const device_schema *device_schema) {
+auto *w = static_cast<storage::TsFileWriter *>(writer);
+for (int column_id = 0; column_id < device_schema->timeseries_num;
+ column_id++) {
+ TimeseriesSchema schema =
+ device_schema->timeseries_schema[column_id]; const ERRNO ret =
+ w->register_timeseries(
+ device_schema->device_name,
+ storage::MeasurementSchema(
+ schema.timeseries_name,
+ static_cast<common::TSDataType>(schema.data_type),
+ static_cast<common::TSEncoding>(schema.encoding),
+ static_cast<common::CompressionType>(schema.compression)));
+ if (ret != common::E_OK) {
+ return ret;
}
- return common::E_OK;
+}
+return common::E_OK;
}
ERRNO tsfile_writer_write_ts_record(TsFileWriter writer, TsRecord data) {
- auto *w = static_cast<storage::TsFileWriter *>(writer);
- const storage::TsRecord *record = static_cast<storage::TsRecord *>(data);
- const int ret = w->write_record(*record);
- return ret;
+auto *w = static_cast<storage::TsFileWriter *>(writer);
+const storage::TsRecord *record = static_cast<storage::TsRecord *>(data);
+const int ret = w->write_record(*record);
+return ret;
}
ERRNO tsfile_writer_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);
+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();
+*/
+ERRNO tsfile_writer_write(TsFileWriter writer, Tablet tablet) {
+ auto *w = static_cast<storage::TsFileTableWriter *>(writer);
+ auto *tbl = static_cast<storage::Tablet *>(tablet);
+ return w->write_table(*tbl);
}
-// Query
+// ERRNO tsfile_writer_flush_data(TsFileWriter writer) {
+// auto *w = static_cast<storage::TsFileWriter *>(writer);
+// return w->flush();
+// }
-ResultSet tsfile_reader_query_table(TsFileReader reader, const char
*table_name,
-char **columns, uint32_t column_num,
- timestamp start_time, timestamp end_time) {
- // TODO: Implement query table with tsfile reader.
- return nullptr;
-}
+// Query
-ResultSet tsfile_reader_query_device(TsFileReader reader, const char*
device_name,
- char** sensor_name, uint32_t sensor_num,
- timestamp start_time, timestamp end_time) {
+ResultSet tsfile_query_table(TsFileReader reader, const char *table_name,
+ char **columns, uint32_t column_num,
+ Timestamp start_time, Timestamp end_time,
+ ERRNO *err_code) {
auto *r = static_cast<storage::TsFileReader *>(reader);
- std::vector<std::string> selected_paths;
- selected_paths.reserve(sensor_num);
- for (uint32_t i = 0; i < sensor_num; i++) {
- selected_paths.push_back(std::string(device_name) + "." +
std::string(sensor_name[i]));
+ storage::ResultSet *table_result_set = nullptr;
+ std::vector<std::string> column_names;
+ for (uint32_t i = 0; i < column_num; i++) {
+ column_names.emplace_back(columns[i]);
+ }
+ *err_code = r->query(table_name, column_names, start_time, end_time,
+ table_result_set);
+ return table_result_set;
+}
+
+// ResultSet tsfile_reader_query_device(TsFileReader reader,
+// const char *device_name,
+// char **sensor_name, uint32_t
sensor_num,
+// Timestamp start_time, Timestamp
+// end_time) {
+// auto *r = static_cast<storage::TsFileReader *>(reader);
+// std::vector<std::string> selected_paths;
+// selected_paths.reserve(sensor_num);
+// for (uint32_t i = 0; i < sensor_num; i++) {
+// selected_paths.push_back(std::string(device_name) + "." +
+// std::string(sensor_name[i]));
+// }
+// storage::ResultSet *qds = nullptr;
+// r->query(selected_paths, start_time, end_time, qds);
+// return qds;
+// }
+
+bool tsfile_result_set_next(ResultSet result_set, ERRNO *err_code) {
+ auto *r = static_cast<storage::TableResultSet *>(result_set);
+ bool has_next = true;
+ int ret = common::E_OK;
+ ret = r->next(has_next);
+ *err_code = ret;
+ if (ret != common::E_OK) {
+ return false;
}
- 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);
- bool has_next = false;
- r->next(has_next);
return has_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);
\
+ auto *r = static_cast<storage::TableResultSet *>(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);
+char *tsfile_result_set_get_value_by_name_string(ResultSet result_set,
+ const char *column_name) {
+ auto *r = static_cast<storage::TableResultSet *>(result_set);
+ common::String *ret = r->get_value<common::String *>(column_name);
+ // Caller should free return's char* 's space.
Review Comment:
Yes,
```c
/**
* @brief Gets string value from current row by column name.
*
* @return char* - String pointer. Caller must free this ptr after usage.
*/
char* tsfile_result_set_get_value_by_name_string(ResultSet result_set,
const char* column_name);
```
--
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: notifications-unsubscr...@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
