You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by ja...@apache.org on 2024/08/29 01:16:40 UTC
(tsfile) branch iotdb updated (127a79f6 -> 999a6a21)
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a change to branch iotdb
in repository https://gitbox.apache.org/repos/asf/tsfile.git
from 127a79f6 change version to 1.2.0-20a94fb3-SNAPSHOT
new efba28a5 Feature/support aligned ts (#192)
new efc8f71d Fix allSatisfy bug in InFilter
new 999a6a21 change version
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
cpp/examples/c_examples/c_examples.c | 1 -
cpp/examples/cpp_examples/cpp_examples.h | 10 +-
cpp/examples/cpp_examples/demo_read.cpp | 80 ++-
cpp/examples/cpp_examples/demo_write.cpp | 26 +-
cpp/examples/examples.cc | 2 +-
cpp/pom.xml | 2 +-
cpp/src/common/allocator/byte_stream.h | 19 +
cpp/src/common/allocator/my_string.h | 24 +
cpp/src/common/config/config.h | 2 +
cpp/src/common/global.cc | 5 +-
cpp/src/common/record.h | 7 +-
cpp/src/common/schema.h | 12 +-
cpp/src/common/statistic.h | 99 +++-
cpp/src/common/tsfile_common.cc | 94 ++-
cpp/src/common/tsfile_common.h | 68 ++-
cpp/src/compress/compressor_factory.h | 3 +-
cpp/src/encoding/bitpack_encoder.h | 3 +-
cpp/src/encoding/dictionary_encoder.h | 3 +-
cpp/src/encoding/plain_encoder.h | 6 +-
cpp/src/encoding/ts2diff_decoder.h | 4 +-
cpp/src/encoding/ts2diff_encoder.h | 4 +-
cpp/src/file/read_file.cc | 8 +-
cpp/src/file/tsfile_io_reader.cc | 63 +-
cpp/src/file/tsfile_io_reader.h | 4 +-
cpp/src/file/tsfile_io_writer.cc | 46 +-
cpp/src/file/tsfile_io_writer.h | 13 +-
cpp/src/file/write_file.cc | 7 +-
cpp/src/reader/aligned_chunk_reader.cc | 631 +++++++++++++++++++++
cpp/src/reader/aligned_chunk_reader.h | 174 ++++++
cpp/src/reader/chunk_reader.cc | 13 +-
cpp/src/reader/chunk_reader.h | 20 +-
cpp/src/reader/ichunk_reader.h | 62 ++
cpp/src/reader/tsfile_series_scan_iterator.cc | 87 ++-
cpp/src/reader/tsfile_series_scan_iterator.h | 33 +-
cpp/src/writer/chunk_writer.cc | 10 +-
cpp/src/writer/page_writer.cc | 28 +-
cpp/src/writer/page_writer.h | 1 -
.../{chunk_writer.cc => time_chunk_writer.cc} | 83 +--
.../writer/{chunk_writer.h => time_chunk_writer.h} | 79 +--
.../writer/{page_writer.cc => time_page_writer.cc} | 98 +---
cpp/src/writer/time_page_writer.h | 127 +++++
cpp/src/writer/tsfile_writer.cc | 444 +++++++++++++--
cpp/src/writer/tsfile_writer.h | 68 ++-
.../{chunk_writer.cc => value_chunk_writer.cc} | 71 +--
.../{chunk_writer.h => value_chunk_writer.h} | 79 ++-
.../{page_writer.cc => value_page_writer.cc} | 121 ++--
.../writer/{page_writer.h => value_page_writer.h} | 143 ++---
cpp/test/common/allocator/alloc_base_test.cc | 32 ++
cpp/test/common/statistic_test.cc | 23 +
cpp/test/writer/chunk_writer_test.cc | 46 +-
cpp/test/writer/page_writer_test.cc | 2 -
cpp/test/writer/time_chunk_writer_test.cc | 82 +++
cpp/test/writer/time_page_writer_test.cc | 84 +++
cpp/test/writer/tsfile_writer_test.cc | 426 ++++++++++++--
cpp/test/writer/value_chunk_writer_test.cc | 101 ++++
cpp/test/writer/value_page_writer_test.cc | 107 ++++
java/common/pom.xml | 2 +-
java/examples/pom.xml | 4 +-
java/pom.xml | 4 +-
java/tools/pom.xml | 6 +-
java/tsfile/pom.xml | 4 +-
.../codegen/templates/FilterOperatorsTemplate.ftl | 5 +
pom.xml | 2 +-
python/pom.xml | 2 +-
64 files changed, 3171 insertions(+), 748 deletions(-)
create mode 100644 cpp/src/reader/aligned_chunk_reader.cc
create mode 100644 cpp/src/reader/aligned_chunk_reader.h
create mode 100644 cpp/src/reader/ichunk_reader.h
copy cpp/src/writer/{chunk_writer.cc => time_chunk_writer.cc} (60%)
copy cpp/src/writer/{chunk_writer.h => time_chunk_writer.h} (53%)
copy cpp/src/writer/{page_writer.cc => time_page_writer.cc} (58%)
create mode 100644 cpp/src/writer/time_page_writer.h
copy cpp/src/writer/{chunk_writer.cc => value_chunk_writer.cc} (67%)
copy cpp/src/writer/{chunk_writer.h => value_chunk_writer.h} (57%)
copy cpp/src/writer/{page_writer.cc => value_page_writer.cc} (57%)
copy cpp/src/writer/{page_writer.h => value_page_writer.h} (54%)
create mode 100644 cpp/test/writer/time_chunk_writer_test.cc
create mode 100644 cpp/test/writer/time_page_writer_test.cc
create mode 100644 cpp/test/writer/value_chunk_writer_test.cc
create mode 100644 cpp/test/writer/value_page_writer_test.cc
(tsfile) 01/03: Feature/support aligned ts (#192)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch iotdb
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit efba28a521f4695b0a9eeb3c5f3ee71626aa62f7
Author: Hongzhi Gao <76...@qq.com>
AuthorDate: Tue Aug 27 11:57:53 2024 +0800
Feature/support aligned ts (#192)
* support aligned ts
* support aligned ts
* Remove invalid URLs
* fix cpp bug that cause tsfile cannot read bu Python API
* Revert "Remove invalid URLs"
This reverts commit d651b2880c789359df895ca7a6a6118f37af1abf.
* Revert "fix cpp bug that cause tsfile cannot read bu Python API"
This reverts commit b7fa6056484b5f95de0113049ec763ff4ada05c6.
* Fix log when closing TsFile.
* fix issues in aligned ts
* remove unused variables
* fix data type in unittest
* rename variables
* Fix the unit tests that are failing.
---
cpp/examples/c_examples/c_examples.c | 1 -
cpp/examples/cpp_examples/cpp_examples.h | 10 +-
cpp/examples/cpp_examples/demo_read.cpp | 80 ++-
cpp/examples/cpp_examples/demo_write.cpp | 26 +-
cpp/examples/examples.cc | 2 +-
cpp/src/common/allocator/byte_stream.h | 19 +
cpp/src/common/allocator/my_string.h | 24 +
cpp/src/common/config/config.h | 2 +
cpp/src/common/global.cc | 5 +-
cpp/src/common/record.h | 7 +-
cpp/src/common/schema.h | 12 +-
cpp/src/common/statistic.h | 99 +++-
cpp/src/common/tsfile_common.cc | 94 ++-
cpp/src/common/tsfile_common.h | 68 ++-
cpp/src/compress/compressor_factory.h | 3 +-
cpp/src/encoding/bitpack_encoder.h | 3 +-
cpp/src/encoding/dictionary_encoder.h | 3 +-
cpp/src/encoding/plain_encoder.h | 6 +-
cpp/src/encoding/ts2diff_decoder.h | 4 +-
cpp/src/encoding/ts2diff_encoder.h | 4 +-
cpp/src/file/read_file.cc | 8 +-
cpp/src/file/tsfile_io_reader.cc | 63 +-
cpp/src/file/tsfile_io_reader.h | 4 +-
cpp/src/file/tsfile_io_writer.cc | 46 +-
cpp/src/file/tsfile_io_writer.h | 13 +-
cpp/src/file/write_file.cc | 7 +-
cpp/src/reader/aligned_chunk_reader.cc | 631 +++++++++++++++++++++
cpp/src/reader/aligned_chunk_reader.h | 174 ++++++
cpp/src/reader/chunk_reader.cc | 13 +-
cpp/src/reader/chunk_reader.h | 20 +-
cpp/src/reader/ichunk_reader.h | 62 ++
cpp/src/reader/tsfile_series_scan_iterator.cc | 87 ++-
cpp/src/reader/tsfile_series_scan_iterator.h | 33 +-
cpp/src/writer/chunk_writer.cc | 10 +-
cpp/src/writer/page_writer.cc | 28 +-
cpp/src/writer/page_writer.h | 1 -
.../{chunk_writer.cc => time_chunk_writer.cc} | 83 +--
cpp/src/writer/time_chunk_writer.h | 109 ++++
.../writer/{page_writer.cc => time_page_writer.cc} | 98 +---
cpp/src/writer/time_page_writer.h | 127 +++++
cpp/src/writer/tsfile_writer.cc | 444 +++++++++++++--
cpp/src/writer/tsfile_writer.h | 68 ++-
.../{chunk_writer.cc => value_chunk_writer.cc} | 71 +--
cpp/src/writer/value_chunk_writer.h | 129 +++++
.../{page_writer.cc => value_page_writer.cc} | 121 ++--
.../writer/{page_writer.h => value_page_writer.h} | 143 ++---
cpp/test/common/allocator/alloc_base_test.cc | 32 ++
cpp/test/common/statistic_test.cc | 23 +
cpp/test/writer/chunk_writer_test.cc | 46 +-
cpp/test/writer/page_writer_test.cc | 2 -
cpp/test/writer/time_chunk_writer_test.cc | 82 +++
cpp/test/writer/time_page_writer_test.cc | 84 +++
cpp/test/writer/tsfile_writer_test.cc | 426 ++++++++++++--
cpp/test/writer/value_chunk_writer_test.cc | 101 ++++
cpp/test/writer/value_page_writer_test.cc | 107 ++++
55 files changed, 3331 insertions(+), 637 deletions(-)
diff --git a/cpp/examples/c_examples/c_examples.c b/cpp/examples/c_examples/c_examples.c
index d2398fae..8e5f8d2d 100644
--- a/cpp/examples/c_examples/c_examples.c
+++ b/cpp/examples/c_examples/c_examples.c
@@ -25,7 +25,6 @@
#include <stdio.h>
#include <unistd.h>
-
#define HANDLE_ERROR(err_no) \
do { \
if (err_no != 0) { \
diff --git a/cpp/examples/cpp_examples/cpp_examples.h b/cpp/examples/cpp_examples/cpp_examples.h
index f8202537..2fdf13cc 100644
--- a/cpp/examples/cpp_examples/cpp_examples.h
+++ b/cpp/examples/cpp_examples/cpp_examples.h
@@ -17,16 +17,16 @@
* under the License.
*/
+#include "common/db_common.h"
#include "common/path.h"
-#include "reader/filter/filter.h"
+#include "common/record.h"
+#include "common/row_record.h"
#include "reader/expression.h"
-#include "reader/tsfile_reader.h"
+#include "reader/filter/filter.h"
#include "reader/qds_with_timegenerator.h"
#include "reader/qds_without_timegenerator.h"
-#include "common/row_record.h"
+#include "reader/tsfile_reader.h"
#include "writer/tsfile_writer.h"
-#include "common/record.h"
-#include "common/db_common.h"
int demo_read();
int demo_write();
diff --git a/cpp/examples/cpp_examples/demo_read.cpp b/cpp/examples/cpp_examples/demo_read.cpp
index 37a0fa33..5b36af3c 100644
--- a/cpp/examples/cpp_examples/demo_read.cpp
+++ b/cpp/examples/cpp_examples/demo_read.cpp
@@ -16,55 +16,53 @@
* specific language governing permissions and limitations
* under the License.
*/
-#include "cpp_examples.h"
-
#include <iostream>
-#include <vector>
#include <string>
+#include <vector>
-std::string field_to_string(storage::Field *value)
-{
- if (value->type_ == common::TEXT) {
- return std::string(value->value_.sval_);
- } else {
- std::stringstream ss;
- switch (value->type_) {
- case common::BOOLEAN:
- ss << (value->value_.bval_ ? "true" : "false");
- break;
- case common::INT32:
- ss << value->value_.ival_;
- break;
- case common::INT64:
- ss << value->value_.lval_;
- break;
- case common::FLOAT:
- ss << value->value_.fval_;
- break;
- case common::DOUBLE:
- ss << value->value_.dval_;
- break;
- case common::NULL_TYPE:
- ss << "NULL";
- break;
- default:
- ASSERT(false);
- break;
+#include "cpp_examples.h"
+
+std::string field_to_string(storage::Field *value) {
+ if (value->type_ == common::TEXT) {
+ return std::string(value->value_.sval_);
+ } else {
+ std::stringstream ss;
+ switch (value->type_) {
+ case common::BOOLEAN:
+ ss << (value->value_.bval_ ? "true" : "false");
+ break;
+ case common::INT32:
+ ss << value->value_.ival_;
+ break;
+ case common::INT64:
+ ss << value->value_.lval_;
+ break;
+ case common::FLOAT:
+ ss << value->value_.fval_;
+ break;
+ case common::DOUBLE:
+ ss << value->value_.dval_;
+ break;
+ case common::NULL_TYPE:
+ ss << "NULL";
+ break;
+ default:
+ ASSERT(false);
+ break;
+ }
+ return ss.str();
}
- return ss.str();
- }
}
-int demo_read()
-{
- std::cout<<"begin to read tsfile from demo_ts.tsfile" << std::endl;
+int demo_read() {
+ std::cout << "begin to read tsfile from demo_ts.tsfile" << std::endl;
std::string device_name = "root.db001.dev001";
std::string measurement_name = "m001";
storage::Path p1(device_name, measurement_name);
std::vector<storage::Path> select_list;
select_list.push_back(p1);
storage::QueryExpression *query_expr =
- storage::QueryExpression::create(select_list, nullptr);
+ storage::QueryExpression::create(select_list, nullptr);
common::init_config_value();
storage::TsFileReader reader;
@@ -75,20 +73,19 @@ int demo_read()
storage::QueryDataSet *qds = nullptr;
ret = reader.query(query_expr, qds);
-
storage::RowRecord *record;
std::cout << "begin to dump data from tsfile ---" << std::endl;
int row_cout = 0;
do {
record = qds->get_next();
if (record) {
- std::cout<< "dump QDS : " << record->get_timestamp() << ",";
+ std::cout << "dump QDS : " << record->get_timestamp() << ",";
if (record) {
int size = record->get_fields()->size();
for (int i = 0; i < size; ++i) {
std::cout << field_to_string(record->get_field(i)) << ",";
}
- std::cout<< std::endl;
+ std::cout << std::endl;
row_cout++;
}
} else {
@@ -96,6 +93,5 @@ int demo_read()
}
} while (true);
- return (0);
+ return (0);
}
-
diff --git a/cpp/examples/cpp_examples/demo_write.cpp b/cpp/examples/cpp_examples/demo_write.cpp
index 7acc4f58..d77b645c 100644
--- a/cpp/examples/cpp_examples/demo_write.cpp
+++ b/cpp/examples/cpp_examples/demo_write.cpp
@@ -17,40 +17,42 @@
* under the License.
*/
-#include "cpp_examples.h"
+#include <time.h>
#include <iostream>
#include <string>
-#include <time.h>
-int demo_write()
-{
+#include "cpp_examples.h"
+
+int demo_write() {
storage::TsFileWriter tsfile_writer;
std::string device_name = "root.db001.dev001";
std::string measurement_name = "m001";
storage::libtsfile_init();
int ret = tsfile_writer.open("cpp_rw.tsfile", O_CREAT | O_RDWR, 0644);
ASSERT(ret == 0);
- ret = tsfile_writer.register_timeseries(device_name, measurement_name, common::INT32, common::PLAIN, common::UNCOMPRESSED);
+ ret = tsfile_writer.register_timeseries(device_name, measurement_name,
+ common::INT32, common::PLAIN,
+ common::UNCOMPRESSED);
ASSERT(ret == 0);
- std::cout<<"get open ret: "<<ret<<std::endl;
+ std::cout << "get open ret: " << ret << std::endl;
int row_count = 100;
for (int i = 1; i < row_count; ++i) {
storage::DataPoint point(measurement_name, 10000 + i);
- storage::TsRecord record(i, device_name,1);
+ storage::TsRecord record(i, device_name, 1);
record.points_.push_back(point);
ret = tsfile_writer.write_record(record);
ASSERT(ret == 0);
}
tsfile_writer.flush();
- std::cout<<"finish flush"<<std::endl;
+ std::cout << "finish flush" << std::endl;
tsfile_writer.close();
- std::cout<<"tsfile closed."<<std::endl;
+ std::cout << "tsfile closed." << std::endl;
storage::libtsfile_destroy();
- std::cout<< "tsfile to destory."<<std::endl;
- std::cout<<"finish writing" << std::endl;
- std::cout<<"will close our files"<<std::endl;
+ std::cout << "tsfile to destory." << std::endl;
+ std::cout << "finish writing" << std::endl;
+ std::cout << "will close our files" << std::endl;
return 0;
}
diff --git a/cpp/examples/examples.cc b/cpp/examples/examples.cc
index 9d20b8bb..845d3f78 100644
--- a/cpp/examples/examples.cc
+++ b/cpp/examples/examples.cc
@@ -17,8 +17,8 @@
* under the License.
*/
-#include "cpp_examples/cpp_examples.h"
#include "c_examples/c_examples.h"
+#include "cpp_examples/cpp_examples.h"
int main() {
// C++ examples
diff --git a/cpp/src/common/allocator/byte_stream.h b/cpp/src/common/allocator/byte_stream.h
index 687839a5..161f4ca7 100644
--- a/cpp/src/common/allocator/byte_stream.h
+++ b/cpp/src/common/allocator/byte_stream.h
@@ -670,6 +670,25 @@ FORCE_INLINE int merge_byte_stream(ByteStream &sea, ByteStream &river,
return ret;
}
+FORCE_INLINE int copy_bs_to_buf(ByteStream &bs, char *src_buf,
+ uint32_t src_buf_len) {
+ ByteStream::BufferIterator buf_iter = bs.init_buffer_iterator();
+ uint32_t copyed_len = 0;
+ while (true) {
+ ByteStream::Buffer buf = buf_iter.get_next_buf();
+ if (buf.buf_ == nullptr) {
+ break;
+ } else {
+ if (src_buf_len - copyed_len < buf.len_) {
+ return E_BUF_NOT_ENOUGH;
+ }
+ memcpy(src_buf + copyed_len, buf.buf_, buf.len_);
+ copyed_len += buf.len_;
+ }
+ }
+ return E_OK;
+}
+
FORCE_INLINE uint32_t get_var_uint_size(
uint32_t
ui32) // return: the length of usigned number after varint encoding.
diff --git a/cpp/src/common/allocator/my_string.h b/cpp/src/common/allocator/my_string.h
index 348c55ed..d7d8890b 100644
--- a/cpp/src/common/allocator/my_string.h
+++ b/cpp/src/common/allocator/my_string.h
@@ -123,6 +123,30 @@ struct String {
}
}
+ bool operator<(const String &other) const {
+ if (this->is_null() && other.is_null()) {
+ return false;
+ }
+ if (this->is_null()) {
+ return true;
+ }
+ if (other.is_null()) {
+ return false;
+ }
+
+ int min_len = std::min(this->len_, other.len_);
+ int cmp = std::memcmp(this->buf_, other.buf_, min_len);
+ if (cmp != 0) {
+ return cmp < 0;
+ }
+
+ return this->len_ < other.len_;
+ }
+
+ bool operator==(const String &that) const {
+ return equal_to(that);
+ }
+
#ifndef NDEBUG
friend std::ostream &operator<<(std::ostream &os, const String &s) {
os << s.len_ << "@";
diff --git a/cpp/src/common/config/config.h b/cpp/src/common/config/config.h
index d2589229..d97bc807 100644
--- a/cpp/src/common/config/config.h
+++ b/cpp/src/common/config/config.h
@@ -48,6 +48,8 @@ typedef struct ConfigValue {
double tsfile_index_bloom_filter_error_percent_;
const char *tsfile_prefix_path_;
TSEncoding time_encoding_type_;
+ TSDataType time_data_type_;
+ CompressionType time_compress_type_;
uint32_t memtable_flusher_poll_interval_seconds_;
int32_t chunk_group_size_threshold_;
int32_t record_count_for_next_mem_check_;
diff --git a/cpp/src/common/global.cc b/cpp/src/common/global.cc
index dfbbfc81..8aa729f6 100644
--- a/cpp/src/common/global.cc
+++ b/cpp/src/common/global.cc
@@ -56,8 +56,9 @@ void init_config_value() {
g_config_value_.chunk_group_size_threshold_ = 128 * 1024 * 1024;
// g_config_value_.tsfile_prefix_path_ = "./data";
g_config_value_.tsfile_prefix_path_ = "";
- // g_config_value_.time_encoding_type_ = TS_2DIFF;
- g_config_value_.time_encoding_type_ = PLAIN;
+ g_config_value_.time_encoding_type_ = TS_2DIFF;
+ g_config_value_.time_data_type_ = INT64;
+ g_config_value_.time_compress_type_ = LZ4;
g_config_value_.memtable_flusher_poll_interval_seconds_ = 1;
}
diff --git a/cpp/src/common/record.h b/cpp/src/common/record.h
index f87df047..1fc275d4 100644
--- a/cpp/src/common/record.h
+++ b/cpp/src/common/record.h
@@ -42,6 +42,7 @@ struct TextType {
* DataPoint is a data value of one measurement of some device.
*/
struct DataPoint {
+ bool isnull = false;
std::string measurement_name_;
common::TSDataType data_type_;
union {
@@ -94,22 +95,26 @@ struct DataPoint {
// text_val_(text) {}
DataPoint(const std::string &measurement_name)
- : measurement_name_(measurement_name) {}
+ : isnull(true), measurement_name_(measurement_name) {}
void set_i32(int32_t i32) {
data_type_ = common::INT32;
u_.i32_val_ = i32;
+ isnull = false;
}
void set_i64(int64_t i64) {
data_type_ = common::INT64;
u_.i64_val_ = i64;
+ isnull = false;
}
void set_float(float f) {
data_type_ = common::FLOAT;
u_.float_val_ = f;
+ isnull = false;
}
void set_double(double d) {
data_type_ = common::DOUBLE;
u_.double_val_ = d;
+ isnull = false;
}
};
diff --git a/cpp/src/common/schema.h b/cpp/src/common/schema.h
index ff733b12..3a457398 100644
--- a/cpp/src/common/schema.h
+++ b/cpp/src/common/schema.h
@@ -24,6 +24,8 @@
#include <string>
#include "common/db_common.h"
+#include "writer/time_chunk_writer.h"
+#include "writer/value_chunk_writer.h"
namespace storage {
class ChunkWriter;
@@ -38,13 +40,15 @@ struct MeasurementSchema {
common::TSEncoding encoding_;
common::CompressionType compression_type_;
storage::ChunkWriter *chunk_writer_;
+ ValueChunkWriter *value_chunk_writer_;
MeasurementSchema()
: measurement_name_(),
data_type_(common::INVALID_DATATYPE),
encoding_(common::INVALID_ENCODING),
compression_type_(common::INVALID_COMPRESSION),
- chunk_writer_(NULL) {}
+ chunk_writer_(nullptr),
+ value_chunk_writer_(nullptr) {}
MeasurementSchema(const std::string &measurement_name,
common::TSDataType data_type, common::TSEncoding encoding,
@@ -53,7 +57,8 @@ struct MeasurementSchema {
data_type_(data_type),
encoding_(encoding),
compression_type_(compression_type),
- chunk_writer_(NULL) {}
+ chunk_writer_(nullptr),
+ value_chunk_writer_(nullptr) {}
};
typedef std::map<std::string, MeasurementSchema *> MeasurementSchemaMap;
@@ -66,7 +71,8 @@ typedef std::pair<MeasurementSchemaMapIter, bool>
struct MeasurementSchemaGroup {
// measurement_name -> MeasurementSchema
MeasurementSchemaMap measurement_schema_map_;
- bool is_aligned_; // currently not used.
+ bool is_aligned_;
+ TimeChunkWriter *time_chunk_writer_ = nullptr;
};
} // end namespace storage
diff --git a/cpp/src/common/statistic.h b/cpp/src/common/statistic.h
index 1a2a91d6..b7df814c 100644
--- a/cpp/src/common/statistic.h
+++ b/cpp/src/common/statistic.h
@@ -120,6 +120,7 @@ class Statistic {
virtual FORCE_INLINE void update(int64_t time, double value) {
ASSERT(false);
}
+ virtual FORCE_INLINE void update(int64_t time) { ASSERT(false); }
virtual int serialize_to(common::ByteStream &out) {
int ret = common::E_OK;
@@ -246,6 +247,34 @@ class Statistic {
return common::E_OK; \
} while (false)
+#define MERGE_TIME_STAT_FROM(StatType, untyped_stat) \
+ do { \
+ if (UNLIKELY(untyped_stat == nullptr)) { \
+ return common::E_INVALID_ARG; \
+ } \
+ StatType *typed_stat = (StatType *)(untyped_stat); \
+ if (UNLIKELY(typed_stat == nullptr)) { \
+ return common::E_TYPE_NOT_MATCH; \
+ } \
+ if (UNLIKELY(typed_stat->count_ == 0)) { \
+ return common::E_OK; \
+ } \
+ if (count_ == 0) { \
+ count_ = typed_stat->count_; \
+ start_time_ = typed_stat->start_time_; \
+ end_time_ = typed_stat->end_time_; \
+ } else { \
+ count_ += typed_stat->count_; \
+ if (typed_stat->start_time_ < start_time_) { \
+ start_time_ = typed_stat->start_time_; \
+ } \
+ if (typed_stat->end_time_ > end_time_) { \
+ end_time_ = typed_stat->end_time_; \
+ } \
+ } \
+ return common::E_OK; \
+ } while (false)
+
#define DEEP_COPY_BOOL_STAT_FROM(StatType, untyped_stat) \
do { \
if (UNLIKELY(untyped_stat == nullptr)) { \
@@ -284,6 +313,21 @@ class Statistic {
return common::E_OK; \
} while (false)
+#define DEEP_COPY_TIME_STAT_FROM(StatType, untyped_stat) \
+ do { \
+ if (UNLIKELY(untyped_stat == nullptr)) { \
+ return common::E_INVALID_ARG; \
+ } \
+ StatType *typed_stat = (StatType *)(untyped_stat); \
+ if (UNLIKELY(typed_stat == nullptr)) { \
+ return common::E_TYPE_NOT_MATCH; \
+ } \
+ count_ = typed_stat->count_; \
+ start_time_ = typed_stat->start_time_; \
+ end_time_ = typed_stat->end_time_; \
+ return common::E_OK; \
+ } while (false)
+
/* ================ Typed Statistics ================*/
class BooleanStatistic : public Statistic {
public:
@@ -657,12 +701,44 @@ class BinaryStatistic : public Statistic
{
// TODO
};
+#endif
-class TimeStatistic : public Statistic
-{
- // TODO
+class TimeStatistic : public Statistic {
+ public:
+ TimeStatistic() {}
+
+ void clone_from(const TimeStatistic &that) {
+ count_ = that.count_;
+ start_time_ = that.start_time_;
+ end_time_ = that.end_time_;
+ }
+
+ FORCE_INLINE void update(int64_t time) {
+ TIME_STAT_UPDATE((time));
+ count_++;
+ }
+
+ FORCE_INLINE common::TSDataType get_type() { return common::VECTOR; }
+
+ int serialize_typed_stat(common::ByteStream &out) { return common::E_OK; }
+ int deserialize_typed_stat(common::ByteStream &in) { return common::E_OK; }
+ int merge_with(Statistic *stat) {
+ MERGE_TIME_STAT_FROM(TimeStatistic, stat);
+ }
+
+ int deep_copy_from(Statistic *stat) {
+ DEEP_COPY_TIME_STAT_FROM(TimeStatistic, stat);
+ }
+
+ std::string to_string() const {
+ const int buf_len = 256;
+ char buf[buf_len];
+ snprintf(buf, buf_len,
+ "{count=%d, start_time=%" PRId64 ", end_time=%" PRId64 "}",
+ count_, start_time_, end_time_);
+ return std::string(buf);
+ }
};
-#endif
FORCE_INLINE uint32_t get_typed_statistic_sizeof(common::TSDataType type) {
uint32_t ret_size = 0;
@@ -685,6 +761,9 @@ FORCE_INLINE uint32_t get_typed_statistic_sizeof(common::TSDataType type) {
case common::TEXT:
ASSERT(false);
break;
+ case common::VECTOR:
+ ret_size = sizeof(TimeStatistic);
+ break;
default:
ASSERT(false);
break;
@@ -714,6 +793,9 @@ FORCE_INLINE Statistic *placement_new_statistic(common::TSDataType type,
case common::TEXT:
ASSERT(false);
break;
+ case common::VECTOR:
+ s = new (buf) TimeStatistic;
+ break;
default:
ASSERT(false);
break;
@@ -753,6 +835,9 @@ FORCE_INLINE void clone_statistic(Statistic *from, Statistic *to,
case common::TEXT:
ASSERT(false);
break;
+ case common::VECTOR:
+ TYPED_CLONE_STATISTIC(TimeStatistic);
+ break;
default:
ASSERT(false);
}
@@ -798,6 +883,9 @@ class StatisticFactory {
case common::TEXT:
ASSERT(false);
break;
+ case common::VECTOR:
+ ALLOC_STATISTIC(TimeStatistic);
+ break;
default:
abort();
ASSERT(false);
@@ -827,6 +915,9 @@ class StatisticFactory {
case common::TEXT:
ASSERT(false);
break;
+ case common::VECTOR:
+ ALLOC_STATISTIC(TimeStatistic);
+ break;
default:
ASSERT(false);
}
diff --git a/cpp/src/common/tsfile_common.cc b/cpp/src/common/tsfile_common.cc
index de91eaa8..3eabeec9 100644
--- a/cpp/src/common/tsfile_common.cc
+++ b/cpp/src/common/tsfile_common.cc
@@ -19,6 +19,8 @@
#include "common/tsfile_common.h"
+#include <algorithm>
+
#include "common/logger/elog.h"
using namespace common;
@@ -48,6 +50,42 @@ int TimeseriesIndex::add_chunk_meta(ChunkMeta *chunk_meta,
/* ================ TSMIterator ================ */
int TSMIterator::init() {
+ // sort chunk_group_meta_list_ ： {[measurementA, offsetA1], [measurementB,
+ // offsetB1], [measurementA, offsetA2], [measurementB, offsetB2]} ->
+ // {[measurementA, offsetA1], [measurementA, offsetA2], [measurementB,
+ // offsetB1], [measurementB, offsetB2]}
+ for (auto chunk_group_meta_iter = chunk_group_meta_list_.begin();
+ chunk_group_meta_iter != chunk_group_meta_list_.end();
+ chunk_group_meta_iter++) {
+ auto chunk_meta_list = chunk_group_meta_iter.get()->chunk_meta_list_;
+ // Use a map to group chunks by measurement_name_
+ std::map<common::String, std::vector<ChunkMeta*>> groups;
+ std::vector<common::String> order;
+ for (auto it = chunk_meta_list.begin(); it != chunk_meta_list.end(); it++) {
+ auto* chunk_meta = it.get();
+ if (groups.find(chunk_meta->measurement_name_) == groups.end()) {
+ order.push_back(chunk_meta->measurement_name_);
+ }
+ groups[chunk_meta->measurement_name_].push_back(chunk_meta);
+ }
+
+ // Sort each group of chunk metas by offset
+ for (auto it = groups.begin(); it != groups.end(); ++it) {
+ std::vector<ChunkMeta*>& group = it->second;
+ std::sort(group.begin(), group.end(), [](ChunkMeta* a, ChunkMeta* b) {
+ return a->offset_of_chunk_header_ < b->offset_of_chunk_header_;
+ });
+ }
+
+ // Clear and refill chunk_group_meta_list
+ chunk_group_meta_iter.get()->chunk_meta_list_.clear();
+ for (const auto& measurement_name : order) {
+ for (auto chunk_meta : groups[measurement_name]) {
+ chunk_group_meta_iter.get()->chunk_meta_list_.push_back(chunk_meta);
+ }
+ }
+ }
+
// FIXME empty list
chunk_group_meta_iter_ = chunk_group_meta_list_.begin();
if (chunk_group_meta_iter_ == chunk_group_meta_list_.end()) {
@@ -169,36 +207,46 @@ int MetaIndexNode::binary_search_children(const String &name, bool exact_search,
std::cout << "Iterating children: " << children_[i]->name_ << std::endl;
}
#endif
+ bool is_aligned = false;
+ if (node_type_ == LEAF_MEASUREMENT && children_.size() == 1 &&
+ children_[0]->name_.len_ == 0) {
+ is_aligned = true;
+ }
// children_[l] <= name < children_[h]
int l = -1;
- int h = (int)children_.size();
- bool found = false;
- while (l < h - 1) {
- int m = (l + h) / 2;
- int cmp = children_[m]->name_.compare(name);
+ if (is_aligned) {
+ l = 0;
+ } else {
+ int h = (int)children_.size();
+ bool found = false;
+ while (l < h - 1) {
+ int m = (l + h) / 2;
+ int cmp = children_[m]->name_.compare(name);
#if DEBUG_SE
- std::cout << "MetaIndexNode::binary_search_children doing, cmp: cur="
- << children_[m]->name_ << ", name=" << name
- << ", exact_search=" << exact_search
- << ", children_.size=" << children_.size() << std::endl;
+ std::cout
+ << "MetaIndexNode::binary_search_children doing, cmp: cur="
+ << children_[m]->name_ << ", name=" << name
+ << ", exact_search=" << exact_search
+ << ", children_.size=" << children_.size() << std::endl;
#endif
- if (cmp == 0) {
- l = m;
- found = true;
- break;
- } else if (cmp > 0) { // children_[m] > name
- h = m;
- } else { // children_[m] < name
- l = m;
+ if (cmp == 0) {
+ l = m;
+ found = true;
+ break;
+ } else if (cmp > 0) { // children_[m] > name
+ h = m;
+ } else { // children_[m] < name
+ l = m;
+ }
}
- }
- if ((l == -1) || (exact_search && !found)) {
+ if ((l == -1) || (exact_search && !found)) {
#if DEBUG_SE
- std::cout << "MetaIndexNode::binary_search_children end, "
- "ret=E_NOT_EXIST, name="
- << name << ", exact_search=" << exact_search << std::endl;
+ std::cout << "MetaIndexNode::binary_search_children end, "
+ "ret=E_NOT_EXIST, name="
+ << name << ", exact_search=" << exact_search << std::endl;
#endif
- return E_NOT_EXIST;
+ return E_NOT_EXIST;
+ }
}
ret_index_entry = *children_[l];
if (l == (int)children_.size() - 1) {
diff --git a/cpp/src/common/tsfile_common.h b/cpp/src/common/tsfile_common.h
index 601d5886..9ac501c4 100644
--- a/cpp/src/common/tsfile_common.h
+++ b/cpp/src/common/tsfile_common.h
@@ -23,6 +23,7 @@
#include <string.h>
#include <iostream>
+#include <map>
#include <string>
#include "common/allocator/my_string.h"
@@ -301,11 +302,32 @@ struct ChunkGroupMeta {
}
};
+class ITimeseriesIndex {
+ public:
+ ITimeseriesIndex() {}
+ ~ITimeseriesIndex() {}
+ virtual common::SimpleList<ChunkMeta *> *get_chunk_meta_list() const {
+ return nullptr;
+ }
+ virtual common::SimpleList<ChunkMeta *> *get_time_chunk_meta_list() const {
+ return nullptr;
+ }
+ virtual common::SimpleList<ChunkMeta *> *get_value_chunk_meta_list() const {
+ return nullptr;
+ }
+
+ virtual common::String get_measurement_name() { return common::String(); }
+ virtual common::TSDataType get_data_type() const {
+ return common::INVALID_DATATYPE;
+ }
+ virtual Statistic *get_statistic() const { return nullptr; }
+};
+
/*
* A TimeseriesIndex may have one or more chunk metas,
* that means we have such a map: <Timeseries, List<ChunkMeta>>.
*/
-class TimeseriesIndex {
+class TimeseriesIndex : public ITimeseriesIndex {
public:
static const uint32_t CHUNK_META_LIST_SERIALIZED_BUF_PAGE_SIZE = 128;
static const uint32_t PAGE_ARENA_PAGE_SIZE = 256;
@@ -353,10 +375,10 @@ class TimeseriesIndex {
FORCE_INLINE void set_measurement_name(common::String &measurement_name) {
measurement_name_.shallow_copy_from(measurement_name);
}
- FORCE_INLINE common::String get_measurement_name() {
+ FORCE_INLINE virtual common::String get_measurement_name() {
return measurement_name_;
}
- common::SimpleList<ChunkMeta *> *get_chunk_meta_list() const {
+ virtual inline common::SimpleList<ChunkMeta *> *get_chunk_meta_list() const {
return chunk_meta_list_;
}
FORCE_INLINE void set_ts_meta_type(char ts_meta_type) {
@@ -365,7 +387,9 @@ class TimeseriesIndex {
FORCE_INLINE void set_data_type(common::TSDataType data_type) {
data_type_ = data_type;
}
- FORCE_INLINE common::TSDataType get_data_type() const { return data_type_; }
+ FORCE_INLINE virtual common::TSDataType get_data_type() const {
+ return data_type_;
+ }
int init_statistic(common::TSDataType data_type) {
statistic_ = StatisticFactory::alloc_statistic(data_type);
if (IS_NULL(statistic_)) {
@@ -374,7 +398,7 @@ class TimeseriesIndex {
statistic_->reset();
return common::E_OK;
}
- Statistic *get_statistic() const { return statistic_; }
+ virtual Statistic *get_statistic() const { return statistic_; }
common::TsID get_ts_id() const { return ts_id_; }
void set_ts_id(const common::TsID &ts_id) {
ts_id_ = ts_id;
@@ -553,6 +577,40 @@ class TimeseriesIndex {
common::SimpleList<ChunkMeta *> *chunk_meta_list_; // for deserialize_from
};
+class AlignedTimeseriesIndex : public ITimeseriesIndex {
+ public:
+ TimeseriesIndex *time_ts_idx_;
+ TimeseriesIndex *value_ts_idx_;
+
+ AlignedTimeseriesIndex() {}
+ ~AlignedTimeseriesIndex() {}
+ virtual common::SimpleList<ChunkMeta *> *get_time_chunk_meta_list() const {
+ return time_ts_idx_->get_chunk_meta_list();
+ }
+ virtual common::SimpleList<ChunkMeta *> *get_value_chunk_meta_list() const {
+ return value_ts_idx_->get_chunk_meta_list();
+ }
+
+ virtual common::String get_measurement_name() {
+ return value_ts_idx_->get_measurement_name();
+ }
+ virtual common::TSDataType get_data_type() const {
+ return time_ts_idx_->get_data_type();
+ }
+ virtual Statistic *get_statistic() const {
+ return value_ts_idx_->get_statistic();
+ }
+
+#ifndef NDEBUG
+ friend std::ostream &operator<<(std::ostream &os,
+ const AlignedTimeseriesIndex &tsi) {
+ os << "time_ts_idx=" << *tsi.time_ts_idx_;
+ os << ", value_ts_idx=" << *tsi.value_ts_idx_;
+ return os;
+ }
+#endif
+};
+
class TSMIterator {
public:
explicit TSMIterator(
diff --git a/cpp/src/compress/compressor_factory.h b/cpp/src/compress/compressor_factory.h
index 4e446559..0624e9df 100644
--- a/cpp/src/compress/compressor_factory.h
+++ b/cpp/src/compress/compressor_factory.h
@@ -21,6 +21,7 @@
#define COMPRESS_COMPRESSOR_FACTORY_H
#include "gzip_compressor.h"
+#include "lz4_compressor.h"
#include "uncompressed_compressor.h"
namespace storage {
@@ -56,7 +57,7 @@ class CompressorFactory {
} else if (type == common::PLA) {
return nullptr;
} else if (type == common::LZ4) {
- return nullptr;
+ ALLOC_AND_RETURN_COMPRESSPR(LZ4Compressor);
} else {
ASSERT(false);
return nullptr;
diff --git a/cpp/src/encoding/bitpack_encoder.h b/cpp/src/encoding/bitpack_encoder.h
index 3a978e01..d169c73e 100644
--- a/cpp/src/encoding/bitpack_encoder.h
+++ b/cpp/src/encoding/bitpack_encoder.h
@@ -170,7 +170,8 @@ class BitPackEncoder {
maxSize += bytesPerGroup;
}
- // Add additional bytes, because each bitpack group has a header of 1 byte and a tail of 1 byte.
+ // Add additional bytes, because each bitpack group has a header of 1
+ // byte and a tail of 1 byte.
maxSize += fullGroups * (1 + 1) + (remainingValues > 0 ? (1 + 1) : 0);
return maxSize;
}
diff --git a/cpp/src/encoding/dictionary_encoder.h b/cpp/src/encoding/dictionary_encoder.h
index 9ea3f6f0..5f2b1437 100644
--- a/cpp/src/encoding/dictionary_encoder.h
+++ b/cpp/src/encoding/dictionary_encoder.h
@@ -96,8 +96,7 @@ class DictionaryEncoder {
values_encoder_.encode_flush(out);
}
- int get_max_byte_size()
- {
+ int get_max_byte_size() {
// 4 bytes for storing dictionary size
return 4 + map_size_ + values_encoder_.get_max_byte_size();
}
diff --git a/cpp/src/encoding/plain_encoder.h b/cpp/src/encoding/plain_encoder.h
index ab1e06f6..71ba3d02 100644
--- a/cpp/src/encoding/plain_encoder.h
+++ b/cpp/src/encoding/plain_encoder.h
@@ -28,8 +28,10 @@ class PlainEncoder : public Encoder {
public:
PlainEncoder() {}
~PlainEncoder() { destroy(); }
- void destroy() { /* do nothing for PlainEncoder */ }
- void reset() { /* do thing for PlainEncoder */ }
+ void destroy() { /* do nothing for PlainEncoder */
+ }
+ void reset() { /* do thing for PlainEncoder */
+ }
FORCE_INLINE int encode(bool value, common::ByteStream &out_stream) {
return common::SerializationUtil::write_i8(value ? 1 : 0, out_stream);
diff --git a/cpp/src/encoding/ts2diff_decoder.h b/cpp/src/encoding/ts2diff_decoder.h
index 8cdd7ca1..2aa23c62 100644
--- a/cpp/src/encoding/ts2diff_decoder.h
+++ b/cpp/src/encoding/ts2diff_decoder.h
@@ -118,7 +118,7 @@ class TS2DIFFDecoder : public Decoder {
};
template <>
-int32_t TS2DIFFDecoder<int32_t>::decode(common::ByteStream &in) {
+inline int32_t TS2DIFFDecoder<int32_t>::decode(common::ByteStream &in) {
int32_t ret_value = stored_value_;
if (UNLIKELY(current_index_ == 0)) {
read_header(in);
@@ -142,7 +142,7 @@ int32_t TS2DIFFDecoder<int32_t>::decode(common::ByteStream &in) {
}
template <>
-int64_t TS2DIFFDecoder<int64_t>::decode(common::ByteStream &in) {
+inline int64_t TS2DIFFDecoder<int64_t>::decode(common::ByteStream &in) {
int64_t ret_value = stored_value_;
if (UNLIKELY(current_index_ == 0)) {
read_header(in);
diff --git a/cpp/src/encoding/ts2diff_encoder.h b/cpp/src/encoding/ts2diff_encoder.h
index 67d2c83d..b5ae74b8 100644
--- a/cpp/src/encoding/ts2diff_encoder.h
+++ b/cpp/src/encoding/ts2diff_encoder.h
@@ -161,7 +161,7 @@ int TS2DIFFEncoder<T>::do_encode(T value, common::ByteStream &out_stream) {
}
template <>
-int TS2DIFFEncoder<int32_t>::flush(common::ByteStream &out_stream) {
+inline int TS2DIFFEncoder<int32_t>::flush(common::ByteStream &out_stream) {
int ret = common::E_OK;
if (write_index_ == -1) {
return common::E_OK;
@@ -187,7 +187,7 @@ int TS2DIFFEncoder<int32_t>::flush(common::ByteStream &out_stream) {
}
template <>
-int TS2DIFFEncoder<int64_t>::flush(common::ByteStream &out_stream) {
+inline int TS2DIFFEncoder<int64_t>::flush(common::ByteStream &out_stream) {
int ret = common::E_OK;
if (write_index_ == -1) {
return common::E_OK;
diff --git a/cpp/src/file/read_file.cc b/cpp/src/file/read_file.cc
index 97385835..cc08bf25 100644
--- a/cpp/src/file/read_file.cc
+++ b/cpp/src/file/read_file.cc
@@ -27,8 +27,8 @@
#include "common/tsfile_common.h"
#ifdef _WIN32
-#include <windows.h>
#include <io.h>
+#include <windows.h>
ssize_t pread(int fd, void *buf, size_t count, uint64_t offset);
#endif
@@ -133,8 +133,7 @@ int ReadFile::read(int32_t offset, char *buf, int32_t buf_size,
} // end namespace storage
#ifdef _WIN32
-ssize_t pread(int fd, void *buf, size_t count, uint64_t offset)
-{
+ssize_t pread(int fd, void *buf, size_t count, uint64_t offset) {
long unsigned int read_bytes = 0;
OVERLAPPED overlapped;
@@ -147,7 +146,8 @@ ssize_t pread(int fd, void *buf, size_t count, uint64_t offset)
SetLastError(0);
bool RF = ReadFile(file, buf, count, &read_bytes, &overlapped);
- // For some reason it errors when it hits end of file so we don't want to check that
+ // For some reason it errors when it hits end of file so we don't want to
+ // check that
if ((RF == 0) && GetLastError() != ERROR_HANDLE_EOF) {
errno = GetLastError();
// printf ("Error reading file : %d\n", GetLastError());
diff --git a/cpp/src/file/tsfile_io_reader.cc b/cpp/src/file/tsfile_io_reader.cc
index 8a455d42..cba47f53 100644
--- a/cpp/src/file/tsfile_io_reader.cc
+++ b/cpp/src/file/tsfile_io_reader.cc
@@ -68,7 +68,7 @@ int TsFileIOReader::alloc_ssi(const std::string &device_path,
if (RET_FAIL(load_timeseries_index_for_ssi(device_path,
measurement_name, ssi))) {
} else if (time_filter != nullptr &&
- !filter_stasify(ssi->timeseries_index_, time_filter)) {
+ !filter_stasify(ssi->itimeseries_index_, time_filter)) {
ret = E_NO_MORE_DATA;
} else if (RET_FAIL(ssi->init_chunk_reader())) {
}
@@ -88,10 +88,10 @@ void TsFileIOReader::revert_ssi(TsFileSeriesScanIterator *ssi) {
}
}
-bool TsFileIOReader::filter_stasify(TimeseriesIndex &ts_index,
+bool TsFileIOReader::filter_stasify(ITimeseriesIndex *ts_index,
Filter *time_filter) {
- ASSERT(ts_index.get_statistic() != nullptr);
- return time_filter->satisfy(ts_index.get_statistic());
+ ASSERT(ts_index->get_statistic() != nullptr);
+ return time_filter->satisfy(ts_index->get_statistic());
}
int TsFileIOReader::load_tsfile_meta_if_necessary() {
@@ -197,6 +197,7 @@ int TsFileIOReader::load_timeseries_index_for_ssi(
int64_t device_ie_end_offset = 0;
MetaIndexEntry measurement_index_entry;
int64_t measurement_ie_end_offset = 0;
+ // bool is_aligned = false;
if (RET_FAIL(load_device_index_entry(device_path, device_index_entry,
device_ie_end_offset))) {
} else if (RET_FAIL(load_measurement_index_entry(
@@ -206,11 +207,18 @@ int TsFileIOReader::load_timeseries_index_for_ssi(
} else if (RET_FAIL(do_load_timeseries_index(
measurement_name, measurement_index_entry.offset_,
measurement_ie_end_offset, ssi->timeseries_index_pa_,
- ssi->timeseries_index_))) {
+ ssi->itimeseries_index_))) {
} else {
#if DEBUG_SE
- std::cout << "load timeseries index: " << ssi->timeseries_index_
- << std::endl;
+ if (measurement_index_entry.name_.len_) {
+ std::cout << "load timeseries index: "
+ << *((TimeseriesIndex *)ssi->itimeseries_index_)
+ << std::endl;
+ } else {
+ std::cout << "load aligned timeseries index: "
+ << *((AlignedTimeseriesIndex *)ssi->itimeseries_index_)
+ << std::endl;
+ }
#endif
}
return ret;
@@ -275,7 +283,6 @@ int TsFileIOReader::load_measurement_index_entry(
<< *top_node << " at file pos " << start_offset << " to " << end_offset
<< std::endl;
#endif
-
// 2. search from top_node in top-down way
if (IS_SUCC(ret)) {
const String measurement_name((char *)measurement_name_str.c_str(),
@@ -377,7 +384,7 @@ int TsFileIOReader::search_from_internal_node(const String &target_name,
int TsFileIOReader::do_load_timeseries_index(
const std::string &measurement_name_str, int64_t start_offset,
int64_t end_offset, PageArena &in_timeseries_index_pa,
- TimeseriesIndex &ret_timeseries_index) {
+ ITimeseriesIndex *&ret_timeseries_index) {
ASSERT(end_offset > start_offset);
int ret = E_OK;
int32_t read_size = (int32_t)(end_offset - start_offset);
@@ -399,19 +406,45 @@ int TsFileIOReader::do_load_timeseries_index(
std::cout << "do_load_timeseries_index, reader file at " << start_offset
<< " to " << end_offset << std::endl;
#endif
+ bool is_aligned = false;
+ AlignedTimeseriesIndex *aligned_ts_idx = nullptr;
while (IS_SUCC(ret)) {
TimeseriesIndex cur_timeseries_index;
PageArena cur_timeseries_index_pa;
cur_timeseries_index_pa.init(512, MOD_TSFILE_READER); // TODO 512
if (RET_FAIL(cur_timeseries_index.deserialize_from(
bs, &cur_timeseries_index_pa))) {
- } else if (cur_timeseries_index.get_measurement_name().equal_to(
- target_measurement_name)) {
- // cur_timeseries_index.set_ts_id(col_desc.ts_id_);
- if (RET_FAIL(ret_timeseries_index.clone_from(
- cur_timeseries_index, &in_timeseries_index_pa))) {
- // log_err("timeseries index clone error, ret=%d", ret);
+ } else if (is_aligned ||
+ cur_timeseries_index.get_data_type() == common::VECTOR) {
+ if (!is_aligned) {
+ is_aligned = true;
+ void *buf = in_timeseries_index_pa.alloc(
+ sizeof(AlignedTimeseriesIndex));
+ aligned_ts_idx = new (buf) AlignedTimeseriesIndex;
+ buf = in_timeseries_index_pa.alloc(sizeof(TimeseriesIndex));
+ aligned_ts_idx->time_ts_idx_ = new (buf) TimeseriesIndex;
+ aligned_ts_idx->time_ts_idx_->clone_from(
+ cur_timeseries_index, &in_timeseries_index_pa);
+ ret_timeseries_index = aligned_ts_idx;
+ } else if (cur_timeseries_index.get_measurement_name().equal_to(
+ target_measurement_name)) {
+ void *buf =
+ in_timeseries_index_pa.alloc(sizeof(TimeseriesIndex));
+ aligned_ts_idx->value_ts_idx_ = new (buf) TimeseriesIndex;
+ aligned_ts_idx->value_ts_idx_->clone_from(
+ cur_timeseries_index, &in_timeseries_index_pa);
+ found = true;
+ break;
}
+ } else if (!is_aligned &&
+ cur_timeseries_index.get_measurement_name().equal_to(
+ target_measurement_name)) {
+ void *buf =
+ in_timeseries_index_pa.alloc(sizeof(TimeseriesIndex));
+ auto ts_idx = new (buf) TimeseriesIndex;
+ ts_idx->clone_from(cur_timeseries_index,
+ &in_timeseries_index_pa);
+ ret_timeseries_index = ts_idx;
found = true;
break;
}
diff --git a/cpp/src/file/tsfile_io_reader.h b/cpp/src/file/tsfile_io_reader.h
index fc95e74e..8c26e547 100644
--- a/cpp/src/file/tsfile_io_reader.h
+++ b/cpp/src/file/tsfile_io_reader.h
@@ -74,7 +74,7 @@ class TsFileIOReader {
int do_load_timeseries_index(const std::string &measurement_name_str,
int64_t start_offset, int64_t end_offset,
common::PageArena &pa,
- TimeseriesIndex &ts_index);
+ ITimeseriesIndex *&ts_index);
int load_timeseries_index_for_ssi(const std::string &device_path,
const std::string &measurement_name,
TsFileSeriesScanIterator *&ssi);
@@ -86,7 +86,7 @@ class TsFileIOReader {
MetaIndexNode *index_node,
MetaIndexEntry &ret_index_entry,
int64_t &ret_end_offset);
- bool filter_stasify(TimeseriesIndex &ts_index, Filter *time_filter);
+ bool filter_stasify(ITimeseriesIndex *ts_index, Filter *time_filter);
private:
ReadFile *read_file_;
diff --git a/cpp/src/file/tsfile_io_writer.cc b/cpp/src/file/tsfile_io_writer.cc
index 3880116e..d1404ef5 100644
--- a/cpp/src/file/tsfile_io_writer.cc
+++ b/cpp/src/file/tsfile_io_writer.cc
@@ -83,13 +83,30 @@ int TsFileIOWriter::start_file() {
return ret;
}
-int TsFileIOWriter::start_flush_chunk_group(const std::string &device_name) {
- int ret = E_OK;
- if (RET_FAIL(write_byte(CHUNK_GROUP_HEADER_MARKER))) {
- } else if (RET_FAIL(write_string(device_name))) {
- } else {
- cur_device_name_ = device_name;
- ASSERT(cur_chunk_group_meta_ == nullptr);
+int TsFileIOWriter::start_flush_chunk_group(const std::string &device_name,
+ bool is_aligned) {
+ int ret = write_byte(CHUNK_GROUP_HEADER_MARKER);
+ if (ret != common::E_OK) {
+ return ret;
+ }
+ ret = write_string(device_name);
+ if (ret != common::E_OK) {
+ return ret;
+ }
+ cur_device_name_ = device_name;
+ ASSERT(cur_chunk_group_meta_ == nullptr);
+ use_prev_alloc_cgm_ = false;
+ for (auto iter = chunk_group_meta_list_.begin();
+ iter != chunk_group_meta_list_.end(); iter++) {
+ common::String cur_device_name((char *)cur_device_name_.c_str(),
+ cur_device_name_.size());
+ if (iter.get()->device_name_.equal_to(cur_device_name)) {
+ use_prev_alloc_cgm_ = true;
+ cur_chunk_group_meta_ = iter.get();
+ break;
+ }
+ }
+ if (!use_prev_alloc_cgm_) {
void *buf = meta_allocator_.alloc(sizeof(*cur_chunk_group_meta_));
if (IS_NULL(buf)) {
ret = E_OOM;
@@ -197,7 +214,11 @@ int TsFileIOWriter::end_flush_chunk(Statistic *chunk_statistic) {
return ret;
}
-int TsFileIOWriter::end_flush_chunk_group() {
+int TsFileIOWriter::end_flush_chunk_group(bool is_aligned) {
+ if (use_prev_alloc_cgm_) {
+ cur_chunk_group_meta_ = nullptr;
+ return common::E_OK;
+ }
int ret = chunk_group_meta_list_.push_back(cur_chunk_group_meta_);
cur_chunk_group_meta_ = nullptr;
return ret;
@@ -340,6 +361,7 @@ int TsFileIOWriter::write_file_index() {
writing_mm, cur_index_node, LEAF_MEASUREMENT))) {
}
}
+
if (IS_SUCC(ret)) {
if (RET_FAIL(alloc_and_init_meta_index_entry(
writing_mm, meta_index_entry, measurement_name))) {
@@ -351,9 +373,11 @@ int TsFileIOWriter::write_file_index() {
if (IS_SUCC(ret)) {
OFFSET_DEBUG("before ts_index written");
- if (RET_FAIL(
- filter.add_path_entry(device_name, measurement_name))) {
- } else if (RET_FAIL(ts_index.serialize_to(write_stream_))) {
+ if (ts_index.get_data_type() != common::VECTOR) {
+ ret = filter.add_path_entry(device_name, measurement_name);
+ }
+
+ if (RET_FAIL(ts_index.serialize_to(write_stream_))) {
} else {
#if DEBUG_SE
std::cout << "ts_index.serialize. ts_index=" << ts_index
diff --git a/cpp/src/file/tsfile_io_writer.h b/cpp/src/file/tsfile_io_writer.h
index 092c543f..07c4b87b 100644
--- a/cpp/src/file/tsfile_io_writer.h
+++ b/cpp/src/file/tsfile_io_writer.h
@@ -64,8 +64,9 @@ class TsFileIOWriter {
cur_chunk_group_meta_(nullptr),
chunk_meta_count_(0),
chunk_group_meta_list_(&meta_allocator_),
+ use_prev_alloc_cgm_(false),
cur_device_name_(),
- file_(NULL),
+ file_(nullptr),
ts_time_index_vector_(),
write_file_created_(false) {}
~TsFileIOWriter() { destroy(); }
@@ -78,7 +79,8 @@ class TsFileIOWriter {
void destroy();
int start_file();
- int start_flush_chunk_group(const std::string &device_name);
+ int start_flush_chunk_group(const std::string &device_name,
+ bool is_aligned = false);
int start_flush_chunk(common::ByteStream &chunk_data,
common::ColumnDesc &col_desc, int32_t num_of_pages);
int start_flush_chunk(common::ByteStream &chunk_data,
@@ -94,11 +96,11 @@ class TsFileIOWriter {
}
int flush_chunk(common::ByteStream &chunk_data);
int end_flush_chunk(Statistic *chunk_statistic);
- int end_flush_chunk_group();
+ int end_flush_chunk_group(bool is_aligned = false);
int end_file();
- FORCE_INLINE std::vector<TimeseriesTimeIndexEntry> &
- get_ts_time_index_vector() {
+ FORCE_INLINE std::vector<TimeseriesTimeIndexEntry>
+ &get_ts_time_index_vector() {
return ts_time_index_vector_;
}
FORCE_INLINE std::string get_file_path() { return file_->get_file_path(); }
@@ -181,6 +183,7 @@ class TsFileIOWriter {
ChunkGroupMeta *cur_chunk_group_meta_;
int32_t chunk_meta_count_; // for debug
common::SimpleList<ChunkGroupMeta *> chunk_group_meta_list_;
+ bool use_prev_alloc_cgm_; // chunk group meta
std::string cur_device_name_;
WriteFile *file_;
std::vector<TimeseriesTimeIndexEntry> ts_time_index_vector_;
diff --git a/cpp/src/file/write_file.cc b/cpp/src/file/write_file.cc
index 112e4349..1f4c30a5 100644
--- a/cpp/src/file/write_file.cc
+++ b/cpp/src/file/write_file.cc
@@ -32,7 +32,7 @@
#include "utils/errno_define.h"
#ifdef _WIN32
-int fsync(int);
+int fsync(int);
#endif
using namespace common;
@@ -136,8 +136,5 @@ int WriteFile::close() {
} // end namespace storage
#ifdef _WIN32
-int fsync(int fd)
-{
- return _commit(fd);
-}
+int fsync(int fd) { return _commit(fd); }
#endif
diff --git a/cpp/src/reader/aligned_chunk_reader.cc b/cpp/src/reader/aligned_chunk_reader.cc
new file mode 100644
index 00000000..0b76353b
--- /dev/null
+++ b/cpp/src/reader/aligned_chunk_reader.cc
@@ -0,0 +1,631 @@
+/*
+ * 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 "aligned_chunk_reader.h"
+
+#include "compress/compressor_factory.h"
+#include "encoding/decoder_factory.h"
+
+using namespace common;
+namespace storage {
+
+int AlignedChunkReader::init(ReadFile *read_file, String m_name,
+ TSDataType data_type, Filter *time_filter) {
+ read_file_ = read_file;
+ measurement_name_.shallow_copy_from(m_name);
+ time_decoder_ = DecoderFactory::alloc_time_decoder();
+ value_decoder_ = nullptr;
+ time_compressor_ = nullptr;
+ value_compressor_ = nullptr;
+ time_filter_ = time_filter;
+ time_uncompressed_buf_ = nullptr;
+ value_uncompressed_buf_ = nullptr;
+ if (IS_NULL(time_decoder_)) {
+ return E_OOM;
+ }
+ return E_OK;
+}
+
+void AlignedChunkReader::reset() {
+ time_chunk_meta_ = nullptr;
+ value_chunk_meta_ = nullptr;
+ time_chunk_header_.reset();
+ value_chunk_header_.reset();
+ cur_time_page_header_.reset();
+ cur_value_page_header_.reset();
+
+ char *file_data_buf = time_in_stream_.get_wrapped_buf();
+ if (file_data_buf != nullptr) {
+ mem_free(file_data_buf);
+ }
+ time_in_stream_.reset();
+ file_data_buf = value_in_stream_.get_wrapped_buf();
+ if (file_data_buf != nullptr) {
+ mem_free(file_data_buf);
+ }
+ value_in_stream_.reset();
+ file_data_time_buf_size_ = 0;
+ file_data_value_buf_size_ = 0;
+ time_chunk_visit_offset_ = 0;
+ value_chunk_visit_offset_ = 0;
+}
+
+void AlignedChunkReader::destroy() {
+ if (time_decoder_ != nullptr) {
+ time_decoder_->~Decoder();
+ DecoderFactory::free(time_decoder_);
+ time_decoder_ = nullptr;
+ }
+ if (value_decoder_ != nullptr) {
+ value_decoder_->~Decoder();
+ DecoderFactory::free(value_decoder_);
+ value_decoder_ = nullptr;
+ }
+ if (time_compressor_ != nullptr) {
+ time_compressor_->~Compressor();
+ CompressorFactory::free(time_compressor_);
+ time_compressor_ = nullptr;
+ }
+ if (value_compressor_ != nullptr) {
+ value_compressor_->~Compressor();
+ CompressorFactory::free(value_compressor_);
+ value_compressor_ = nullptr;
+ }
+ char *buf = time_in_stream_.get_wrapped_buf();
+ if (buf != nullptr) {
+ mem_free(buf);
+ time_in_stream_.clear_wrapped_buf();
+ }
+ cur_time_page_header_.reset();
+ buf = value_in_stream_.get_wrapped_buf();
+ if (buf != nullptr) {
+ mem_free(buf);
+ value_in_stream_.clear_wrapped_buf();
+ }
+ cur_value_page_header_.reset();
+}
+
+int AlignedChunkReader::load_by_aligned_meta(ChunkMeta *time_chunk_meta,
+ ChunkMeta *value_chunk_meta) {
+ int ret = E_OK;
+ time_chunk_meta_ = time_chunk_meta;
+ value_chunk_meta_ = value_chunk_meta;
+#if DEBUG_SE
+ std::cout << "AlignedChunkReader::load_by_meta, meta=" << *time_chunk_meta
+ << ", " << *value_chunk_meta << std::endl;
+#endif
+ /* ================ deserialize time_chunk_header ================*/
+ // TODO configurable
+ file_data_time_buf_size_ = 1024;
+ file_data_value_buf_size_ = 1024;
+ int32_t ret_read_len = 0;
+ char *time_file_data_buf =
+ (char *)mem_alloc(file_data_time_buf_size_, MOD_CHUNK_READER);
+ if (IS_NULL(time_file_data_buf)) {
+ return E_OOM;
+ }
+ ret = read_file_->read(time_chunk_meta_->offset_of_chunk_header_,
+ time_file_data_buf, file_data_time_buf_size_,
+ ret_read_len);
+ if (IS_SUCC(ret) && ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
+ ret = E_TSFILE_CORRUPTED;
+ LOGE("file corrupted, ret=" << ret << ", offset="
+ << time_chunk_meta_->offset_of_chunk_header_
+ << "read_len=" << ret_read_len);
+ mem_free(time_file_data_buf);
+ }
+ if (IS_SUCC(ret)) {
+ time_in_stream_.wrap_from(time_file_data_buf, ret_read_len);
+ if (RET_FAIL(time_chunk_header_.deserialize_from(time_in_stream_))) {
+ } else {
+ time_chunk_visit_offset_ = time_in_stream_.read_pos();
+ }
+ }
+ /* ================ deserialize value_chunk_header ================*/
+ ret_read_len = 0;
+ char *value_file_data_buf =
+ (char *)mem_alloc(file_data_value_buf_size_, MOD_CHUNK_READER);
+ if (IS_NULL(value_file_data_buf)) {
+ return E_OOM;
+ }
+ ret = read_file_->read(value_chunk_meta_->offset_of_chunk_header_,
+ value_file_data_buf, file_data_value_buf_size_,
+ ret_read_len);
+ if (IS_SUCC(ret) && ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
+ ret = E_TSFILE_CORRUPTED;
+ LOGE("file corrupted, ret="
+ << ret << ", offset=" << value_chunk_meta_->offset_of_chunk_header_
+ << "read_len=" << ret_read_len);
+ mem_free(value_file_data_buf);
+ }
+ if (IS_SUCC(ret)) {
+ value_in_stream_.wrap_from(value_file_data_buf, ret_read_len);
+ if (RET_FAIL(value_chunk_header_.deserialize_from(value_in_stream_))) {
+ } else if (RET_FAIL(alloc_compressor_and_decoder(
+ time_decoder_, time_compressor_,
+ time_chunk_header_.encoding_type_,
+ time_chunk_header_.data_type_,
+ time_chunk_header_.compression_type_))) {
+ } else if (RET_FAIL(alloc_compressor_and_decoder(
+ value_decoder_, value_compressor_,
+ value_chunk_header_.encoding_type_,
+ value_chunk_header_.data_type_,
+ value_chunk_header_.compression_type_))) {
+ } else {
+ value_chunk_visit_offset_ = value_in_stream_.read_pos();
+#if DEBUG_SE
+ std::cout << "AlignedChunkReader::load_by_meta, time_chunk_header="
+ << time_chunk_header_
+ << ", value_chunk_header=" << value_chunk_header_
+ << std::endl;
+#endif
+ }
+ }
+ return ret;
+}
+
+int AlignedChunkReader::alloc_compressor_and_decoder(
+ storage::Decoder *&decoder, storage::Compressor *&compressor,
+ TSEncoding encoding, TSDataType data_type, CompressionType compression) {
+ if (decoder != nullptr) {
+ decoder->reset();
+ } else {
+ decoder = DecoderFactory::alloc_value_decoder(encoding, data_type);
+ if (IS_NULL(decoder)) {
+ return E_OOM;
+ }
+ }
+
+ if (compressor != nullptr) {
+ compressor->reset(false);
+ } else {
+ compressor = CompressorFactory::alloc_compressor(compression);
+ if (compressor == nullptr) {
+ return E_OOM;
+ }
+ }
+ return E_OK;
+}
+
+int AlignedChunkReader::get_next_page(TsBlock *ret_tsblock,
+ Filter *oneshoot_filter) {
+ int ret = E_OK;
+ Filter *filter =
+ (oneshoot_filter != nullptr ? oneshoot_filter : time_filter_);
+
+ if (prev_time_page_not_finish() && prev_value_page_not_finish()) {
+ ret = decode_time_value_buf_into_tsblock(ret_tsblock, oneshoot_filter);
+ return ret;
+ }
+ if (!prev_time_page_not_finish() && !prev_value_page_not_finish()) {
+ while (IS_SUCC(ret)) {
+ if (RET_FAIL(get_cur_page_header(
+ time_chunk_meta_, time_in_stream_, cur_time_page_header_,
+ time_chunk_visit_offset_, time_chunk_header_))) {
+ } else if (RET_FAIL(get_cur_page_header(
+ value_chunk_meta_, value_in_stream_,
+ cur_value_page_header_, value_chunk_visit_offset_,
+ value_chunk_header_))) {
+ } else if (cur_page_statisify_filter(filter)) {
+ break;
+ } else if (RET_FAIL(skip_cur_page())) {
+ }
+ }
+ if (IS_SUCC(ret)) {
+ ret = decode_cur_time_page_data() || decode_cur_value_page_data();
+ }
+ }
+ if (IS_SUCC(ret)) {
+ ret = decode_time_value_buf_into_tsblock(ret_tsblock, oneshoot_filter);
+ }
+ return ret;
+}
+
+int AlignedChunkReader::get_cur_page_header(ChunkMeta *&chunk_meta,
+ common::ByteStream &in_stream,
+ PageHeader &cur_page_header,
+ uint32_t &chunk_visit_offset,
+ ChunkHeader &chunk_header) {
+ int ret = E_OK;
+ bool retry = true;
+ int cur_page_header_serialized_size = 0;
+ // TODO： configurable
+ int retry_read_want_size = 1024;
+ do {
+ in_stream.mark_read_pos();
+ cur_page_header.reset();
+ ret = cur_page_header.deserialize_from(
+ in_stream, !chunk_has_only_one_page(chunk_header),
+ chunk_header.data_type_);
+ cur_page_header_serialized_size = in_stream.get_mark_len();
+ if (deserialize_buf_not_enough(ret) && retry) {
+ retry = false;
+ retry_read_want_size += 1024;
+ int32_t file_data_buf_size =
+ chunk_header.data_type_ == common::VECTOR
+ ? file_data_time_buf_size_
+ : file_data_value_buf_size_;
+ if (E_OK == read_from_file_and_rewrap(
+ in_stream, chunk_meta, chunk_visit_offset,
+ file_data_buf_size, retry_read_want_size)) {
+ continue;
+ }
+ }
+ break;
+ } while (true);
+ if (IS_SUCC(ret)) {
+ // visit a header
+ chunk_visit_offset += cur_page_header_serialized_size;
+ }
+#if DEBUG_SE
+ std::cout << "get_cur_page_header, ret=" << ret << ", retry=" << retry
+ << ", cur_page_header=" << cur_page_header
+ << ", chunk_meta->offset_of_chunk_header_="
+ << chunk_meta->offset_of_chunk_header_
+ << ", cur_page_header_serialized_size="
+ << cur_page_header_serialized_size << std::endl;
+#endif
+ return ret;
+}
+
+// reader at least @want_size bytes from file and wrap the buffer into
+// @in_stream_
+int AlignedChunkReader::read_from_file_and_rewrap(
+ common::ByteStream &in_stream_, ChunkMeta *&chunk_meta,
+ uint32_t &chunk_visit_offset, int32_t file_data_buf_size, int want_size) {
+ int ret = E_OK;
+ const int DEFAULT_READ_SIZE = 4096; // may use page_size + page_header_size
+ char *file_data_buf = in_stream_.get_wrapped_buf();
+ int offset = chunk_meta->offset_of_chunk_header_ + chunk_visit_offset;
+ int read_size =
+ (want_size < DEFAULT_READ_SIZE ? DEFAULT_READ_SIZE : want_size);
+ if (file_data_buf_size < read_size || read_size < file_data_buf_size / 10) {
+ file_data_buf = (char *)mem_realloc(file_data_buf, read_size);
+ if (IS_NULL(file_data_buf)) {
+ return E_OOM;
+ }
+ file_data_buf_size = read_size;
+ }
+ int ret_read_len = 0;
+ if (RET_FAIL(
+ read_file_->read(offset, file_data_buf, read_size, ret_read_len))) {
+ } else {
+ in_stream_.wrap_from(file_data_buf, ret_read_len);
+#ifdef DEBUG_SE
+ std::cout << "file offset = " << offset << " len = " << ret_read_len
+ << std::endl;
+ DEBUG_hex_dump_buf("wrapped buf = ", file_data_buf, 256);
+#endif
+ }
+ return ret;
+}
+
+bool AlignedChunkReader::cur_page_statisify_filter(Filter *filter) {
+ bool value_satisfy = filter == nullptr ||
+ cur_value_page_header_.statistic_ == nullptr ||
+ filter->satisfy(cur_value_page_header_.statistic_);
+ bool time_satisfy = filter == nullptr ||
+ cur_time_page_header_.statistic_ == nullptr ||
+ filter->satisfy(cur_time_page_header_.statistic_);
+ return time_satisfy && value_satisfy;
+}
+
+int AlignedChunkReader::skip_cur_page() {
+ int ret = E_OK;
+ // visit a page tv data
+ time_chunk_visit_offset_ += cur_time_page_header_.compressed_size_;
+ time_in_stream_.wrapped_buf_advance_read_pos(
+ cur_time_page_header_.compressed_size_);
+ value_chunk_visit_offset_ += cur_value_page_header_.compressed_size_;
+ value_in_stream_.wrapped_buf_advance_read_pos(
+ cur_value_page_header_.compressed_size_);
+ return ret;
+}
+
+int AlignedChunkReader::decode_cur_time_page_data() {
+ int ret = E_OK;
+
+ // Step 1: make sure we load the whole page data in @in_stream_
+ if (time_in_stream_.remaining_size() <
+ cur_time_page_header_.compressed_size_) {
+ // std::cout << "decode_cur_page_data. in_stream_.remaining_size="<<
+ // in_stream_.remaining_size() << ", cur_page_header_.compressed_size_="
+ // << cur_page_header_.compressed_size_ << std::endl;
+ if (RET_FAIL(read_from_file_and_rewrap(
+ time_in_stream_, time_chunk_meta_, time_chunk_visit_offset_,
+ cur_time_page_header_.compressed_size_,
+ file_data_time_buf_size_))) {
+ }
+ }
+
+ char *time_compressed_buf = nullptr;
+ char *time_uncompressed_buf = nullptr;
+ uint32_t time_compressed_buf_size = 0;
+ uint32_t time_uncompressed_buf_size = 0;
+ char *time_buf = nullptr;
+ uint32_t time_buf_size = 0;
+
+ // Step 2: do uncompress
+ if (IS_SUCC(ret)) {
+ time_compressed_buf =
+ time_in_stream_.get_wrapped_buf() + time_in_stream_.read_pos();
+#ifdef DEBUG_SE
+ std::cout << "AlignedChunkReader::decode_cur_page_data,time_in_stream_."
+ "get_wrapped_buf="
+ << (void *)(time_in_stream_.get_wrapped_buf())
+ << ", time_in_stream_.read_pos=" << time_in_stream_.read_pos()
+ << std::endl;
+#endif
+ time_compressed_buf_size = cur_time_page_header_.compressed_size_;
+ time_in_stream_.wrapped_buf_advance_read_pos(time_compressed_buf_size);
+ time_chunk_visit_offset_ += time_compressed_buf_size;
+ if (RET_FAIL(time_compressor_->reset(false))) {
+ } else if (RET_FAIL(time_compressor_->uncompress(
+ time_compressed_buf, time_compressed_buf_size,
+ time_uncompressed_buf, time_uncompressed_buf_size))) {
+ } else {
+ time_uncompressed_buf_ = time_uncompressed_buf;
+ }
+#ifdef DEBUG_SE
+ DEBUG_hex_dump_buf(
+ "AlignedChunkReader reader, time_uncompressed buf = ",
+ time_uncompressed_buf, time_uncompressed_buf_size);
+#endif
+ if (ret != E_OK || time_uncompressed_buf_size !=
+ cur_time_page_header_.uncompressed_size_) {
+ ret = E_TSFILE_CORRUPTED;
+ ASSERT(false);
+ }
+ }
+
+ // Step 3: get time_buf
+ if (IS_SUCC(ret)) {
+ int var_size = 0;
+ if (RET_FAIL(SerializationUtil::read_var_uint(
+ time_buf_size, time_uncompressed_buf,
+ time_uncompressed_buf_size, &var_size))) {
+ } else {
+ time_buf = time_uncompressed_buf + var_size;
+ if (time_uncompressed_buf_size < var_size + time_buf_size) {
+ ret = E_TSFILE_CORRUPTED;
+ ASSERT(false);
+ }
+ }
+ }
+ time_decoder_->reset();
+#ifdef DEBUG_SE
+ DEBUG_hex_dump_buf("AlignedChunkReader reader, time_buf = ", time_buf,
+ time_buf_size);
+#endif
+ time_in_.wrap_from(time_buf, time_buf_size);
+ return ret;
+}
+
+int AlignedChunkReader::decode_cur_value_page_data() {
+ int ret = E_OK;
+
+ // Step 1: make sure we load the whole page data in @in_stream_
+ if (value_in_stream_.remaining_size() <
+ cur_value_page_header_.compressed_size_) {
+ // std::cout << "decode_cur_page_data. in_stream_.remaining_size="<<
+ // in_stream_.remaining_size() << ", cur_page_header_.compressed_size_="
+ // << cur_page_header_.compressed_size_ << std::endl;
+ if (RET_FAIL(read_from_file_and_rewrap(
+ value_in_stream_, value_chunk_meta_, value_chunk_visit_offset_,
+ cur_value_page_header_.compressed_size_,
+ file_data_value_buf_size_))) {
+ }
+ }
+
+ char *value_compressed_buf = nullptr;
+ char *value_uncompressed_buf = nullptr;
+ uint32_t value_compressed_buf_size = 0;
+ uint32_t value_uncompressed_buf_size = 0;
+ char *value_buf = nullptr;
+ uint32_t value_buf_size = 0;
+
+ // Step 2: do uncompress
+ if (IS_SUCC(ret)) {
+ value_compressed_buf =
+ value_in_stream_.get_wrapped_buf() + value_in_stream_.read_pos();
+ value_compressed_buf_size = cur_value_page_header_.compressed_size_;
+ value_in_stream_.wrapped_buf_advance_read_pos(
+ value_compressed_buf_size);
+ value_chunk_visit_offset_ += value_compressed_buf_size;
+ if (RET_FAIL(value_compressor_->reset(false))) {
+ } else if (RET_FAIL(value_compressor_->uncompress(
+ value_compressed_buf, value_compressed_buf_size,
+ value_uncompressed_buf, value_uncompressed_buf_size))) {
+ } else {
+ value_uncompressed_buf_ = value_uncompressed_buf;
+ }
+#ifdef DEBUG_SE
+ DEBUG_hex_dump_buf(
+ "AlignedChunkReader reader, value_uncompressed buf = ",
+ value_uncompressed_buf, value_uncompressed_buf_size);
+#endif
+ if (ret != E_OK || value_uncompressed_buf_size !=
+ cur_value_page_header_.uncompressed_size_) {
+ ret = E_TSFILE_CORRUPTED;
+ ASSERT(false);
+ }
+ }
+ // Step 3: get value_buf
+ if (IS_SUCC(ret)) {
+ uint32_t value_uncompressed_buf_offset = 0;
+ value_page_data_num_ =
+ SerializationUtil::read_ui32(value_uncompressed_buf);
+ value_uncompressed_buf_offset += sizeof(uint32_t);
+ value_page_col_notnull_bitmap_.resize((value_page_data_num_ + 7) / 8);
+ for (unsigned char &i : value_page_col_notnull_bitmap_) {
+ i = *(value_uncompressed_buf + value_uncompressed_buf_offset);
+ value_uncompressed_buf_offset++;
+ }
+ cur_value_index = -1;
+ value_buf = value_uncompressed_buf + value_uncompressed_buf_offset;
+ value_buf_size =
+ value_uncompressed_buf_size - value_uncompressed_buf_offset;
+ }
+ value_decoder_->reset();
+#ifdef DEBUG_SE
+ DEBUG_hex_dump_buf("AlignedChunkReader reader, value_buf = ", value_buf,
+ value_buf_size);
+#endif
+ value_in_.wrap_from(value_buf, value_buf_size);
+ return ret;
+}
+
+int AlignedChunkReader::decode_time_value_buf_into_tsblock(
+ TsBlock *&ret_tsblock, Filter *filter) {
+ int ret = common::E_OK;
+ ret = decode_tv_buf_into_tsblock_by_datatype(time_in_, value_in_,
+ ret_tsblock, filter);
+ // if we return during @decode_tv_buf_into_tsblock, we should keep
+ // @uncompressed_buf_ valid until all TV pairs are decoded.
+ if (ret != E_OVERFLOW) {
+ if (time_uncompressed_buf_ != nullptr) {
+ time_compressor_->after_uncompress(time_uncompressed_buf_);
+ time_uncompressed_buf_ = nullptr;
+ }
+ if (value_uncompressed_buf_ != nullptr) {
+ value_compressor_->after_uncompress(value_uncompressed_buf_);
+ value_uncompressed_buf_ = nullptr;
+ }
+ if (!prev_value_page_not_finish()) {
+ value_in_.reset();
+ }
+ if (!prev_time_page_not_finish()) {
+ time_in_.reset();
+ }
+ } else {
+ ret = E_OK;
+ }
+ return ret;
+}
+
+#define DECODE_TYPED_TV_INTO_TSBLOCK(CppType, ReadType, time_in, value_in, \
+ row_appender) \
+ do { \
+ uint32_t mask = 1 << 7; \
+ int64_t time = 0; \
+ CppType value; \
+ while ((time_decoder_->has_remaining() && \
+ value_decoder_->has_remaining()) || \
+ (time_in.has_remaining() && value_in.has_remaining())) { \
+ cur_value_index++; \
+ if (((value_page_col_notnull_bitmap_[cur_value_index / 8] & \
+ 0xFF) & \
+ (mask >> (cur_value_index % 8))) == 0) { \
+ RET_FAIL(time_decoder_->read_int64(time, time_in)); \
+ if (ret != E_OK) { \
+ break; \
+ } \
+ } \
+ if (UNLIKELY(!row_appender.add_row())) { \
+ ret = E_OVERFLOW; \
+ break; \
+ } else if (RET_FAIL(time_decoder_->read_int64(time, time_in))) { \
+ } else if (RET_FAIL(value_decoder_->read_##ReadType(value, \
+ value_in))) { \
+ } else if (filter != nullptr && !filter->satisfy(time, value)) { \
+ row_appender.backoff_add_row(); \
+ continue; \
+ } else { \
+ /*std::cout << "decoder: time=" << time << ", value=" << value \
+ * << std::endl;*/ \
+ row_appender.append(0, (char *)&time, sizeof(time)); \
+ row_appender.append(1, (char *)&value, sizeof(value)); \
+ } \
+ } \
+ } while (false)
+
+int AlignedChunkReader::i32_DECODE_TYPED_TV_INTO_TSBLOCK(
+ ByteStream &time_in, ByteStream &value_in, RowAppender &row_appender,
+ Filter *filter) {
+ int ret = E_OK;
+ uint32_t mask = 1 << 7;
+ do {
+ int64_t time = 0;
+ int32_t value;
+ while ((time_decoder_->has_remaining() &&
+ value_decoder_->has_remaining()) ||
+ (time_in.has_remaining() && value_in.has_remaining())) {
+ cur_value_index++;
+ if (((value_page_col_notnull_bitmap_[cur_value_index / 8] & 0xFF) &
+ (mask >> (cur_value_index % 8))) == 0) {
+ RET_FAIL(time_decoder_->read_int64(time, time_in));
+ continue;
+ }
+ if (UNLIKELY(!row_appender.add_row())) {
+ ret = E_OVERFLOW;
+ break;
+ } else if (RET_FAIL(time_decoder_->read_int64(time, time_in))) {
+ }
+ if (RET_FAIL(value_decoder_->read_int32(value, value_in))) {
+ } else if (filter != nullptr && !filter->satisfy(time, value)) {
+ row_appender.backoff_add_row();
+ continue;
+ } else {
+ /*std::cout << "decoder: time=" << time << ", value=" << value
+ * << std::endl;*/
+ row_appender.append(0, (char *)&time, sizeof(time));
+ row_appender.append(1, (char *)&value, sizeof(value));
+ }
+ }
+ } while (false);
+ return ret;
+}
+
+int AlignedChunkReader::decode_tv_buf_into_tsblock_by_datatype(
+ ByteStream &time_in, ByteStream &value_in, TsBlock *ret_tsblock,
+ Filter *filter) {
+ int ret = E_OK;
+ RowAppender row_appender(ret_tsblock);
+ switch (value_chunk_header_.data_type_) {
+ case common::BOOLEAN:
+ DECODE_TYPED_TV_INTO_TSBLOCK(bool, boolean, time_in_, value_in_,
+ row_appender);
+ break;
+ case common::INT32:
+ DECODE_TYPED_TV_INTO_TSBLOCK(int32_t, int32, time_in_, value_in_,
+ row_appender);
+ break;
+ case common::INT64:
+ DECODE_TYPED_TV_INTO_TSBLOCK(int64_t, int64, time_in_, value_in_,
+ row_appender);
+ break;
+ case common::FLOAT:
+ DECODE_TYPED_TV_INTO_TSBLOCK(float, float, time_in_, value_in_,
+ row_appender);
+ break;
+ case common::DOUBLE:
+ DECODE_TYPED_TV_INTO_TSBLOCK(double, double, time_in_, value_in_,
+ row_appender);
+ break;
+ default:
+ ret = E_NOT_SUPPORT;
+ ASSERT(false);
+ }
+ if (ret_tsblock->get_row_count() == 0 && ret == E_OK) {
+ ret = E_NO_MORE_DATA;
+ }
+ return ret;
+}
+
+} // end namespace storage
\ No newline at end of file
diff --git a/cpp/src/reader/aligned_chunk_reader.h b/cpp/src/reader/aligned_chunk_reader.h
new file mode 100644
index 00000000..c2d2d395
--- /dev/null
+++ b/cpp/src/reader/aligned_chunk_reader.h
@@ -0,0 +1,174 @@
+/*
+ * 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 READER_CHUNK_ALIGNED_READER_H
+#define READER_CHUNK_ALIGNED_READER_H
+
+#include "common/allocator/my_string.h"
+#include "common/tsfile_common.h"
+#include "compress/compressor.h"
+#include "encoding/decoder.h"
+#include "file/read_file.h"
+#include "reader/filter/filter.h"
+#include "reader/ichunk_reader.h"
+
+namespace storage {
+
+class AlignedChunkReader : public IChunkReader {
+ public:
+ AlignedChunkReader()
+ : read_file_(nullptr),
+ time_chunk_meta_(nullptr),
+ value_chunk_meta_(nullptr),
+ measurement_name_(),
+ time_chunk_header_(),
+ value_chunk_header_(),
+ cur_time_page_header_(),
+ cur_value_page_header_(),
+ time_in_stream_(),
+ value_in_stream_(),
+ file_data_time_buf_size_(0),
+ file_data_value_buf_size_(0),
+ time_chunk_visit_offset_(0),
+ value_chunk_visit_offset_(0),
+ time_compressor_(nullptr),
+ value_compressor_(nullptr),
+ time_filter_(nullptr),
+ time_decoder_(nullptr),
+ value_decoder_(nullptr),
+ time_in_(),
+ value_in_(),
+ time_uncompressed_buf_(nullptr),
+ value_uncompressed_buf_(nullptr),
+ cur_value_index(-1) {}
+ virtual int init(ReadFile *read_file, common::String m_name,
+ common::TSDataType data_type, Filter *time_filter);
+ virtual void reset();
+ virtual void destroy();
+
+ virtual bool has_more_data() const {
+ return prev_value_page_not_finish() ||
+ (value_chunk_visit_offset_ -
+ value_chunk_header_.serialized_size_ <
+ value_chunk_header_.data_size_);
+ }
+ virtual ChunkHeader &get_chunk_header() { return value_chunk_header_; }
+ virtual int load_by_aligned_meta(ChunkMeta *time_meta,
+ ChunkMeta *value_meta);
+
+ virtual int get_next_page(common::TsBlock *tsblock,
+ Filter *oneshoot_filter);
+
+ private:
+ FORCE_INLINE bool chunk_has_only_one_page(
+ const ChunkHeader &chunk_header) const {
+ return chunk_header.chunk_type_ == ONLY_ONE_PAGE_CHUNK_HEADER_MARKER;
+ }
+ int alloc_compressor_and_decoder(storage::Decoder* &decoder, storage::Compressor* &compressor,
+ common::TSEncoding encoding, common::TSDataType data_type,
+ common::CompressionType compression_type);
+ int get_cur_page_header(ChunkMeta *&chunk_meta,
+ common::ByteStream &in_stream_,
+ PageHeader &cur_page_header_,
+ uint32_t &chunk_visit_offset,
+ ChunkHeader &chunk_header);
+ int read_from_file_and_rewrap(common::ByteStream &in_stream_,
+ ChunkMeta *&chunk_meta,
+ uint32_t &chunk_visit_offset,
+ int32_t file_data_buf_size,
+ int want_size = 0);
+ bool cur_page_statisify_filter(Filter *filter);
+ int skip_cur_page();
+ int decode_cur_time_page_data();
+ int decode_cur_value_page_data();
+ int decode_time_value_buf_into_tsblock(common::TsBlock *&ret_tsblock,
+ Filter *filter);
+ int decode_tv_buf_into_tsblock(char *time_buf, char *value_buf,
+ uint32_t time_buf_size,
+ uint32_t value_buf_size,
+ common::TsBlock *ret_tsblock,
+ Filter *filter);
+ bool prev_time_page_not_finish() const {
+ return (time_decoder_ && time_decoder_->has_remaining()) ||
+ time_in_.has_remaining();
+ }
+
+ bool prev_value_page_not_finish() const {
+ return (value_decoder_ && value_decoder_->has_remaining()) ||
+ value_in_.has_remaining();
+ }
+
+ int decode_tv_buf_into_tsblock_by_datatype(common::ByteStream &time_in,
+ common::ByteStream &value_in,
+ common::TsBlock *ret_tsblock,
+ Filter *filter);
+ int i32_DECODE_TYPED_TV_INTO_TSBLOCK(common::ByteStream &time_in,
+ common::ByteStream &value_in,
+ common::RowAppender &row_appender,
+ Filter *filter);
+
+ private:
+ ReadFile *read_file_;
+ ChunkMeta *time_chunk_meta_;
+ ChunkMeta *value_chunk_meta_;
+ common::String measurement_name_;
+ ChunkHeader time_chunk_header_;
+ // TODO: support reading more than one measurement in AlignedChunkReader.
+ ChunkHeader value_chunk_header_;
+ PageHeader cur_time_page_header_;
+ PageHeader cur_value_page_header_;
+
+ /*
+ * Data reader from file is stored in @in_stream_, and the size
+ * is stored in @file_data_buf_size_. Note, in_stream_.total_size_
+ * is used to limit deserialization, that is why we still have
+ * @file_data_buf_size_.
+ *
+ * Since we may want keep data of current page (and page header
+ * of next page) in memory, we need a byte-size cursor to tell
+ * us which byte we are processing, so we have @chunk_visit_offset_
+ * it refer to position from the start of chunk_header_,
+ * also refer to offset within the chunk (including chunk header).
+ * It advanced by step of a page header or a page tv data.
+ */
+ common::ByteStream time_in_stream_;
+ common::ByteStream value_in_stream_;
+ int32_t file_data_time_buf_size_;
+ int32_t file_data_value_buf_size_;
+ uint32_t time_chunk_visit_offset_;
+ uint32_t value_chunk_visit_offset_;
+
+ // Statistic *page_statistic_;
+ Compressor *time_compressor_;
+ Compressor *value_compressor_;
+ Filter *time_filter_;
+
+ Decoder *time_decoder_;
+ Decoder *value_decoder_;
+ common::ByteStream time_in_;
+ common::ByteStream value_in_;
+ char *time_uncompressed_buf_;
+ char *value_uncompressed_buf_;
+ std::vector<uint8_t> value_page_col_notnull_bitmap_;
+ uint32_t value_page_data_num_;
+ int32_t cur_value_index;
+};
+
+} // end namespace storage
+#endif // READER_CHUNK_READER_H
diff --git a/cpp/src/reader/chunk_reader.cc b/cpp/src/reader/chunk_reader.cc
index d1b77818..701c8795 100644
--- a/cpp/src/reader/chunk_reader.cc
+++ b/cpp/src/reader/chunk_reader.cc
@@ -101,6 +101,7 @@ int ChunkReader::load_by_meta(ChunkMeta *meta) {
LOGE("file corrupted, ret=" << ret << ", offset="
<< chunk_meta_->offset_of_chunk_header_
<< "read_len=" << ret_read_len);
+ mem_free(file_data_buf);
}
/* ================ Step 2: deserialize chunk_header ================*/
@@ -127,12 +128,12 @@ int ChunkReader::load_by_meta(ChunkMeta *meta) {
int ChunkReader::alloc_compressor_and_value_decoder(
TSEncoding encoding, TSDataType data_type, CompressionType compression) {
if (value_decoder_ != nullptr) {
- delete value_decoder_;
- value_decoder_ = nullptr;
- }
- value_decoder_ = DecoderFactory::alloc_value_decoder(encoding, data_type);
- if (IS_NULL(value_decoder_)) {
- return E_OOM;
+ value_decoder_->reset();
+ } else {
+ value_decoder_ = DecoderFactory::alloc_value_decoder(encoding, data_type);
+ if (IS_NULL(value_decoder_)) {
+ return E_OOM;
+ }
}
if (compressor_ != nullptr) {
diff --git a/cpp/src/reader/chunk_reader.h b/cpp/src/reader/chunk_reader.h
index 83548a5c..a53b6e1c 100644
--- a/cpp/src/reader/chunk_reader.h
+++ b/cpp/src/reader/chunk_reader.h
@@ -26,10 +26,11 @@
#include "encoding/decoder.h"
#include "file/read_file.h"
#include "reader/filter/filter.h"
+#include "reader/ichunk_reader.h"
namespace storage {
-class ChunkReader {
+class ChunkReader : public IChunkReader {
public:
ChunkReader()
: read_file_(nullptr),
@@ -47,25 +48,26 @@ class ChunkReader {
time_in_(),
value_in_(),
uncompressed_buf_(nullptr) {}
- int init(ReadFile *read_file, common::String m_name,
- common::TSDataType data_type, Filter *time_filter);
- void reset();
- void destroy();
+ virtual int init(ReadFile *read_file, common::String m_name,
+ common::TSDataType data_type, Filter *time_filter);
+ virtual void reset();
+ virtual void destroy();
- bool has_more_data() const {
+ virtual bool has_more_data() const {
return prev_page_not_finish() ||
(chunk_visit_offset_ - chunk_header_.serialized_size_ <
chunk_header_.data_size_);
}
- ChunkHeader &get_chunk_header() { return chunk_header_; }
+ virtual ChunkHeader &get_chunk_header() { return chunk_header_; }
/*
* prepare data buffer, load the chunk_header
* and the first page_header.
*/
- int load_by_meta(ChunkMeta *meta);
+ virtual int load_by_meta(ChunkMeta *meta);
- int get_next_page(common::TsBlock *tsblock, Filter *oneshoot_filter);
+ virtual int get_next_page(common::TsBlock *tsblock,
+ Filter *oneshoot_filter);
private:
FORCE_INLINE bool chunk_has_only_one_page() const {
diff --git a/cpp/src/reader/ichunk_reader.h b/cpp/src/reader/ichunk_reader.h
new file mode 100644
index 00000000..8f2bd7ee
--- /dev/null
+++ b/cpp/src/reader/ichunk_reader.h
@@ -0,0 +1,62 @@
+/*
+ * 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 READER_ICHUNK_READER_H
+#define READER_ICHUNK_READER_H
+
+#include "common/allocator/my_string.h"
+#include "common/tsfile_common.h"
+#include "compress/compressor.h"
+#include "encoding/decoder.h"
+#include "file/read_file.h"
+#include "reader/filter/filter.h"
+
+namespace storage {
+
+class IChunkReader {
+ public:
+ IChunkReader() {}
+ virtual int init(ReadFile *read_file, common::String m_name,
+ common::TSDataType data_type, Filter *time_filter) {
+ return common::E_OK;
+ }
+ virtual void reset() {}
+ virtual void destroy() {}
+
+ virtual bool has_more_data() const { return false; }
+
+ virtual int load_by_meta(ChunkMeta *meta) { return common::E_INVALID_ARG; }
+ virtual int load_by_aligned_meta(ChunkMeta *time_meta,
+ ChunkMeta *value_meta) {
+ return common::E_INVALID_ARG;
+ }
+
+ virtual int get_next_page(common::TsBlock *tsblock,
+ Filter *oneshoot_filter) {
+ return common::E_OK;
+ }
+
+ virtual ChunkHeader &get_chunk_header() { return chunk_header_; }
+
+ protected:
+ ChunkHeader chunk_header_;
+};
+
+} // end namespace storage
+#endif // READER_ICHUNK_READER_H
diff --git a/cpp/src/reader/tsfile_series_scan_iterator.cc b/cpp/src/reader/tsfile_series_scan_iterator.cc
index 50379a5b..5cc6a87a 100644
--- a/cpp/src/reader/tsfile_series_scan_iterator.cc
+++ b/cpp/src/reader/tsfile_series_scan_iterator.cc
@@ -25,7 +25,11 @@ namespace storage {
void TsFileSeriesScanIterator::destroy() {
timeseries_index_pa_.destroy();
- chunk_reader_.destroy();
+ if (chunk_reader_ != nullptr) {
+ chunk_reader_->destroy();
+ common::mem_free(chunk_reader_);
+ chunk_reader_ = nullptr;
+ }
if (tsblock_ != nullptr) {
delete tsblock_;
tsblock_ = nullptr;
@@ -38,19 +42,35 @@ int TsFileSeriesScanIterator::get_next(TsBlock *&ret_tsblock, bool alloc,
int ret = E_OK;
Filter *filter =
(oneshoot_filter != nullptr) ? oneshoot_filter : time_filter_;
- if (!chunk_reader_.has_more_data()) {
+ if (!chunk_reader_->has_more_data()) {
while (true) {
if (!has_next_chunk()) {
return E_NO_MORE_DATA;
} else {
- advance_to_next_chunk();
- ChunkMeta *cm = get_current_chunk_meta();
- if (filter != nullptr && cm->statistic_ != nullptr &&
- !filter->satisfy(cm->statistic_)) {
- continue;
- }
- chunk_reader_.reset();
- if (RET_FAIL(chunk_reader_.load_by_meta(cm))) {
+ if (!is_aligned_) {
+ ChunkMeta *cm = get_current_chunk_meta();
+ advance_to_next_chunk();
+ if (filter != nullptr && cm->statistic_ != nullptr &&
+ !filter->satisfy(cm->statistic_)) {
+ continue;
+ }
+ chunk_reader_->reset();
+ if (RET_FAIL(chunk_reader_->load_by_meta(cm))) {
+ }
+ break;
+ } else {
+ ChunkMeta *value_cm = value_chunk_meta_cursor_.get();
+ ChunkMeta *time_cm = time_chunk_meta_cursor_.get();
+ advance_to_next_chunk();
+ if (filter != nullptr && value_cm->statistic_ != nullptr &&
+ !filter->satisfy(value_cm->statistic_)) {
+ continue;
+ }
+ chunk_reader_->reset();
+ if (RET_FAIL(chunk_reader_->load_by_aligned_meta(
+ time_cm, value_cm))) {
+ }
+ break;
}
}
}
@@ -59,7 +79,7 @@ int TsFileSeriesScanIterator::get_next(TsBlock *&ret_tsblock, bool alloc,
if (alloc) {
ret_tsblock = alloc_tsblock();
}
- ret = chunk_reader_.get_next_page(ret_tsblock, oneshoot_filter);
+ ret = chunk_reader_->get_next_page(ret_tsblock, oneshoot_filter);
}
return ret;
}
@@ -74,21 +94,48 @@ void TsFileSeriesScanIterator::revert_tsblock() {
int TsFileSeriesScanIterator::init_chunk_reader() {
int ret = E_OK;
- chunk_meta_cursor_ = timeseries_index_.get_chunk_meta_list()->begin();
- ChunkMeta *cm = chunk_meta_cursor_.get();
- ASSERT(!chunk_reader_.has_more_data());
- if (RET_FAIL(chunk_reader_.init(
- read_file_, timeseries_index_.get_measurement_name(),
- timeseries_index_.get_data_type(), time_filter_))) {
- } else if (RET_FAIL(chunk_reader_.load_by_meta(cm))) {
+ is_aligned_ = itimeseries_index_->get_data_type() == common::VECTOR;
+ if (!is_aligned_) {
+ void *buf = common::mem_alloc(sizeof(ChunkReader), common::MOD_DEFAULT);
+ chunk_reader_ = new (buf) ChunkReader;
+ chunk_meta_cursor_ = itimeseries_index_->get_chunk_meta_list()->begin();
+ ChunkMeta *cm = chunk_meta_cursor_.get();
+ ASSERT(!chunk_reader_->has_more_data());
+ if (RET_FAIL(chunk_reader_->init(
+ read_file_, itimeseries_index_->get_measurement_name(),
+ itimeseries_index_->get_data_type(), time_filter_))) {
+ } else if (RET_FAIL(chunk_reader_->load_by_meta(cm))) {
+ } else {
+ chunk_meta_cursor_++;
+ }
} else {
- chunk_meta_cursor_++;
+ void *buf =
+ common::mem_alloc(sizeof(AlignedChunkReader), common::MOD_DEFAULT);
+ chunk_reader_ = new (buf) AlignedChunkReader;
+ time_chunk_meta_cursor_ =
+ itimeseries_index_->get_time_chunk_meta_list()->begin();
+ value_chunk_meta_cursor_ =
+ itimeseries_index_->get_value_chunk_meta_list()->begin();
+ ChunkMeta *time_cm = time_chunk_meta_cursor_.get();
+ ChunkMeta *value_cm = value_chunk_meta_cursor_.get();
+ chunk_reader_->has_more_data();
+ ASSERT(!chunk_reader_->has_more_data());
+ if (RET_FAIL(chunk_reader_->init(
+ read_file_, itimeseries_index_->get_measurement_name(),
+ itimeseries_index_->get_data_type(), time_filter_))) {
+ } else if (RET_FAIL(chunk_reader_->load_by_aligned_meta(time_cm,
+ value_cm))) {
+ } else {
+ time_chunk_meta_cursor_++;
+ value_chunk_meta_cursor_++;
+ }
}
+
return ret;
}
TsBlock *TsFileSeriesScanIterator::alloc_tsblock() {
- ChunkHeader &ch = chunk_reader_.get_chunk_header();
+ ChunkHeader &ch = chunk_reader_->get_chunk_header();
TsID dummy_ts_id;
// TODO config
diff --git a/cpp/src/reader/tsfile_series_scan_iterator.h b/cpp/src/reader/tsfile_series_scan_iterator.h
index fdc57b77..3896b7ad 100644
--- a/cpp/src/reader/tsfile_series_scan_iterator.h
+++ b/cpp/src/reader/tsfile_series_scan_iterator.h
@@ -22,6 +22,7 @@
#include <string>
+#include "aligned_chunk_reader.h"
#include "common/tsblock/tsblock.h"
#include "file/read_file.h"
#include "file/tsfile_io_reader.h"
@@ -39,13 +40,14 @@ class TsFileSeriesScanIterator {
: read_file_(nullptr),
device_path_(),
measurement_name_(),
- timeseries_index_(),
+ itimeseries_index_(),
timeseries_index_pa_(),
chunk_meta_cursor_(),
- chunk_reader_(),
+ chunk_reader_(nullptr),
tuple_desc_(),
tsblock_(nullptr),
- time_filter_(nullptr) {}
+ time_filter_(nullptr),
+ is_aligned_(false) {}
~TsFileSeriesScanIterator() { destroy(); }
int init(const std::string &device_path,
const std::string &measurement_name, ReadFile *read_file,
@@ -70,10 +72,22 @@ class TsFileSeriesScanIterator {
private:
int init_chunk_reader();
FORCE_INLINE bool has_next_chunk() const {
- return chunk_meta_cursor_ !=
- timeseries_index_.get_chunk_meta_list()->end();
+ if (is_aligned_) {
+ return value_chunk_meta_cursor_ !=
+ itimeseries_index_->get_value_chunk_meta_list()->end();
+ } else {
+ return chunk_meta_cursor_ !=
+ itimeseries_index_->get_chunk_meta_list()->end();
+ }
+ }
+ FORCE_INLINE void advance_to_next_chunk() {
+ if (is_aligned_) {
+ time_chunk_meta_cursor_++;
+ value_chunk_meta_cursor_++;
+ } else {
+ chunk_meta_cursor_++;
+ }
}
- FORCE_INLINE void advance_to_next_chunk() { chunk_meta_cursor_++; }
FORCE_INLINE ChunkMeta *get_current_chunk_meta() {
return chunk_meta_cursor_.get();
}
@@ -84,14 +98,17 @@ class TsFileSeriesScanIterator {
std::string device_path_;
std::string measurement_name_;
- TimeseriesIndex timeseries_index_;
+ ITimeseriesIndex *itimeseries_index_;
common::PageArena timeseries_index_pa_;
common::SimpleList<ChunkMeta *>::Iterator chunk_meta_cursor_;
- ChunkReader chunk_reader_;
+ common::SimpleList<ChunkMeta *>::Iterator time_chunk_meta_cursor_;
+ common::SimpleList<ChunkMeta *>::Iterator value_chunk_meta_cursor_;
+ IChunkReader *chunk_reader_;
common::TupleDesc tuple_desc_;
common::TsBlock *tsblock_;
Filter *time_filter_;
+ bool is_aligned_ = false;
};
} // end namespace storage
diff --git a/cpp/src/writer/chunk_writer.cc b/cpp/src/writer/chunk_writer.cc
index a283fb96..c08e725d 100644
--- a/cpp/src/writer/chunk_writer.cc
+++ b/cpp/src/writer/chunk_writer.cc
@@ -152,12 +152,10 @@ int ChunkWriter::end_encode_chunk() {
return ret;
}
-
-int64_t ChunkWriter::estimate_max_series_mem_size(){
- return chunk_data_.total_size()
- + page_writer_.estimate_max_mem_size()
- + PageHeader::estimat_max_page_header_size_without_statistics()
- + get_typed_statistic_sizeof(page_writer_.get_statistic()->get_type());
+int64_t ChunkWriter::estimate_max_series_mem_size() {
+ return chunk_data_.total_size() + page_writer_.estimate_max_mem_size() +
+ PageHeader::estimat_max_page_header_size_without_statistics() +
+ get_typed_statistic_sizeof(page_writer_.get_statistic()->get_type());
}
} // end namespace storage
diff --git a/cpp/src/writer/page_writer.cc b/cpp/src/writer/page_writer.cc
index 539f6069..983818e4 100644
--- a/cpp/src/writer/page_writer.cc
+++ b/cpp/src/writer/page_writer.cc
@@ -47,9 +47,10 @@ int PageData::init(ByteStream &time_bs, ByteStream &value_bs,
}
if (RET_FAIL(SerializationUtil::write_var_uint(
time_buf_size_, uncompressed_buf_, var_size))) {
- } else if (RET_FAIL(copy_bs_to_buf(time_bs, uncompressed_buf_ + var_size,
- uncompressed_size_ - var_size))) {
- } else if (RET_FAIL(copy_bs_to_buf(
+ } else if (RET_FAIL(
+ common::copy_bs_to_buf(time_bs, uncompressed_buf_ + var_size,
+ uncompressed_size_ - var_size))) {
+ } else if (RET_FAIL(common::copy_bs_to_buf(
value_bs, uncompressed_buf_ + var_size + time_buf_size_,
uncompressed_size_ - var_size - time_buf_size_))) {
} else {
@@ -72,26 +73,6 @@ int PageData::init(ByteStream &time_bs, ByteStream &value_bs,
return ret;
}
-int PageData::copy_bs_to_buf(ByteStream &bs, char *src_buf,
- uint32_t src_buf_len) {
- ByteStream::BufferIterator buf_iter = bs.init_buffer_iterator();
- uint32_t copyed_len = 0;
- while (true) {
- ByteStream::Buffer buf = buf_iter.get_next_buf();
- if (buf.buf_ == nullptr) {
- break;
- } else {
- if (src_buf_len - copyed_len < buf.len_) {
- ASSERT(false);
- return E_BUF_NOT_ENOUGH;
- }
- memcpy(src_buf + copyed_len, buf.buf_, buf.len_);
- copyed_len += buf.len_;
- }
- }
- return E_OK;
-}
-
/* ================ PageWriter ================ */
int PageWriter::init(TSDataType data_type, TSEncoding encoding,
CompressionType compression) {
@@ -147,7 +128,6 @@ void PageWriter::destroy() {
EncoderFactory::free(time_encoder_);
EncoderFactory::free(value_encoder_);
StatisticFactory::free(statistic_);
- compressor_->destroy();
CompressorFactory::free(compressor_);
}
}
diff --git a/cpp/src/writer/page_writer.h b/cpp/src/writer/page_writer.h
index 6800c7ea..e348cfad 100644
--- a/cpp/src/writer/page_writer.h
+++ b/cpp/src/writer/page_writer.h
@@ -70,7 +70,6 @@ struct PageData {
compressed_buf_ = nullptr;
}
}
- int copy_bs_to_buf(common::ByteStream &bs, char *buf, uint32_t buf_len);
};
/* ================ PageWriter ================ */
diff --git a/cpp/src/writer/chunk_writer.cc b/cpp/src/writer/time_chunk_writer.cc
similarity index 60%
copy from cpp/src/writer/chunk_writer.cc
copy to cpp/src/writer/time_chunk_writer.cc
index a283fb96..f8298f93 100644
--- a/cpp/src/writer/chunk_writer.cc
+++ b/cpp/src/writer/time_chunk_writer.cc
@@ -17,7 +17,7 @@
* under the License.
*/
-#include "chunk_writer.h"
+#include "time_chunk_writer.h"
#include "common/logger/elog.h"
@@ -25,34 +25,34 @@ using namespace common;
namespace storage {
-int ChunkWriter::init(const ColumnDesc &col_desc) {
- return init(col_desc.column_name_, col_desc.type_, col_desc.encoding_,
+int TimeChunkWriter::init(const ColumnDesc &col_desc) {
+ return init(col_desc.column_name_, col_desc.encoding_,
col_desc.compression_);
}
-int ChunkWriter::init(const std::string &measurement_name, TSDataType data_type,
- TSEncoding encoding, CompressionType compression_type) {
+int TimeChunkWriter::init(const std::string &measurement_name,
+ TSEncoding encoding,
+ CompressionType compression_type) {
int ret = E_OK;
- chunk_statistic_ = StatisticFactory::alloc_statistic(data_type);
+ chunk_statistic_ = StatisticFactory::alloc_statistic(common::VECTOR);
if (chunk_statistic_ == nullptr) {
return E_OOM;
- } else if (RET_FAIL(
- page_writer_.init(data_type, encoding, compression_type))) {
- } else if (IS_NULL((first_page_statistic_ =
- StatisticFactory::alloc_statistic(data_type)))) {
+ } else if (RET_FAIL(time_page_writer_.init(encoding, compression_type))) {
+ } else if (IS_NULL(
+ (first_page_statistic_ =
+ StatisticFactory::alloc_statistic(common::VECTOR)))) {
ret = E_OOM;
} else {
- data_type_ = data_type;
chunk_header_.measurement_name_ = measurement_name;
- chunk_header_.data_type_ = data_type;
+ chunk_header_.data_type_ = common::VECTOR;
chunk_header_.compression_type_ = compression_type;
chunk_header_.encoding_type_ = encoding;
}
return ret;
}
-void ChunkWriter::destroy() {
- page_writer_.destroy();
+void TimeChunkWriter::destroy() {
+ time_page_writer_.destroy();
if (chunk_statistic_ != nullptr) {
StatisticFactory::free(chunk_statistic_);
chunk_statistic_ = nullptr;
@@ -66,30 +66,33 @@ void ChunkWriter::destroy() {
num_of_pages_ = 0;
}
-int ChunkWriter::seal_cur_page(bool end_chunk) {
+int TimeChunkWriter::seal_cur_page(bool end_chunk) {
int ret = E_OK;
- if (RET_FAIL(chunk_statistic_->merge_with(page_writer_.get_statistic()))) {
+ if (RET_FAIL(
+ chunk_statistic_->merge_with(time_page_writer_.get_statistic()))) {
return ret;
}
if (num_of_pages_ == 0) {
if (end_chunk) {
// this page is the only one page of this chunk
- ret = page_writer_.write_to_chunk(chunk_data_, /*header*/ true,
- /*stat*/ false, /*data*/ true);
- page_writer_.destroy_page_data();
- page_writer_.destroy();
+ ret =
+ time_page_writer_.write_to_chunk(chunk_data_, /*header*/ true,
+ /*stat*/ false, /*data*/ true);
+ time_page_writer_.destroy_page_data();
+ time_page_writer_.destroy();
} else {
/*
* if the chunk has only one page, do not writer page statistic.
* so we save the data of first page and see if the chunk has more
* page later.
*/
- ret = page_writer_.write_to_chunk(chunk_data_, /*header*/ true,
- /*stat*/ false, /*data*/ false);
+ ret = time_page_writer_.write_to_chunk(chunk_data_, /*header*/ true,
+ /*stat*/ false,
+ /*data*/ false);
if (IS_SUCC(ret)) {
- save_first_page_data(page_writer_);
- // page_writer_.destroy_page_data();
- page_writer_.reset();
+ save_first_page_data(time_page_writer_);
+ // time_page_writer_.destroy_page_data();
+ time_page_writer_.reset();
}
}
} else {
@@ -100,12 +103,12 @@ int ChunkWriter::seal_cur_page(bool end_chunk) {
free_first_writer_data();
}
if (IS_SUCC(ret)) {
- if (RET_FAIL(page_writer_.write_to_chunk(
+ if (RET_FAIL(time_page_writer_.write_to_chunk(
chunk_data_, /*header*/ true, /*stat*/ true,
/*data*/ true))) {
}
- page_writer_.destroy_page_data();
- page_writer_.reset();
+ time_page_writer_.destroy_page_data();
+ time_page_writer_.reset();
}
}
num_of_pages_++;
@@ -119,12 +122,12 @@ int ChunkWriter::seal_cur_page(bool end_chunk) {
return ret;
}
-void ChunkWriter::save_first_page_data(PageWriter &first_page_writer) {
+void TimeChunkWriter::save_first_page_data(TimePageWriter &first_page_writer) {
first_page_data_ = first_page_writer.get_cur_page_data();
first_page_statistic_->deep_copy_from(first_page_writer.get_statistic());
}
-int ChunkWriter::write_first_page_data(ByteStream &pages_data) {
+int TimeChunkWriter::write_first_page_data(ByteStream &pages_data) {
int ret = E_OK;
if (RET_FAIL(first_page_statistic_->serialize_to(pages_data))) {
} else if (RET_FAIL(
@@ -134,9 +137,9 @@ int ChunkWriter::write_first_page_data(ByteStream &pages_data) {
return ret;
}
-int ChunkWriter::end_encode_chunk() {
+int TimeChunkWriter::end_encode_chunk() {
int ret = E_OK;
- if (page_writer_.get_statistic()->count_ > 0) {
+ if (time_page_writer_.get_statistic()->count_ > 0) {
ret = seal_cur_page(/*end_chunk*/ true);
if (E_OK == ret) {
chunk_header_.data_size_ = chunk_data_.total_size();
@@ -144,20 +147,20 @@ int ChunkWriter::end_encode_chunk() {
}
}
#if DEBUG_SE
- std::cout << "end_encode_chunk: num_of_pages_=" << num_of_pages_
+ std::cout << "end_encode_time_chunk: num_of_pages_=" << num_of_pages_
<< ", chunk_header_.data_size_=" << chunk_header_.data_size_
<< ", page_writer.get_statistic()->count_="
- << page_writer_.get_statistic()->count_ << std::endl;
+ << time_page_writer_.get_statistic()->count_ << std::endl;
#endif
return ret;
}
-
-int64_t ChunkWriter::estimate_max_series_mem_size(){
- return chunk_data_.total_size()
- + page_writer_.estimate_max_mem_size()
- + PageHeader::estimat_max_page_header_size_without_statistics()
- + get_typed_statistic_sizeof(page_writer_.get_statistic()->get_type());
+int64_t TimeChunkWriter::estimate_max_series_mem_size() {
+ return chunk_data_.total_size() +
+ time_page_writer_.estimate_max_mem_size() +
+ PageHeader::estimat_max_page_header_size_without_statistics() +
+ get_typed_statistic_sizeof(
+ time_page_writer_.get_statistic()->get_type());
}
} // end namespace storage
diff --git a/cpp/src/writer/time_chunk_writer.h b/cpp/src/writer/time_chunk_writer.h
new file mode 100644
index 00000000..cac3169f
--- /dev/null
+++ b/cpp/src/writer/time_chunk_writer.h
@@ -0,0 +1,109 @@
+/*
+ * 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 WRITER_CHUNK_TIME_WRITER_H
+#define WRITER_CHUNK_TIME_WRITER_H
+
+#include "common/allocator/byte_stream.h"
+#include "common/global.h"
+#include "common/statistic.h"
+#include "common/tsfile_common.h"
+#include "time_page_writer.h"
+
+namespace storage {
+
+class TimeChunkWriter {
+ public:
+ static const int32_t PAGES_DATA_PAGE_SIZE = 1024;
+
+ public:
+ TimeChunkWriter()
+ : time_page_writer_(),
+ chunk_statistic_(nullptr),
+ chunk_data_(PAGES_DATA_PAGE_SIZE, common::MOD_CW_PAGES_DATA),
+ first_page_data_(),
+ first_page_statistic_(nullptr),
+ chunk_header_(),
+ num_of_pages_(0) {}
+ ~TimeChunkWriter() { destroy(); }
+ int init(const common::ColumnDesc &col_desc);
+ int init(const std::string &measurement_name, common::TSEncoding encoding,
+ common::CompressionType compression_type);
+ void destroy();
+
+ storage::ChunkHeader get_chunk_header() const { return chunk_header_; }
+
+ FORCE_INLINE int write(int64_t timestamp) {
+ int ret = common::E_OK;
+ if (RET_FAIL(time_page_writer_.write(timestamp))) {
+ return ret;
+ }
+ if (RET_FAIL(seal_cur_page_if_full())) {
+ return ret;
+ }
+ return ret;
+ }
+
+ int end_encode_chunk();
+ common::ByteStream &get_chunk_data() { return chunk_data_; }
+ Statistic *get_chunk_statistic() { return chunk_statistic_; }
+ FORCE_INLINE int32_t num_of_pages() const { return num_of_pages_; }
+
+ int64_t estimate_max_series_mem_size();
+
+ private:
+ FORCE_INLINE bool is_cur_page_full() const {
+ // FIXME
+ return (time_page_writer_.get_point_numer() >=
+ common::g_config_value_.page_writer_max_point_num_) ||
+ (time_page_writer_.get_page_memory_size() >=
+ common::g_config_value_.page_writer_max_memory_bytes_);
+ }
+ FORCE_INLINE int seal_cur_page_if_full() {
+ if (UNLIKELY(is_cur_page_full())) {
+ return seal_cur_page(false);
+ }
+ return common::E_OK;
+ }
+ FORCE_INLINE void free_first_writer_data() {
+ // free memory
+ first_page_data_.destroy();
+ StatisticFactory::free(first_page_statistic_);
+ first_page_statistic_ = nullptr;
+ }
+ int seal_cur_page(bool end_chunk);
+ void save_first_page_data(TimePageWriter &first_time_page_writer);
+ int write_first_page_data(common::ByteStream &pages_data);
+
+ private:
+ TimePageWriter time_page_writer_;
+ Statistic *chunk_statistic_;
+ common::ByteStream chunk_data_;
+
+ // to save first page data
+ TimePageData first_page_data_;
+ Statistic *first_page_statistic_;
+
+ ChunkHeader chunk_header_;
+ int32_t num_of_pages_;
+};
+
+} // end namespace storage
+
+#endif // WRITER_CHUNK_TIME_WRITER_H
diff --git a/cpp/src/writer/page_writer.cc b/cpp/src/writer/time_page_writer.cc
similarity index 58%
copy from cpp/src/writer/page_writer.cc
copy to cpp/src/writer/time_page_writer.cc
index 539f6069..215b0b76 100644
--- a/cpp/src/writer/page_writer.cc
+++ b/cpp/src/writer/time_page_writer.cc
@@ -17,7 +17,7 @@
* under the License.
*/
-#include "page_writer.h"
+#include "time_page_writer.h"
#include "common/config/config.h"
#include "common/logger/elog.h"
@@ -28,30 +28,25 @@ using namespace common;
namespace storage {
-/* ================ PageData ================ */
-int PageData::init(ByteStream &time_bs, ByteStream &value_bs,
- Compressor *compressor) {
+int TimePageData::init(ByteStream &time_bs, Compressor *compressor) {
int ret = E_OK;
time_buf_size_ = time_bs.total_size();
- value_buf_size_ = value_bs.total_size();
uint32_t var_size = get_var_uint_size(time_buf_size_);
- uncompressed_size_ = var_size + time_buf_size_ + value_buf_size_;
+ uncompressed_size_ = var_size + time_buf_size_;
uncompressed_buf_ =
(char *)mem_alloc(uncompressed_size_, MOD_PAGE_WRITER_OUTPUT_STREAM);
compressor_ = compressor;
if (IS_NULL(uncompressed_buf_)) {
return E_OOM;
}
- if (time_buf_size_ == 0 || value_buf_size_ == 0) {
+ if (time_buf_size_ == 0) {
return E_INVALID_ARG;
}
if (RET_FAIL(SerializationUtil::write_var_uint(
time_buf_size_, uncompressed_buf_, var_size))) {
- } else if (RET_FAIL(copy_bs_to_buf(time_bs, uncompressed_buf_ + var_size,
- uncompressed_size_ - var_size))) {
- } else if (RET_FAIL(copy_bs_to_buf(
- value_bs, uncompressed_buf_ + var_size + time_buf_size_,
- uncompressed_size_ - var_size - time_buf_size_))) {
+ } else if (RET_FAIL(
+ common::copy_bs_to_buf(time_bs, uncompressed_buf_ + var_size,
+ uncompressed_size_ - var_size))) {
} else {
// TODO
// NOTE: different compressor may have different compress API
@@ -63,47 +58,22 @@ int PageData::init(ByteStream &time_bs, ByteStream &value_bs,
}
}
#if DEBUG_SE
- std::cout << "PageData::init. time_buf_size=" << time_buf_size_
- << ", value_buf_size=" << value_buf_size_
+ std::cout << "TimePageData::init. time_buf_size=" << time_buf_size_
<< ", uncompressed_size=" << uncompressed_size_
<< ", compressed_size=" << compressed_size_ << std::endl;
- // DEBUG_hex_dump_buf("compressed_buf=", compressed_buf_, compressed_size_);
+ DEBUG_hex_dump_buf("uncompressed_buf=", uncompressed_buf_,
+ uncompressed_size_);
+ DEBUG_hex_dump_buf("compressed_buf=", compressed_buf_, compressed_size_);
#endif
return ret;
}
-int PageData::copy_bs_to_buf(ByteStream &bs, char *src_buf,
- uint32_t src_buf_len) {
- ByteStream::BufferIterator buf_iter = bs.init_buffer_iterator();
- uint32_t copyed_len = 0;
- while (true) {
- ByteStream::Buffer buf = buf_iter.get_next_buf();
- if (buf.buf_ == nullptr) {
- break;
- } else {
- if (src_buf_len - copyed_len < buf.len_) {
- ASSERT(false);
- return E_BUF_NOT_ENOUGH;
- }
- memcpy(src_buf + copyed_len, buf.buf_, buf.len_);
- copyed_len += buf.len_;
- }
- }
- return E_OK;
-}
-
-/* ================ PageWriter ================ */
-int PageWriter::init(TSDataType data_type, TSEncoding encoding,
- CompressionType compression) {
+int TimePageWriter::init(TSEncoding encoding, CompressionType compression) {
int ret = E_OK;
- data_type_ = data_type;
if (nullptr == (time_encoder_ = EncoderFactory::alloc_time_encoder())) {
ret = E_OOM;
- } else if (nullptr == (value_encoder_ = EncoderFactory::alloc_value_encoder(
- encoding, data_type))) {
- ret = E_OOM;
- } else if (nullptr ==
- (statistic_ = StatisticFactory::alloc_statistic(data_type))) {
+ } else if (nullptr == (statistic_ = StatisticFactory::alloc_statistic(
+ common::VECTOR))) {
ret = E_OOM;
} else if (nullptr == (compressor_ = CompressorFactory::alloc_compressor(
compression))) {
@@ -113,9 +83,6 @@ int PageWriter::init(TSDataType data_type, TSEncoding encoding,
if (time_encoder_ != nullptr) {
EncoderFactory::free(time_encoder_);
}
- if (value_encoder_ != nullptr) {
- EncoderFactory::free(value_encoder_);
- }
if (statistic_ != nullptr) {
StatisticFactory::free(statistic_);
}
@@ -126,45 +93,36 @@ int PageWriter::init(TSDataType data_type, TSEncoding encoding,
return ret;
}
-/*
- * free out_stream memory, reset statistic_,
- */
-void PageWriter::reset() {
+void TimePageWriter::reset() {
time_encoder_->reset();
- value_encoder_->reset();
statistic_->reset();
time_out_stream_.reset();
- value_out_stream_.reset();
}
-void PageWriter::destroy() {
+void TimePageWriter::destroy() {
if (is_inited_) {
is_inited_ = false;
time_encoder_->destroy();
- value_encoder_->destroy();
statistic_->destroy();
EncoderFactory::free(time_encoder_);
- EncoderFactory::free(value_encoder_);
StatisticFactory::free(statistic_);
- compressor_->destroy();
CompressorFactory::free(compressor_);
}
}
-int PageWriter::write_to_chunk(ByteStream &pages_data, bool write_header,
- bool write_statistic,
- bool write_data_to_chunk_data) {
+int TimePageWriter::write_to_chunk(ByteStream &pages_data, bool write_header,
+ bool write_statistic,
+ bool write_data_to_chunk_data) {
#if DEBUG_SE
- std::cout << "PageWriter::write_to_chunk at position "
+ std::cout << "TimePageWriter::write_to_chunk at position "
<< pages_data.total_size() << " of chunk_data." << std::endl;
#endif
int ret = E_OK;
if (RET_FAIL(prepare_end_page())) {
return ret;
}
- if (RET_FAIL(cur_page_data_.init(time_out_stream_, value_out_stream_,
- compressor_))) {
+ if (RET_FAIL(cur_page_data_.init(time_out_stream_, compressor_))) {
}
if (IS_SUCC(ret) && write_header) {
@@ -174,15 +132,15 @@ int PageWriter::write_to_chunk(ByteStream &pages_data, bool write_header,
cur_page_data_.compressed_size_, pages_data))) {
}
}
- // std::cout << "PageWriter::write_to_chunk after write_header. pos=" <<
+ // std::cout << "TimePageWriter::write_to_chunk after write_header. pos=" <<
// pages_data.total_size() << std::endl;
if (IS_SUCC(ret) && write_statistic) {
if (RET_FAIL(statistic_->serialize_to(pages_data))) {
}
}
- // std::cout << "PageWriter::write_to_chunk after write_stat. pos=" <<
+ // std::cout << "TimePageWriter::write_to_chunk after write_stat. pos=" <<
// pages_data.total_size() << std::endl; DEBUG_print_byte_stream("In
- // PageWriter::write_to_chunk, before writer page data: pages_data = ",
+ // TimePageWriter::write_to_chunk, before writer page data: pages_data = ",
// pages_data);
if (IS_SUCC(ret) && write_data_to_chunk_data) {
// DEBUG_hex_dump_buf("cur_page_data_.compressed_buf_ = ",
@@ -191,12 +149,12 @@ int PageWriter::write_to_chunk(ByteStream &pages_data, bool write_header,
cur_page_data_.compressed_size_))) {
}
}
- // DEBUG_print_byte_stream("In PageWriter::write_to_chunk, after writer page
- // data: pages_data = ", pages_data); std::cout <<
- // "PageWriter::write_to_chunk after write_data. pos=" <<
+ // DEBUG_print_byte_stream("In TimePageWriter::write_to_chunk, after writer
+ // page data: pages_data = ", pages_data); std::cout <<
+ // "TimePageWriter::write_to_chunk after write_data. pos=" <<
// pages_data.total_size() << std::endl;
#if DEBUG_SE
- std::cout << "PageWriter write_to_chunk: ret=" << ret
+ std::cout << "TimePageWriter write_to_chunk: ret=" << ret
<< ", flags(HSD)=" << write_header << write_statistic
<< write_data_to_chunk_data
<< ", uncompressed_size=" << cur_page_data_.uncompressed_size_
diff --git a/cpp/src/writer/time_page_writer.h b/cpp/src/writer/time_page_writer.h
new file mode 100644
index 00000000..780398c4
--- /dev/null
+++ b/cpp/src/writer/time_page_writer.h
@@ -0,0 +1,127 @@
+/*
+ * 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 WRITER_PAGE_TIME_WRITER_H
+#define WRITER_PAGE_TIME_WRITER_H
+
+#include "common/allocator/byte_stream.h"
+#include "common/statistic.h"
+#include "compress/compressor.h"
+#include "encoding/encoder.h"
+#include "utils/db_utils.h"
+
+namespace storage {
+
+struct TimePageData {
+ uint32_t time_buf_size_;
+ uint32_t uncompressed_size_;
+ uint32_t compressed_size_;
+ char *uncompressed_buf_;
+ char *compressed_buf_;
+ Compressor *compressor_;
+
+ TimePageData()
+ : time_buf_size_(0),
+ uncompressed_size_(0),
+ compressed_size_(0),
+ uncompressed_buf_(nullptr),
+ compressed_buf_(nullptr),
+ compressor_(nullptr) {}
+ int init(common::ByteStream &time_bs, Compressor *compressor);
+ void destroy() {
+ if (uncompressed_buf_ != nullptr) {
+ common::mem_free(uncompressed_buf_);
+ uncompressed_buf_ = nullptr;
+ }
+ if (compressed_buf_ != nullptr && compressor_ != nullptr) {
+ compressor_->after_compress(compressed_buf_);
+ compressed_buf_ = nullptr;
+ }
+ }
+};
+
+class TimePageWriter {
+ public:
+ TimePageWriter()
+ : data_type_(common::VECTOR),
+ time_encoder_(nullptr),
+ statistic_(nullptr),
+ time_out_stream_(OUT_STREAM_PAGE_SIZE,
+ common::MOD_PAGE_WRITER_OUTPUT_STREAM),
+ cur_page_data_(),
+ compressor_(nullptr),
+ is_inited_(false) {}
+ int init(common::TSEncoding encoding, common::CompressionType compression);
+ void reset();
+ void destroy();
+
+ FORCE_INLINE int write(int64_t timestamp) {
+ int ret = common::E_OK;
+ if (RET_FAIL(time_encoder_->encode(timestamp, time_out_stream_))) {
+ } else {
+ statistic_->update(timestamp);
+ }
+ return ret;
+ }
+
+ FORCE_INLINE uint32_t get_point_numer() const { return statistic_->count_; }
+ FORCE_INLINE uint32_t get_time_out_stream_size() const {
+ return time_out_stream_.total_size();
+ }
+ FORCE_INLINE uint32_t get_page_memory_size() const {
+ return time_out_stream_.total_size();
+ }
+ FORCE_INLINE uint32_t estimate_max_mem_size() const {
+ return time_out_stream_.total_size() +
+ time_encoder_->get_max_byte_size();
+ }
+ int write_to_chunk(common::ByteStream &pages_data, bool write_header,
+ bool write_statistic, bool write_data_to_chunk_data);
+ FORCE_INLINE common::ByteStream &get_time_data() {
+ return time_out_stream_;
+ }
+ FORCE_INLINE Statistic *get_statistic() { return statistic_; }
+ TimePageData get_cur_page_data() { return cur_page_data_; }
+ void destroy_page_data() { cur_page_data_.destroy(); }
+
+ private:
+ FORCE_INLINE int prepare_end_page() {
+ int ret = common::E_OK;
+ if (RET_FAIL(time_encoder_->flush(time_out_stream_))) {
+ }
+ return ret;
+ }
+ int copy_page_data_to(common::ByteStream &my_page_data,
+ common::ByteStream &pages_data);
+
+ private:
+ static const uint32_t OUT_STREAM_PAGE_SIZE = 1024;
+
+ private:
+ common::TSDataType data_type_;
+ Encoder *time_encoder_;
+ Statistic *statistic_;
+ common::ByteStream time_out_stream_;
+ TimePageData cur_page_data_;
+ Compressor *compressor_;
+ bool is_inited_;
+};
+
+} // end namespace storage
+
+#endif // WRITER_PAGE_TIME_WRITER_H
diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc
index 2ec90a63..53a5464a 100644
--- a/cpp/src/writer/tsfile_writer.cc
+++ b/cpp/src/writer/tsfile_writer.cc
@@ -60,6 +60,7 @@ TsFileWriter::TsFileWriter()
: write_file_(nullptr),
io_writer_(nullptr),
schemas_(),
+ start_file_done_(false),
record_count_since_last_flush_(0),
record_count_for_next_mem_check_(
g_config_value_.record_count_for_next_mem_check_),
@@ -125,12 +126,30 @@ int TsFileWriter::open(const std::string &file_path, int flags, mode_t mode) {
if (RET_FAIL(write_file_->create(file_path, flags, mode))) {
} else {
io_writer_->init(write_file_);
- if (RET_FAIL(io_writer_->start_file())) {
+ }
+ return ret;
+}
+
+int TsFileWriter::register_aligned_timeseries(
+ const std::string &device_path, const std::string &measurement_name,
+ common::TSDataType data_type, common::TSEncoding encoding,
+ common::CompressionType compression_type) {
+ MeasurementSchema *ms = new MeasurementSchema(measurement_name, data_type,
+ encoding, compression_type);
+ return register_timeseries(device_path, ms, true);
+}
+
+int TsFileWriter::register_aligned_timeseries(
+ const std::string &device_path,
+ const std::vector<MeasurementSchema *> &measurement_schema_vec) {
+ int ret = E_OK;
+ std::vector<MeasurementSchema *>::const_iterator it =
+ measurement_schema_vec.begin();
+ for (; it != measurement_schema_vec.end(); it++) {
+ ret = register_timeseries(device_path, *it, true);
+ if (ret != E_OK) {
return ret;
}
-#if DEBUG_SE
- std::cout << "finish writing magic code" << std::endl;
-#endif
}
return ret;
}
@@ -145,7 +164,8 @@ int TsFileWriter::register_timeseries(
}
int TsFileWriter::register_timeseries(const std::string &device_path,
- MeasurementSchema *measurement_schema) {
+ MeasurementSchema *measurement_schema,
+ bool is_aligned) {
DeviceSchemaIter device_iter = schemas_.find(device_path);
if (device_iter != schemas_.end()) {
MeasurementSchemaMap &msm =
@@ -157,6 +177,7 @@ int TsFileWriter::register_timeseries(const std::string &device_path,
}
} else {
MeasurementSchemaGroup *ms_group = new MeasurementSchemaGroup;
+ ms_group->is_aligned_ = is_aligned;
ms_group->measurement_schema_map_.insert(std::make_pair(
measurement_schema->measurement_name_, measurement_schema));
schemas_.insert(std::make_pair(device_path, ms_group));
@@ -180,12 +201,44 @@ int TsFileWriter::register_timeseries(
return ret;
}
-struct MeasurementNamesFromRecord {
- const TsRecord &record_;
+struct MeasurementSchemaMapNamesGetter {
+ public:
+ explicit MeasurementSchemaMapNamesGetter(
+ const MeasurementSchemaMap &measurement_schema_map)
+ : measurement_schema_map_(
+ const_cast<MeasurementSchemaMap &>(measurement_schema_map)) {
+ measurement_name_idx_ = measurement_schema_map_.begin();
+ }
+
+ FORCE_INLINE uint32_t get_count() const {
+ return measurement_schema_map_.size();
+ }
+
+ FORCE_INLINE const std::string &next() {
+ ASSERT(measurement_name_idx_ != measurement_schema_map_.end());
+ std::string &ret = measurement_name_idx_->second->measurement_name_;
+ measurement_name_idx_++;
+ return ret;
+ }
+ private:
+ MeasurementSchemaMap &measurement_schema_map_;
+ MeasurementSchemaMap::iterator measurement_name_idx_;
+};
+
+struct MeasurementNamesFromRecord {
+ public:
explicit MeasurementNamesFromRecord(const TsRecord &record)
- : record_(record) {}
+ : record_(record), measurement_name_idx_(0) {}
FORCE_INLINE uint32_t get_count() const { return record_.points_.size(); }
+
+ FORCE_INLINE const std::string &next() {
+ return this->at(measurement_name_idx_++);
+ }
+
+ private:
+ const TsRecord &record_;
+ size_t measurement_name_idx_;
FORCE_INLINE const std::string &at(size_t idx) const {
ASSERT(idx < record_.points_.size());
return record_.points_[idx].measurement_name_;
@@ -193,13 +246,18 @@ struct MeasurementNamesFromRecord {
};
struct MeasurementNamesFromTablet {
- const Tablet &tablet_;
-
explicit MeasurementNamesFromTablet(const Tablet &tablet)
- : tablet_(tablet) {}
+ : tablet_(tablet), measurement_name_idx_(0) {}
FORCE_INLINE uint32_t get_count() const {
return tablet_.schema_vec_->size();
}
+ FORCE_INLINE const std::string &next() {
+ return this->at(measurement_name_idx_++);
+ }
+
+ private:
+ const Tablet &tablet_;
+ size_t measurement_name_idx_;
FORCE_INLINE const std::string &at(size_t idx) const {
ASSERT(idx < tablet_.schema_vec_->size());
return tablet_.schema_vec_->at(idx).measurement_name_;
@@ -221,7 +279,7 @@ int TsFileWriter::do_check_schema(const std::string &device_name,
uint32_t measurement_count = measurement_names.get_count();
// chunk_writers.reserve(measurement_count);
for (uint32_t i = 0; i < measurement_count; i++) {
- MeasurementSchemaMapIter ms_iter = msm.find(measurement_names.at(i));
+ MeasurementSchemaMapIter ms_iter = msm.find(measurement_names.next());
if (UNLIKELY(ms_iter == msm.end())) {
chunk_writers.push_back(NULL);
} else {
@@ -236,8 +294,15 @@ int TsFileWriter::do_check_schema(const std::string &device_name,
if (IS_SUCC(ret)) {
chunk_writers.push_back(ms->chunk_writer_);
} else {
- std::cout << "get error when chunk init" << std::endl;
- chunk_writers.push_back(NULL);
+ for (size_t chunk_writer_idx = 0;
+ chunk_writer_idx < chunk_writers.size();
+ chunk_writer_idx++) {
+ if (!chunk_writers[chunk_writer_idx]) {
+ delete chunk_writers[chunk_writer_idx];
+ }
+ }
+ ret = common::E_INVALID_ARG;
+ return ret;
}
} else {
chunk_writers.push_back(ms->chunk_writer_);
@@ -247,6 +312,62 @@ int TsFileWriter::do_check_schema(const std::string &device_name,
return ret;
}
+template <typename MeasurementNamesGetter>
+int TsFileWriter::do_check_schema_aligned(
+ const std::string &device_name, MeasurementNamesGetter &measurement_names,
+ storage::TimeChunkWriter *&time_chunk_writer,
+ common::SimpleVector<storage::ValueChunkWriter *> &value_chunk_writers) {
+ int ret = E_OK;
+ auto dev_it = schemas_.find(device_name);
+ MeasurementSchemaGroup *device_schema = NULL;
+ if (UNLIKELY(dev_it == schemas_.end()) ||
+ IS_NULL(device_schema = dev_it->second)) {
+ return E_DEVICE_NOT_EXIST;
+ }
+ if (IS_NULL(device_schema->time_chunk_writer_)) {
+ device_schema->time_chunk_writer_ = new TimeChunkWriter();
+ device_schema->time_chunk_writer_->init(
+ "", g_config_value_.time_encoding_type_,
+ g_config_value_.time_compress_type_);
+ }
+ time_chunk_writer = device_schema->time_chunk_writer_;
+ MeasurementSchemaMap &msm = device_schema->measurement_schema_map_;
+ uint32_t measurement_count = measurement_names.get_count();
+ for (uint32_t i = 0; i < measurement_count; i++) {
+ auto ms_iter = msm.find(measurement_names.next());
+ if (UNLIKELY(ms_iter == msm.end())) {
+ value_chunk_writers.push_back(NULL);
+ } else {
+ // Here we may check data_type against ms_iter. But in Java
+ // libtsfile, no check here.
+ MeasurementSchema *ms = ms_iter->second;
+ if (IS_NULL(ms->value_chunk_writer_)) {
+ ms->value_chunk_writer_ = new ValueChunkWriter;
+ ret = ms->value_chunk_writer_->init(
+ ms->measurement_name_, ms->data_type_, ms->encoding_,
+ ms->compression_type_);
+ if (IS_SUCC(ret)) {
+ value_chunk_writers.push_back(ms->value_chunk_writer_);
+ } else {
+ value_chunk_writers.push_back(NULL);
+ for (size_t chunk_writer_idx = 0;
+ chunk_writer_idx < value_chunk_writers.size();
+ chunk_writer_idx++) {
+ if (!value_chunk_writers[chunk_writer_idx]) {
+ delete value_chunk_writers[chunk_writer_idx];
+ }
+ }
+ ret = common::E_INVALID_ARG;
+ return ret;
+ }
+ } else {
+ value_chunk_writers.push_back(ms->value_chunk_writer_);
+ }
+ }
+ }
+ return ret;
+}
+
int64_t TsFileWriter::calculate_mem_size_for_all_group() {
int64_t mem_total_size = 0;
DeviceSchemaIter device_iter;
@@ -309,6 +430,31 @@ int TsFileWriter::write_record(const TsRecord &record) {
return ret;
}
+int TsFileWriter::write_record_aligned(const TsRecord &record) {
+ int ret = E_OK;
+ SimpleVector<ValueChunkWriter *> value_chunk_writers;
+ TimeChunkWriter *time_chunk_writer;
+ MeasurementNamesFromRecord mnames_getter(record);
+ if (RET_FAIL(do_check_schema_aligned(record.device_name_, mnames_getter,
+ time_chunk_writer,
+ value_chunk_writers))) {
+ return ret;
+ }
+ if (value_chunk_writers.size() != record.points_.size()) {
+ return E_INVALID_ARG;
+ }
+ time_chunk_writer->write(record.timestamp_);
+ for (uint32_t c = 0; c < value_chunk_writers.size(); c++) {
+ ValueChunkWriter *value_chunk_writer = value_chunk_writers[c];
+ if (IS_NULL(value_chunk_writer)) {
+ continue;
+ }
+ write_point_aligned(value_chunk_writer, record.timestamp_,
+ record.points_[c]);
+ }
+ return ret;
+}
+
int TsFileWriter::write_point(ChunkWriter *chunk_writer, int64_t timestamp,
const DataPoint &point) {
switch (point.data_type_) {
@@ -330,9 +476,57 @@ int TsFileWriter::write_point(ChunkWriter *chunk_writer, int64_t timestamp,
}
}
+int TsFileWriter::write_point_aligned(ValueChunkWriter *value_chunk_writer,
+ int64_t timestamp,
+ const DataPoint &point) {
+ bool isnull = point.isnull;
+ switch (point.data_type_) {
+ case common::BOOLEAN:
+ return value_chunk_writer->write(timestamp, point.u_.bool_val_,
+ isnull);
+ case common::INT32:
+ return value_chunk_writer->write(timestamp, point.u_.i32_val_,
+ isnull);
+ case common::INT64:
+ return value_chunk_writer->write(timestamp, point.u_.i64_val_,
+ isnull);
+ case common::FLOAT:
+ return value_chunk_writer->write(timestamp, point.u_.float_val_,
+ isnull);
+ case common::DOUBLE:
+ return value_chunk_writer->write(timestamp, point.u_.double_val_,
+ isnull);
+ case common::TEXT:
+ ASSERT(false);
+ return E_OK;
+ default:
+ return E_INVALID_DATA_POINT;
+ }
+}
+
+int TsFileWriter::write_tablet_aligned(const Tablet &tablet) {
+ int ret = E_OK;
+ SimpleVector<ValueChunkWriter *> value_chunk_writers;
+ TimeChunkWriter *time_chunk_writer = nullptr;
+ MeasurementNamesFromTablet mnames_getter(tablet);
+ if (RET_FAIL(do_check_schema_aligned(tablet.device_name_, mnames_getter,
+ time_chunk_writer,
+ value_chunk_writers))) {
+ return ret;
+ }
+ ASSERT(value_chunk_writers.size() == tablet.get_column_count());
+ for (uint32_t c = 0; c < value_chunk_writers.size(); c++) {
+ ValueChunkWriter *value_chunk_writer = value_chunk_writers[c];
+ if (IS_NULL(value_chunk_writer)) {
+ continue;
+ }
+ value_write_column(value_chunk_writer, tablet, c);
+ }
+ return ret;
+}
+
int TsFileWriter::write_tablet(const Tablet &tablet) {
int ret = E_OK;
- // std::vector<ChunkWriter*> chunk_writers;
SimpleVector<ChunkWriter *> chunk_writers;
MeasurementNamesFromTablet mnames_getter(tablet);
if (RET_FAIL(do_check_schema(tablet.device_name_, mnames_getter,
@@ -361,90 +555,204 @@ int TsFileWriter::write_column(ChunkWriter *chunk_writer, const Tablet &tablet,
TSDataType data_type = tablet.schema_vec_->at(col_idx).data_type_;
int64_t *timestamps = tablet.timestamps_;
void *col_values = tablet.value_matrix_[col_idx];
- BitMap &col_bitmap = tablet.bitmaps_[col_idx];
+ BitMap &col_notnull_bitmap = tablet.bitmaps_[col_idx];
int32_t row_count = tablet.max_rows_;
if (data_type == common::BOOLEAN) {
ret = write_typed_column(chunk_writer, timestamps, (bool *)col_values,
- col_bitmap, row_count);
+ col_notnull_bitmap, row_count);
} else if (data_type == common::INT32) {
- ret = write_typed_column(chunk_writer, timestamps,
- (int32_t *)col_values, col_bitmap, row_count);
+ ret =
+ write_typed_column(chunk_writer, timestamps, (int32_t *)col_values,
+ col_notnull_bitmap, row_count);
} else if (data_type == common::INT64) {
- ret = write_typed_column(chunk_writer, timestamps,
- (int64_t *)col_values, col_bitmap, row_count);
+ ret =
+ write_typed_column(chunk_writer, timestamps, (int64_t *)col_values,
+ col_notnull_bitmap, row_count);
} else if (data_type == common::FLOAT) {
ret = write_typed_column(chunk_writer, timestamps, (float *)col_values,
- col_bitmap, row_count);
+ col_notnull_bitmap, row_count);
} else if (data_type == common::DOUBLE) {
ret = write_typed_column(chunk_writer, timestamps, (double *)col_values,
- col_bitmap, row_count);
+ col_notnull_bitmap, row_count);
} else {
ASSERT(false);
}
return ret;
}
+int TsFileWriter::value_write_column(ValueChunkWriter *value_chunk_writer,
+ const Tablet &tablet, int col_idx) {
+ int ret = E_OK;
+
+ TSDataType data_type = tablet.schema_vec_->at(col_idx).data_type_;
+ int64_t *timestamps = tablet.timestamps_;
+ void *col_values = tablet.value_matrix_[col_idx];
+ BitMap &col_notnull_bitmap = tablet.bitmaps_[col_idx];
+ int32_t row_count = tablet.max_rows_;
+
+ if (data_type == common::BOOLEAN) {
+ ret = write_typed_column(value_chunk_writer, timestamps,
+ (bool *)col_values, col_notnull_bitmap,
+ row_count);
+ } else if (data_type == common::INT32) {
+ ret = write_typed_column(value_chunk_writer, timestamps,
+ (int32_t *)col_values, col_notnull_bitmap,
+ row_count);
+ } else if (data_type == common::INT64) {
+ ret = write_typed_column(value_chunk_writer, timestamps,
+ (int64_t *)col_values, col_notnull_bitmap,
+ row_count);
+ } else if (data_type == common::FLOAT) {
+ ret = write_typed_column(value_chunk_writer, timestamps,
+ (float *)col_values, col_notnull_bitmap,
+ row_count);
+ } else if (data_type == common::DOUBLE) {
+ ret = write_typed_column(value_chunk_writer, timestamps,
+ (double *)col_values, col_notnull_bitmap,
+ row_count);
+ } else {
+ return E_NOT_SUPPORT;
+ }
+ return ret;
+}
+
#define DO_WRITE_TYPED_COLUMN() \
do { \
int ret = E_OK; \
for (int r = 0; r < row_count; r++) { \
- if (LIKELY(col_bitmap.test(r))) { \
+ if (LIKELY(col_notnull_bitmap.test(r))) { \
ret = chunk_writer->write(timestamps[r], col_values[r]); \
} \
} \
return ret; \
} while (false)
+#define DO_VALUE_WRITE_TYPED_COLUMN() \
+ do { \
+ int ret = E_OK; \
+ for (int r = 0; r < row_count; r++) { \
+ if (LIKELY(col_notnull_bitmap.test(r))) { \
+ ret = value_chunk_writer->write(timestamps[r], col_values[r], \
+ false); \
+ } else { \
+ ret = value_chunk_writer->write(timestamps[r], col_values[r], \
+ true); \
+ } \
+ } \
+ return ret; \
+ } while (false)
+
int TsFileWriter::write_typed_column(ChunkWriter *chunk_writer,
int64_t *timestamps, bool *col_values,
- BitMap &col_bitmap, int32_t row_count) {
+ BitMap &col_notnull_bitmap,
+ int32_t row_count) {
DO_WRITE_TYPED_COLUMN();
}
int TsFileWriter::write_typed_column(ChunkWriter *chunk_writer,
int64_t *timestamps, int32_t *col_values,
- BitMap &col_bitmap, int32_t row_count) {
+ BitMap &col_notnull_bitmap,
+ int32_t row_count) {
DO_WRITE_TYPED_COLUMN();
}
int TsFileWriter::write_typed_column(ChunkWriter *chunk_writer,
int64_t *timestamps, int64_t *col_values,
- BitMap &col_bitmap, int32_t row_count) {
+ BitMap &col_notnull_bitmap,
+ int32_t row_count) {
DO_WRITE_TYPED_COLUMN();
}
int TsFileWriter::write_typed_column(ChunkWriter *chunk_writer,
int64_t *timestamps, float *col_values,
- BitMap &col_bitmap, int32_t row_count) {
+ BitMap &col_notnull_bitmap,
+ int32_t row_count) {
DO_WRITE_TYPED_COLUMN();
}
int TsFileWriter::write_typed_column(ChunkWriter *chunk_writer,
int64_t *timestamps, double *col_values,
- BitMap &col_bitmap, int32_t row_count) {
+ BitMap &col_notnull_bitmap,
+ int32_t row_count) {
DO_WRITE_TYPED_COLUMN();
}
+int TsFileWriter::write_typed_column(ValueChunkWriter *value_chunk_writer,
+ int64_t *timestamps, bool *col_values,
+ BitMap &col_notnull_bitmap,
+ int32_t row_count) {
+ DO_VALUE_WRITE_TYPED_COLUMN();
+}
+
+int TsFileWriter::write_typed_column(ValueChunkWriter *value_chunk_writer,
+ int64_t *timestamps, int32_t *col_values,
+ BitMap &col_notnull_bitmap,
+ int32_t row_count) {
+ DO_VALUE_WRITE_TYPED_COLUMN();
+}
+
+int TsFileWriter::write_typed_column(ValueChunkWriter *value_chunk_writer,
+ int64_t *timestamps, int64_t *col_values,
+ BitMap &col_notnull_bitmap,
+ int32_t row_count) {
+ DO_VALUE_WRITE_TYPED_COLUMN();
+}
+
+int TsFileWriter::write_typed_column(ValueChunkWriter *value_chunk_writer,
+ int64_t *timestamps, float *col_values,
+ BitMap &col_notnull_bitmap,
+ int32_t row_count) {
+ DO_VALUE_WRITE_TYPED_COLUMN();
+}
+
+int TsFileWriter::write_typed_column(ValueChunkWriter *value_chunk_writer,
+ int64_t *timestamps, double *col_values,
+ BitMap &col_notnull_bitmap,
+ int32_t row_count) {
+ DO_VALUE_WRITE_TYPED_COLUMN();
+}
+
// TODO make sure ret is meaningful to SDK user
int TsFileWriter::flush() {
int ret = E_OK;
+ if (!start_file_done_) {
+ if (RET_FAIL(io_writer_->start_file())) {
+ return ret;
+ }
+ start_file_done_ = true;
+ }
/* since @schemas_ used std::map which is rbtree underlying,
- so map itself is ordered by device name. */
+ so map itself is ordered by device name. */
std::map<std::string, MeasurementSchemaGroup *>::iterator device_iter;
for (device_iter = schemas_.begin(); device_iter != schemas_.end();
- device_iter++) {
- if (check_chunk_group_empty(device_iter->second)) {
- continue;
+ device_iter++) { // cppcheck-suppress postfixOperator
+ if (device_iter->second->is_aligned_) {
+ SimpleVector<ValueChunkWriter *> value_chunk_writers;
+ TimeChunkWriter *time_chunk_writer;
+ MeasurementSchemaMapNamesGetter mnames_getter(
+ device_iter->second->measurement_schema_map_);
+ if (RET_FAIL(do_check_schema_aligned(
+ device_iter->first, mnames_getter, time_chunk_writer,
+ value_chunk_writers))) {
+ return ret;
+ }
+ } else {
+ SimpleVector<ChunkWriter *> chunk_writers;
+ MeasurementSchemaMapNamesGetter mnames_getter(
+ device_iter->second->measurement_schema_map_);
+ if (RET_FAIL(do_check_schema(device_iter->first, mnames_getter,
+ chunk_writers))) {
+ return ret;
+ }
}
-
- if (RET_FAIL(io_writer_->start_flush_chunk_group(device_iter->first))) {
- return ret;
- } else if (RET_FAIL(flush_chunk_group(device_iter->second))) {
- return ret;
- } else if (RET_FAIL(io_writer_->end_flush_chunk_group())) {
- return ret;
+ bool is_aligned = device_iter->second->is_aligned_;
+ if (RET_FAIL(io_writer_->start_flush_chunk_group(device_iter->first,
+ is_aligned))) {
+ } else if (RET_FAIL(
+ flush_chunk_group(device_iter->second, is_aligned))) {
+ } else if (RET_FAIL(io_writer_->end_flush_chunk_group(is_aligned))) {
}
}
record_count_since_last_flush_ = 0;
@@ -467,31 +775,53 @@ bool TsFileWriter::check_chunk_group_empty(
return true;
}
-int TsFileWriter::flush_chunk_group(MeasurementSchemaGroup *chunk_group) {
+#define FLUSH_CHUNK(writer, io_writer, name, data_type, encoding, compression, \
+ num_pages) \
+ if (RET_FAIL(writer->end_encode_chunk())) { \
+ } else if (RET_FAIL(io_writer->start_flush_chunk( \
+ writer->get_chunk_data(), name, data_type, encoding, \
+ compression, num_pages))) { \
+ } else if (RET_FAIL(io_writer->flush_chunk(writer->get_chunk_data()))) { \
+ } else if (RET_FAIL(io_writer->end_flush_chunk( \
+ writer->get_chunk_statistic()))) { \
+ } else { \
+ writer->destroy(); \
+ writer = nullptr; \
+ }
+
+int TsFileWriter::flush_chunk_group(MeasurementSchemaGroup *chunk_group,
+ bool is_aligned) {
int ret = E_OK;
MeasurementSchemaMap &map = chunk_group->measurement_schema_map_;
+
+ if (chunk_group->is_aligned_) {
+ TimeChunkWriter *&time_chunk_writer = chunk_group->time_chunk_writer_;
+ ChunkHeader chunk_header = time_chunk_writer->get_chunk_header();
+ FLUSH_CHUNK(time_chunk_writer, io_writer_,
+ chunk_header.measurement_name_, chunk_header.data_type_,
+ chunk_header.encoding_type_, chunk_header.compression_type_,
+ time_chunk_writer->num_of_pages())
+ }
+
for (MeasurementSchemaMapIter ms_iter = map.begin(); ms_iter != map.end();
ms_iter++) {
MeasurementSchema *m_schema = ms_iter->second;
- ChunkWriter *&chunk_writer = m_schema->chunk_writer_;
- if (RET_FAIL(chunk_writer->end_encode_chunk())) {
- } else if (RET_FAIL(io_writer_->start_flush_chunk(
- chunk_writer->get_chunk_data(),
- m_schema->measurement_name_, m_schema->data_type_,
- m_schema->encoding_, m_schema->compression_type_,
- chunk_writer->num_of_pages()))) {
- return ret;
- } else if (RET_FAIL(io_writer_->flush_chunk(
- chunk_writer->get_chunk_data()))) {
- return ret;
- } else if (RET_FAIL(io_writer_->end_flush_chunk(
- chunk_writer->get_chunk_statistic()))) {
- return ret;
+ if (!chunk_group->is_aligned_) {
+ ChunkWriter *&chunk_writer = m_schema->chunk_writer_;
+ FLUSH_CHUNK(chunk_writer, io_writer_, m_schema->measurement_name_,
+ m_schema->data_type_, m_schema->encoding_,
+ m_schema->compression_type_,
+ chunk_writer->num_of_pages())
} else {
- chunk_writer->destroy();
- chunk_writer = nullptr;
+ ValueChunkWriter *&value_chunk_writer =
+ m_schema->value_chunk_writer_;
+ FLUSH_CHUNK(value_chunk_writer, io_writer_,
+ m_schema->measurement_name_, m_schema->data_type_,
+ m_schema->encoding_, m_schema->compression_type_,
+ value_chunk_writer->num_of_pages())
}
}
+
return ret;
}
diff --git a/cpp/src/writer/tsfile_writer.h b/cpp/src/writer/tsfile_writer.h
index 099c6467..128715a4 100644
--- a/cpp/src/writer/tsfile_writer.h
+++ b/cpp/src/writer/tsfile_writer.h
@@ -57,11 +57,20 @@ class TsFileWriter {
common::TSDataType data_type,
common::TSEncoding encoding,
common::CompressionType compression_type);
+ int register_aligned_timeseries(const std::string &device_path,
+ const std::string &measurement_name,
+ common::TSDataType data_type,
+ common::TSEncoding encoding,
+ common::CompressionType compression_type);
+ int register_aligned_timeseries(
+ const std::string &device_path,
+ const std::vector<MeasurementSchema *> &measurement_schema_vec);
int write_record(const TsRecord &record);
int write_tablet(const Tablet &tablet);
+ int write_record_aligned(const TsRecord &record);
+ int write_tablet_aligned(const Tablet &tablet);
int64_t calculate_mem_size_for_all_group();
int check_memory_size_and_may_flush_chunks();
-
/*
* Flush buffer to disk file, but do not writer file index part.
* TsFileWriter allows user to flush many times.
@@ -78,34 +87,48 @@ class TsFileWriter {
int write_point(storage::ChunkWriter *chunk_writer, int64_t timestamp,
const DataPoint &point);
bool check_chunk_group_empty(MeasurementSchemaGroup *chunk_group);
- int flush_chunk_group(MeasurementSchemaGroup *chunk_group);
+ int write_point_aligned(ValueChunkWriter *value_chunk_writer,
+ int64_t timestamp, const DataPoint &point);
+ int flush_chunk_group(MeasurementSchemaGroup *chunk_group, bool is_aligned);
int write_typed_column(storage::ChunkWriter *chunk_writer,
int64_t *timestamps, bool *col_values,
- common::BitMap &col_bitmap, int32_t row_count);
+ common::BitMap &col_notnull_bitmap,
+ int32_t row_count);
int write_typed_column(storage::ChunkWriter *chunk_writer,
int64_t *timestamps, int32_t *col_values,
- common::BitMap &col_bitmap, int32_t row_count);
+ common::BitMap &col_notnull_bitmap,
+ int32_t row_count);
int write_typed_column(storage::ChunkWriter *chunk_writer,
int64_t *timestamps, int64_t *col_values,
- common::BitMap &col_bitmap, int32_t row_count);
+ common::BitMap &col_notnull_bitmap,
+ int32_t row_count);
int write_typed_column(storage::ChunkWriter *chunk_writer,
int64_t *timestamps, float *col_values,
- common::BitMap &col_bitmap, int32_t row_count);
+ common::BitMap &col_notnull_bitmap,
+ int32_t row_count);
int write_typed_column(storage::ChunkWriter *chunk_writer,
int64_t *timestamps, double *col_values,
- common::BitMap &col_bitmap, int32_t row_count);
+ common::BitMap &col_notnull_bitmap,
+ int32_t row_count);
template <typename MeasurementNamesGetter>
int do_check_schema(
const std::string &device_name,
MeasurementNamesGetter &measurement_names,
common::SimpleVector<storage::ChunkWriter *> &chunk_writers);
+ template <typename MeasurementNamesGetter>
+ int do_check_schema_aligned(
+ const std::string &device_name,
+ MeasurementNamesGetter &measurement_names,
+ storage::TimeChunkWriter *&time_chunk_writer,
+ common::SimpleVector<storage::ValueChunkWriter *> &value_chunk_writers);
// std::vector<storage::ChunkWriter*> &chunk_writers);
int write_column(storage::ChunkWriter *chunk_writer, const Tablet &tablet,
int col_idx);
int register_timeseries(const std::string &device_path,
- MeasurementSchema *measurement_schema);
+ MeasurementSchema *measurement_schema,
+ bool is_aligned = false);
int register_timeseries(
const std::string &device_path,
const std::vector<MeasurementSchema *> &measurement_schema_vec);
@@ -115,11 +138,40 @@ class TsFileWriter {
storage::TsFileIOWriter *io_writer_;
// device_name -> MeasurementSchemaGroup
std::map<std::string, MeasurementSchemaGroup *> schemas_;
+ bool start_file_done_;
// record count since last flush
int64_t record_count_since_last_flush_;
// record count for next memory check
int64_t record_count_for_next_mem_check_;
bool write_file_created_;
+
+ int write_typed_column(ValueChunkWriter *value_chunk_writer,
+ int64_t *timestamps, bool *col_values,
+ common::BitMap &col_notnull_bitmap,
+ int32_t row_count);
+
+ int write_typed_column(ValueChunkWriter *value_chunk_writer,
+ int64_t *timestamps, double *col_values,
+ common::BitMap &col_notnull_bitmap,
+ int32_t row_count);
+
+ int write_typed_column(ValueChunkWriter *value_chunk_writer,
+ int64_t *timestamps, float *col_values,
+ common::BitMap &col_notnull_bitmap,
+ int32_t row_count);
+
+ int write_typed_column(ValueChunkWriter *value_chunk_writer,
+ int64_t *timestamps, int32_t *col_values,
+ common::BitMap &col_notnull_bitmap,
+ int32_t row_count);
+
+ int write_typed_column(ValueChunkWriter *value_chunk_writer,
+ int64_t *timestamps, int64_t *col_values,
+ common::BitMap &col_notnull_bitmap,
+ int32_t row_count);
+
+ int value_write_column(ValueChunkWriter *value_chunk_writer,
+ const Tablet &tablet, int col_idx);
};
} // end namespace storage
diff --git a/cpp/src/writer/chunk_writer.cc b/cpp/src/writer/value_chunk_writer.cc
similarity index 67%
copy from cpp/src/writer/chunk_writer.cc
copy to cpp/src/writer/value_chunk_writer.cc
index a283fb96..7585a610 100644
--- a/cpp/src/writer/chunk_writer.cc
+++ b/cpp/src/writer/value_chunk_writer.cc
@@ -17,7 +17,7 @@
* under the License.
*/
-#include "chunk_writer.h"
+#include "value_chunk_writer.h"
#include "common/logger/elog.h"
@@ -25,19 +25,20 @@ using namespace common;
namespace storage {
-int ChunkWriter::init(const ColumnDesc &col_desc) {
+int ValueChunkWriter::init(const ColumnDesc &col_desc) {
return init(col_desc.column_name_, col_desc.type_, col_desc.encoding_,
col_desc.compression_);
}
-int ChunkWriter::init(const std::string &measurement_name, TSDataType data_type,
- TSEncoding encoding, CompressionType compression_type) {
+int ValueChunkWriter::init(const std::string &measurement_name,
+ TSDataType data_type, TSEncoding encoding,
+ CompressionType compression_type) {
int ret = E_OK;
chunk_statistic_ = StatisticFactory::alloc_statistic(data_type);
if (chunk_statistic_ == nullptr) {
return E_OOM;
- } else if (RET_FAIL(
- page_writer_.init(data_type, encoding, compression_type))) {
+ } else if (RET_FAIL(value_page_writer_.init(data_type, encoding,
+ compression_type))) {
} else if (IS_NULL((first_page_statistic_ =
StatisticFactory::alloc_statistic(data_type)))) {
ret = E_OOM;
@@ -51,8 +52,8 @@ int ChunkWriter::init(const std::string &measurement_name, TSDataType data_type,
return ret;
}
-void ChunkWriter::destroy() {
- page_writer_.destroy();
+void ValueChunkWriter::destroy() {
+ value_page_writer_.destroy();
if (chunk_statistic_ != nullptr) {
StatisticFactory::free(chunk_statistic_);
chunk_statistic_ = nullptr;
@@ -66,30 +67,33 @@ void ChunkWriter::destroy() {
num_of_pages_ = 0;
}
-int ChunkWriter::seal_cur_page(bool end_chunk) {
+int ValueChunkWriter::seal_cur_page(bool end_chunk) {
int ret = E_OK;
- if (RET_FAIL(chunk_statistic_->merge_with(page_writer_.get_statistic()))) {
+ if (RET_FAIL(
+ chunk_statistic_->merge_with(value_page_writer_.get_statistic()))) {
return ret;
}
if (num_of_pages_ == 0) {
if (end_chunk) {
// this page is the only one page of this chunk
- ret = page_writer_.write_to_chunk(chunk_data_, /*header*/ true,
- /*stat*/ false, /*data*/ true);
- page_writer_.destroy_page_data();
- page_writer_.destroy();
+ ret = value_page_writer_.write_to_chunk(
+ chunk_data_, /*header*/ true,
+ /*stat*/ false, /*data*/ true);
+ value_page_writer_.destroy_page_data();
+ value_page_writer_.destroy();
} else {
/*
* if the chunk has only one page, do not writer page statistic.
* so we save the data of first page and see if the chunk has more
* page later.
*/
- ret = page_writer_.write_to_chunk(chunk_data_, /*header*/ true,
- /*stat*/ false, /*data*/ false);
+ ret = value_page_writer_.write_to_chunk(
+ chunk_data_, /*header*/ true,
+ /*stat*/ false, /*data*/ false);
if (IS_SUCC(ret)) {
- save_first_page_data(page_writer_);
- // page_writer_.destroy_page_data();
- page_writer_.reset();
+ save_first_page_data(value_page_writer_);
+ // value_page_writer_.destroy_page_data();
+ value_page_writer_.reset();
}
}
} else {
@@ -100,12 +104,12 @@ int ChunkWriter::seal_cur_page(bool end_chunk) {
free_first_writer_data();
}
if (IS_SUCC(ret)) {
- if (RET_FAIL(page_writer_.write_to_chunk(
+ if (RET_FAIL(value_page_writer_.write_to_chunk(
chunk_data_, /*header*/ true, /*stat*/ true,
/*data*/ true))) {
}
- page_writer_.destroy_page_data();
- page_writer_.reset();
+ value_page_writer_.destroy_page_data();
+ value_page_writer_.reset();
}
}
num_of_pages_++;
@@ -119,12 +123,13 @@ int ChunkWriter::seal_cur_page(bool end_chunk) {
return ret;
}
-void ChunkWriter::save_first_page_data(PageWriter &first_page_writer) {
+void ValueChunkWriter::save_first_page_data(
+ ValuePageWriter &first_page_writer) {
first_page_data_ = first_page_writer.get_cur_page_data();
first_page_statistic_->deep_copy_from(first_page_writer.get_statistic());
}
-int ChunkWriter::write_first_page_data(ByteStream &pages_data) {
+int ValueChunkWriter::write_first_page_data(ByteStream &pages_data) {
int ret = E_OK;
if (RET_FAIL(first_page_statistic_->serialize_to(pages_data))) {
} else if (RET_FAIL(
@@ -134,9 +139,9 @@ int ChunkWriter::write_first_page_data(ByteStream &pages_data) {
return ret;
}
-int ChunkWriter::end_encode_chunk() {
+int ValueChunkWriter::end_encode_chunk() {
int ret = E_OK;
- if (page_writer_.get_statistic()->count_ > 0) {
+ if (value_page_writer_.get_statistic()->count_ > 0) {
ret = seal_cur_page(/*end_chunk*/ true);
if (E_OK == ret) {
chunk_header_.data_size_ = chunk_data_.total_size();
@@ -147,17 +152,17 @@ int ChunkWriter::end_encode_chunk() {
std::cout << "end_encode_chunk: num_of_pages_=" << num_of_pages_
<< ", chunk_header_.data_size_=" << chunk_header_.data_size_
<< ", page_writer.get_statistic()->count_="
- << page_writer_.get_statistic()->count_ << std::endl;
+ << value_page_writer_.get_statistic()->count_ << std::endl;
#endif
return ret;
}
-
-int64_t ChunkWriter::estimate_max_series_mem_size(){
- return chunk_data_.total_size()
- + page_writer_.estimate_max_mem_size()
- + PageHeader::estimat_max_page_header_size_without_statistics()
- + get_typed_statistic_sizeof(page_writer_.get_statistic()->get_type());
+int64_t ValueChunkWriter::estimate_max_series_mem_size() {
+ return chunk_data_.total_size() +
+ value_page_writer_.estimate_max_mem_size() +
+ PageHeader::estimat_max_page_header_size_without_statistics() +
+ get_typed_statistic_sizeof(
+ value_page_writer_.get_statistic()->get_type());
}
} // end namespace storage
diff --git a/cpp/src/writer/value_chunk_writer.h b/cpp/src/writer/value_chunk_writer.h
new file mode 100644
index 00000000..3093dfff
--- /dev/null
+++ b/cpp/src/writer/value_chunk_writer.h
@@ -0,0 +1,129 @@
+/*
+ * 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 WRITER_CHUNK_VALUE_WRITER_H
+#define WRITER_CHUNK_VALUE_WRITER_H
+
+#include "common/allocator/byte_stream.h"
+#include "common/global.h"
+#include "common/statistic.h"
+#include "common/tsfile_common.h"
+#include "value_page_writer.h"
+
+namespace storage {
+
+#define VCW_DO_WRITE_FOR_TYPE(TSDATATYPE, ISNULL) \
+ { \
+ int ret = common::E_OK; \
+ if (UNLIKELY(data_type_ != TSDATATYPE)) { \
+ return common::E_TYPE_NOT_MATCH; \
+ } \
+ if (RET_FAIL(value_page_writer_.write(timestamp, value, ISNULL))) { \
+ return ret; \
+ } \
+ if (RET_FAIL(seal_cur_page_if_full())) { \
+ return ret; \
+ } \
+ return ret; \
+ }
+
+class ValueChunkWriter {
+ public:
+ static const int32_t PAGES_DATA_PAGE_SIZE = 1024;
+
+ public:
+ ValueChunkWriter()
+ : data_type_(common::VECTOR),
+ value_page_writer_(),
+ chunk_statistic_(nullptr),
+ chunk_data_(PAGES_DATA_PAGE_SIZE, common::MOD_CW_PAGES_DATA),
+ first_page_data_(),
+ first_page_statistic_(nullptr),
+ chunk_header_(),
+ num_of_pages_(0) {}
+ ~ValueChunkWriter() { destroy(); }
+ int init(const common::ColumnDesc &col_desc);
+ int init(const std::string &measurement_name, common::TSDataType data_type,
+ common::TSEncoding encoding,
+ common::CompressionType compression_type);
+ void destroy();
+
+ FORCE_INLINE int write(int64_t timestamp, bool value, bool isnull) {
+ VCW_DO_WRITE_FOR_TYPE(common::BOOLEAN, isnull);
+ }
+ FORCE_INLINE int write(int64_t timestamp, int32_t value, bool isnull) {
+ VCW_DO_WRITE_FOR_TYPE(common::INT32, isnull);
+ }
+ FORCE_INLINE int write(int64_t timestamp, int64_t value, bool isnull) {
+ VCW_DO_WRITE_FOR_TYPE(common::INT64, isnull);
+ }
+ FORCE_INLINE int write(int64_t timestamp, float value, bool isnull) {
+ VCW_DO_WRITE_FOR_TYPE(common::FLOAT, isnull);
+ }
+ FORCE_INLINE int write(int64_t timestamp, double value, bool isnull) {
+ VCW_DO_WRITE_FOR_TYPE(common::DOUBLE, isnull);
+ }
+
+ int end_encode_chunk();
+ common::ByteStream &get_chunk_data() { return chunk_data_; }
+ Statistic *get_chunk_statistic() { return chunk_statistic_; }
+ FORCE_INLINE int32_t num_of_pages() const { return num_of_pages_; }
+
+ int64_t estimate_max_series_mem_size();
+
+ private:
+ FORCE_INLINE bool is_cur_page_full() const {
+ // FIXME
+ return (value_page_writer_.get_point_numer() >=
+ common::g_config_value_.page_writer_max_point_num_) ||
+ (value_page_writer_.get_page_memory_size() >=
+ common::g_config_value_.page_writer_max_memory_bytes_);
+ }
+ FORCE_INLINE int seal_cur_page_if_full() {
+ if (UNLIKELY(is_cur_page_full())) {
+ return seal_cur_page(false);
+ }
+ return common::E_OK;
+ }
+ FORCE_INLINE void free_first_writer_data() {
+ first_page_data_.destroy();
+ StatisticFactory::free(first_page_statistic_);
+ first_page_statistic_ = nullptr;
+ }
+ int seal_cur_page(bool end_chunk);
+ void save_first_page_data(ValuePageWriter &first_page_writer);
+ int write_first_page_data(common::ByteStream &pages_data);
+
+ private:
+ common::TSDataType data_type_;
+ ValuePageWriter value_page_writer_;
+ Statistic *chunk_statistic_;
+ common::ByteStream chunk_data_;
+
+ // to save first page data
+ ValuePageData first_page_data_;
+ Statistic *first_page_statistic_;
+
+ ChunkHeader chunk_header_;
+ int32_t num_of_pages_;
+};
+
+} // end namespace storage
+
+#endif // WRITER_CHUNK_WRITER_H
diff --git a/cpp/src/writer/page_writer.cc b/cpp/src/writer/value_page_writer.cc
similarity index 57%
copy from cpp/src/writer/page_writer.cc
copy to cpp/src/writer/value_page_writer.cc
index 539f6069..f636af8b 100644
--- a/cpp/src/writer/page_writer.cc
+++ b/cpp/src/writer/value_page_writer.cc
@@ -17,7 +17,7 @@
* under the License.
*/
-#include "page_writer.h"
+#include "value_page_writer.h"
#include "common/config/config.h"
#include "common/logger/elog.h"
@@ -28,30 +28,37 @@ using namespace common;
namespace storage {
-/* ================ PageData ================ */
-int PageData::init(ByteStream &time_bs, ByteStream &value_bs,
- Compressor *compressor) {
+uint32_t ValuePageWriter::MASK = 1 << 7;
+
+int ValuePageData::init(ByteStream &col_notnull_bitmap_bs, ByteStream &value_bs,
+ Compressor *compressor, uint32_t size) {
int ret = E_OK;
- time_buf_size_ = time_bs.total_size();
+ col_notnull_bitmap_buf_size_ = col_notnull_bitmap_bs.total_size();
value_buf_size_ = value_bs.total_size();
- uint32_t var_size = get_var_uint_size(time_buf_size_);
- uncompressed_size_ = var_size + time_buf_size_ + value_buf_size_;
+ uncompressed_size_ =
+ sizeof(size) + col_notnull_bitmap_buf_size_ + value_buf_size_;
uncompressed_buf_ =
(char *)mem_alloc(uncompressed_size_, MOD_PAGE_WRITER_OUTPUT_STREAM);
compressor_ = compressor;
if (IS_NULL(uncompressed_buf_)) {
return E_OOM;
}
- if (time_buf_size_ == 0 || value_buf_size_ == 0) {
+ if (col_notnull_bitmap_buf_size_ == 0 || value_buf_size_ == 0) {
return E_INVALID_ARG;
}
- if (RET_FAIL(SerializationUtil::write_var_uint(
- time_buf_size_, uncompressed_buf_, var_size))) {
- } else if (RET_FAIL(copy_bs_to_buf(time_bs, uncompressed_buf_ + var_size,
- uncompressed_size_ - var_size))) {
- } else if (RET_FAIL(copy_bs_to_buf(
- value_bs, uncompressed_buf_ + var_size + time_buf_size_,
- uncompressed_size_ - var_size - time_buf_size_))) {
+ uncompressed_buf_[0] = (unsigned char)((size >> 24) & 0xFF);
+ uncompressed_buf_[1] = (unsigned char)((size >> 16) & 0xFF);
+ uncompressed_buf_[2] = (unsigned char)((size >> 8) & 0xFF);
+ uncompressed_buf_[3] = (unsigned char)((size)&0xFF);
+
+ if (RET_FAIL(common::copy_bs_to_buf(col_notnull_bitmap_bs,
+ uncompressed_buf_ + sizeof(size),
+ col_notnull_bitmap_buf_size_))) {
+ } else if (RET_FAIL(common::copy_bs_to_buf(value_bs,
+ uncompressed_buf_ +
+ sizeof(size) +
+ col_notnull_bitmap_buf_size_,
+ value_buf_size_))) {
} else {
// TODO
// NOTE: different compressor may have different compress API
@@ -63,44 +70,23 @@ int PageData::init(ByteStream &time_bs, ByteStream &value_bs,
}
}
#if DEBUG_SE
- std::cout << "PageData::init. time_buf_size=" << time_buf_size_
+ std::cout << "ValuePageData::init. col_notnull_bitmap_buf_size="
+ << col_notnull_bitmap_buf_size_ << ", size_=" << size
<< ", value_buf_size=" << value_buf_size_
<< ", uncompressed_size=" << uncompressed_size_
<< ", compressed_size=" << compressed_size_ << std::endl;
- // DEBUG_hex_dump_buf("compressed_buf=", compressed_buf_, compressed_size_);
+ DEBUG_hex_dump_buf("uncompressed_buf=", uncompressed_buf_,
+ uncompressed_size_);
#endif
return ret;
}
-int PageData::copy_bs_to_buf(ByteStream &bs, char *src_buf,
- uint32_t src_buf_len) {
- ByteStream::BufferIterator buf_iter = bs.init_buffer_iterator();
- uint32_t copyed_len = 0;
- while (true) {
- ByteStream::Buffer buf = buf_iter.get_next_buf();
- if (buf.buf_ == nullptr) {
- break;
- } else {
- if (src_buf_len - copyed_len < buf.len_) {
- ASSERT(false);
- return E_BUF_NOT_ENOUGH;
- }
- memcpy(src_buf + copyed_len, buf.buf_, buf.len_);
- copyed_len += buf.len_;
- }
- }
- return E_OK;
-}
-
-/* ================ PageWriter ================ */
-int PageWriter::init(TSDataType data_type, TSEncoding encoding,
- CompressionType compression) {
+int ValuePageWriter::init(TSDataType data_type, TSEncoding encoding,
+ CompressionType compression) {
int ret = E_OK;
data_type_ = data_type;
- if (nullptr == (time_encoder_ = EncoderFactory::alloc_time_encoder())) {
- ret = E_OOM;
- } else if (nullptr == (value_encoder_ = EncoderFactory::alloc_value_encoder(
- encoding, data_type))) {
+ if (nullptr == (value_encoder_ = EncoderFactory::alloc_value_encoder(
+ encoding, data_type))) {
ret = E_OOM;
} else if (nullptr ==
(statistic_ = StatisticFactory::alloc_statistic(data_type))) {
@@ -110,9 +96,6 @@ int PageWriter::init(TSDataType data_type, TSEncoding encoding,
ret = E_OOM;
}
if (ret != E_OK) {
- if (time_encoder_ != nullptr) {
- EncoderFactory::free(time_encoder_);
- }
if (value_encoder_ != nullptr) {
EncoderFactory::free(value_encoder_);
}
@@ -126,46 +109,42 @@ int PageWriter::init(TSDataType data_type, TSEncoding encoding,
return ret;
}
-/*
- * free out_stream memory, reset statistic_,
- */
-void PageWriter::reset() {
- time_encoder_->reset();
+void ValuePageWriter::reset() {
value_encoder_->reset();
statistic_->reset();
- time_out_stream_.reset();
+ col_notnull_bitmap_out_stream_.reset();
value_out_stream_.reset();
}
-void PageWriter::destroy() {
+void ValuePageWriter::destroy() {
if (is_inited_) {
is_inited_ = false;
- time_encoder_->destroy();
value_encoder_->destroy();
statistic_->destroy();
- EncoderFactory::free(time_encoder_);
EncoderFactory::free(value_encoder_);
StatisticFactory::free(statistic_);
- compressor_->destroy();
CompressorFactory::free(compressor_);
}
}
-int PageWriter::write_to_chunk(ByteStream &pages_data, bool write_header,
- bool write_statistic,
- bool write_data_to_chunk_data) {
+int ValuePageWriter::write_to_chunk(ByteStream &pages_data, bool write_header,
+ bool write_statistic,
+ bool write_data_to_chunk_data) {
#if DEBUG_SE
- std::cout << "PageWriter::write_to_chunk at position "
+ std::cout << "ValuePageWriter::write_to_chunk at position "
<< pages_data.total_size() << " of chunk_data." << std::endl;
#endif
int ret = E_OK;
if (RET_FAIL(prepare_end_page())) {
return ret;
}
- if (RET_FAIL(cur_page_data_.init(time_out_stream_, value_out_stream_,
- compressor_))) {
+ if (RET_FAIL(cur_page_data_.init(col_notnull_bitmap_out_stream_,
+ value_out_stream_, compressor_, size_))) {
}
+ col_notnull_bitmap_.clear();
+ size_ = 0;
+ col_notnull_bitmap_out_stream_.reset();
if (IS_SUCC(ret) && write_header) {
if (RET_FAIL(SerializationUtil::write_var_uint(
@@ -174,15 +153,15 @@ int PageWriter::write_to_chunk(ByteStream &pages_data, bool write_header,
cur_page_data_.compressed_size_, pages_data))) {
}
}
- // std::cout << "PageWriter::write_to_chunk after write_header. pos=" <<
- // pages_data.total_size() << std::endl;
+ // std::cout << "ValuePageWriter::write_to_chunk after write_header. pos="
+ // << pages_data.total_size() << std::endl;
if (IS_SUCC(ret) && write_statistic) {
if (RET_FAIL(statistic_->serialize_to(pages_data))) {
}
}
- // std::cout << "PageWriter::write_to_chunk after write_stat. pos=" <<
+ // std::cout << "ValuePageWriter::write_to_chunk after write_stat. pos=" <<
// pages_data.total_size() << std::endl; DEBUG_print_byte_stream("In
- // PageWriter::write_to_chunk, before writer page data: pages_data = ",
+ // ValuePageWriter::write_to_chunk, before writer page data: pages_data = ",
// pages_data);
if (IS_SUCC(ret) && write_data_to_chunk_data) {
// DEBUG_hex_dump_buf("cur_page_data_.compressed_buf_ = ",
@@ -191,12 +170,12 @@ int PageWriter::write_to_chunk(ByteStream &pages_data, bool write_header,
cur_page_data_.compressed_size_))) {
}
}
- // DEBUG_print_byte_stream("In PageWriter::write_to_chunk, after writer page
- // data: pages_data = ", pages_data); std::cout <<
- // "PageWriter::write_to_chunk after write_data. pos=" <<
+ // DEBUG_print_byte_stream("In ValuePageWriter::write_to_chunk, after writer
+ // page data: pages_data = ", pages_data); std::cout <<
+ // "ValuePageWriter::write_to_chunk after write_data. pos=" <<
// pages_data.total_size() << std::endl;
#if DEBUG_SE
- std::cout << "PageWriter write_to_chunk: ret=" << ret
+ std::cout << "ValuePageWriter write_to_chunk: ret=" << ret
<< ", flags(HSD)=" << write_header << write_statistic
<< write_data_to_chunk_data
<< ", uncompressed_size=" << cur_page_data_.uncompressed_size_
diff --git a/cpp/src/writer/page_writer.h b/cpp/src/writer/value_page_writer.h
similarity index 54%
copy from cpp/src/writer/page_writer.h
copy to cpp/src/writer/value_page_writer.h
index 6800c7ea..5ce65a5d 100644
--- a/cpp/src/writer/page_writer.h
+++ b/cpp/src/writer/value_page_writer.h
@@ -16,10 +16,13 @@
* specific language governing permissions and limitations
* under the License.
*/
-#ifndef WRITER_PAGE_WRITER_H
-#define WRITER_PAGE_WRITER_H
+#ifndef WRITER_PAGE_VALUE_WRITER_H
+#define WRITER_PAGE_VALUE_WRITER_H
+
+#include <vector>
#include "common/allocator/byte_stream.h"
+#include "common/container/bit_map.h"
#include "common/statistic.h"
#include "compress/compressor.h"
#include "encoding/encoder.h"
@@ -27,21 +30,8 @@
namespace storage {
-/* ================ PageData ================ */
-/*
- * PageData consists of:
- * 1. time_buf_size
- * 2. time_buf
- * 3. value_buf
- *
- * Since page header consists compressed_size_ and uncompressed_size_,
- * so we can deduce the value_buf_size_.
- *
- * We abstract this struct because chunk writer will delay writer
- * the first page data. So we have to save the first page data.
- */
-struct PageData {
- uint32_t time_buf_size_;
+struct ValuePageData {
+ uint32_t col_notnull_bitmap_buf_size_;
uint32_t value_buf_size_;
uint32_t uncompressed_size_;
uint32_t compressed_size_;
@@ -49,16 +39,17 @@ struct PageData {
char *compressed_buf_;
Compressor *compressor_;
- PageData()
- : time_buf_size_(0),
+ ValuePageData()
+ : col_notnull_bitmap_buf_size_(0),
value_buf_size_(0),
uncompressed_size_(0),
compressed_size_(0),
uncompressed_buf_(nullptr),
compressed_buf_(nullptr),
compressor_(nullptr) {}
- int init(common::ByteStream &time_bs, common::ByteStream &value_bs,
- Compressor *compressor);
+ int init(common::ByteStream &col_notnull_bitmap_bs,
+ common::ByteStream &value_bs, Compressor *compressor,
+ uint32_t size);
void destroy() {
// Be careful about the memory
if (uncompressed_buf_ != nullptr) {
@@ -70,69 +61,75 @@ struct PageData {
compressed_buf_ = nullptr;
}
}
- int copy_bs_to_buf(common::ByteStream &bs, char *buf, uint32_t buf_len);
};
-/* ================ PageWriter ================ */
-#define PW_DO_WRITE_FOR_TYPE(TSDATATYPE) \
- { \
- int ret = common::E_OK; \
- /* std::cout << "page_writer writer: time=" << timestamp << ", value=" \
- * << value << std::endl; */ \
- if (UNLIKELY(data_type_ != TSDATATYPE)) { \
- ret = common::E_TYPE_NOT_MATCH; \
- } else if (RET_FAIL( \
- time_encoder_->encode(timestamp, time_out_stream_))) { \
- } else if (RET_FAIL( \
- value_encoder_->encode(value, value_out_stream_))) { \
- } else { \
- statistic_->update(timestamp, value); \
- } \
- return ret; \
+#define VPW_DO_WRITE_FOR_TYPE(TSDATATYPE, ISNULL) \
+ { \
+ int ret = common::E_OK; \
+ if (UNLIKELY(data_type_ != TSDATATYPE)) { \
+ ret = common::E_TYPE_NOT_MATCH; \
+ return ret; \
+ } \
+ if (!ISNULL) { \
+ if ((size_ / 8) + 1 > col_notnull_bitmap_.size()) { \
+ col_notnull_bitmap_.push_back(0); \
+ } \
+ col_notnull_bitmap_[size_ / 8] |= (MASK >> (size_ % 8)); \
+ } \
+ size_++; \
+ if (ISNULL) { \
+ return ret; \
+ } \
+ if (RET_FAIL(value_encoder_->encode(value, value_out_stream_))) { \
+ } else { \
+ statistic_->update(timestamp, value); \
+ } \
+ return ret; \
}
-class PageWriter {
+class ValuePageWriter {
public:
- PageWriter()
+ ValuePageWriter()
: data_type_(common::VECTOR),
- time_encoder_(nullptr),
value_encoder_(nullptr),
statistic_(nullptr),
- time_out_stream_(OUT_STREAM_PAGE_SIZE,
- common::MOD_PAGE_WRITER_OUTPUT_STREAM),
+ col_notnull_bitmap_out_stream_(OUT_STREAM_PAGE_SIZE,
+ common::MOD_PAGE_WRITER_OUTPUT_STREAM),
value_out_stream_(OUT_STREAM_PAGE_SIZE,
common::MOD_PAGE_WRITER_OUTPUT_STREAM),
cur_page_data_(),
compressor_(nullptr),
- is_inited_(false) {}
+ is_inited_(false),
+ col_notnull_bitmap_(),
+ size_(0) {}
int init(common::TSDataType data_type, common::TSEncoding encoding,
common::CompressionType compression);
- // reset statistic_, time_out_stream_, value_out_stream_
void reset();
void destroy();
- FORCE_INLINE int write(int64_t timestamp, bool value) {
- PW_DO_WRITE_FOR_TYPE(common::BOOLEAN);
+ FORCE_INLINE int write(int64_t timestamp, bool value, bool isnull) {
+ VPW_DO_WRITE_FOR_TYPE(common::BOOLEAN, isnull);
}
- FORCE_INLINE int write(int64_t timestamp, int32_t value) {
- PW_DO_WRITE_FOR_TYPE(common::INT32);
+ FORCE_INLINE int write(int64_t timestamp, int32_t value, bool isnull) {
+ VPW_DO_WRITE_FOR_TYPE(common::INT32, isnull);
}
- FORCE_INLINE int write(int64_t timestamp, int64_t value) {
- PW_DO_WRITE_FOR_TYPE(common::INT64);
+ FORCE_INLINE int write(int64_t timestamp, int64_t value, bool isnull) {
+ VPW_DO_WRITE_FOR_TYPE(common::INT64, isnull);
}
- FORCE_INLINE int write(int64_t timestamp, float value) {
- PW_DO_WRITE_FOR_TYPE(common::FLOAT);
+ FORCE_INLINE int write(int64_t timestamp, float value, bool isnull) {
+ VPW_DO_WRITE_FOR_TYPE(common::FLOAT, isnull);
}
- FORCE_INLINE int write(int64_t timestamp, double value) {
- PW_DO_WRITE_FOR_TYPE(common::DOUBLE);
+ FORCE_INLINE int write(int64_t timestamp, double value, bool isnull) {
+ VPW_DO_WRITE_FOR_TYPE(common::DOUBLE, isnull);
}
FORCE_INLINE uint32_t get_point_numer() const { return statistic_->count_; }
- FORCE_INLINE uint32_t get_time_out_stream_size() const {
- return time_out_stream_.total_size();
+ FORCE_INLINE uint32_t get_col_notnull_bitmap_out_stream_size() const {
+ return col_notnull_bitmap_out_stream_.total_size();
}
FORCE_INLINE uint32_t get_page_memory_size() const {
- return time_out_stream_.total_size() + value_out_stream_.total_size();
+ return col_notnull_bitmap_out_stream_.total_size() +
+ value_out_stream_.total_size();
}
/**
* calculate max possible memory size it occupies, including time
@@ -142,27 +139,31 @@ class PageWriter {
* @return allocated size in time, value and outputStream
*/
FORCE_INLINE uint32_t estimate_max_mem_size() const {
- return time_out_stream_.total_size() + value_out_stream_.total_size() +
- time_encoder_->get_max_byte_size() +
+ return sizeof(int32_t) + 1 +
+ col_notnull_bitmap_out_stream_.total_size() +
+ value_out_stream_.total_size() +
value_encoder_->get_max_byte_size();
}
int write_to_chunk(common::ByteStream &pages_data, bool write_header,
bool write_statistic, bool write_data_to_chunk_data);
- FORCE_INLINE common::ByteStream &get_time_data() {
- return time_out_stream_;
+ FORCE_INLINE common::ByteStream &get_col_notnull_bitmap_data() {
+ return col_notnull_bitmap_out_stream_;
}
FORCE_INLINE common::ByteStream &get_value_data() {
return value_out_stream_;
}
FORCE_INLINE Statistic *get_statistic() { return statistic_; }
- PageData get_cur_page_data() { return cur_page_data_; }
+ ValuePageData get_cur_page_data() { return cur_page_data_; }
void destroy_page_data() { cur_page_data_.destroy(); }
private:
FORCE_INLINE int prepare_end_page() {
int ret = common::E_OK;
- if (RET_FAIL(time_encoder_->flush(time_out_stream_))) {
- } else if (RET_FAIL(value_encoder_->flush(value_out_stream_))) {
+ if (RET_FAIL(value_encoder_->flush(value_out_stream_))) {
+ }
+ for (auto col_notnull_bitmap_byte : col_notnull_bitmap_) {
+ col_notnull_bitmap_out_stream_.write_buf(&col_notnull_bitmap_byte,
+ 1);
}
return ret;
}
@@ -170,21 +171,23 @@ class PageWriter {
common::ByteStream &pages_data);
private:
- // static const uint32_t OUT_STREAM_PAGE_SIZE = 48;
static const uint32_t OUT_STREAM_PAGE_SIZE = 1024;
private:
common::TSDataType data_type_;
- Encoder *time_encoder_;
Encoder *value_encoder_;
Statistic *statistic_;
- common::ByteStream time_out_stream_;
+ common::ByteStream col_notnull_bitmap_out_stream_;
common::ByteStream value_out_stream_;
- PageData cur_page_data_;
+ ValuePageData cur_page_data_;
Compressor *compressor_;
bool is_inited_;
+ std::vector<uint8_t> col_notnull_bitmap_;
+ uint32_t size_;
+
+ static uint32_t MASK;
};
} // end namespace storage
-#endif // WRITER_PAGE_WRITER_H
+#endif // WRITER_PAGE_VALUE_WRITER_H
diff --git a/cpp/test/common/allocator/alloc_base_test.cc b/cpp/test/common/allocator/alloc_base_test.cc
index 9f22ba8f..525f5baf 100644
--- a/cpp/test/common/allocator/alloc_base_test.cc
+++ b/cpp/test/common/allocator/alloc_base_test.cc
@@ -61,4 +61,36 @@ TEST(AllocBaseTest, AllocLargeSize) {
mem_free(ptr);
}
+TEST(AllocBaseTest, ReallocateToLargerSize) {
+ const uint32_t initial_size = 1000;
+ const uint32_t new_size = 2000;
+
+ void *ptr = mem_alloc(initial_size, MOD_DEFAULT);
+ ASSERT_NE(ptr, nullptr);
+
+ ptr = mem_realloc(ptr, new_size);
+ ASSERT_NE(ptr, nullptr);
+
+ uint32_t *header = (uint32_t *)((char *)ptr - HEADER_SIZE_4B);
+ ASSERT_EQ(*header >> 8, new_size);
+
+ mem_free(ptr);
+}
+
+TEST(AllocBaseTest, ReallocateToSmallerSize) {
+ const uint32_t initial_size = 2000;
+ const uint32_t new_size = 1000;
+
+ void *ptr = mem_alloc(initial_size, MOD_DEFAULT);
+ ASSERT_NE(ptr, nullptr);
+
+ ptr = mem_realloc(ptr, new_size);
+ ASSERT_NE(ptr, nullptr);
+
+ uint32_t *header = (uint32_t *)((char *)ptr - HEADER_SIZE_4B);
+ ASSERT_EQ(*header >> 8, new_size);
+
+ mem_free(ptr);
+}
+
} // namespace common
\ No newline at end of file
diff --git a/cpp/test/common/statistic_test.cc b/cpp/test/common/statistic_test.cc
index 1b098ecf..ec52c099 100644
--- a/cpp/test/common/statistic_test.cc
+++ b/cpp/test/common/statistic_test.cc
@@ -199,4 +199,27 @@ TEST(DoubleStatisticTest, BasicFunctionality) {
EXPECT_DOUBLE_EQ(stat_deserialized.last_value_, stat.last_value_);
}
+TEST(TimeStatisticTest, BasicFunctionality) {
+ TimeStatistic stat;
+ EXPECT_EQ(stat.count_, 0);
+ EXPECT_EQ(stat.start_time_, 0);
+ EXPECT_EQ(stat.end_time_, 0);
+
+ stat.update(1000);
+ stat.update(2000);
+
+ EXPECT_EQ(stat.count_, 2);
+ EXPECT_EQ(stat.start_time_, 1000);
+ EXPECT_EQ(stat.end_time_, 2000);
+
+ common::ByteStream out(1024, common::MOD_STATISTIC_OBJ);
+ stat.serialize_typed_stat(out);
+
+ TimeStatistic stat_deserialized;
+ EXPECT_EQ(stat_deserialized.deserialize_typed_stat(out), common::E_OK);
+ EXPECT_EQ(stat.count_, 2);
+ EXPECT_EQ(stat.start_time_, 1000);
+ EXPECT_EQ(stat.end_time_, 2000);
+}
+
} // namespace storage
\ No newline at end of file
diff --git a/cpp/test/writer/chunk_writer_test.cc b/cpp/test/writer/chunk_writer_test.cc
index c43853e0..55e7d630 100644
--- a/cpp/test/writer/chunk_writer_test.cc
+++ b/cpp/test/writer/chunk_writer_test.cc
@@ -31,23 +31,23 @@ using namespace common;
class ChunkWriterTest : public ::testing::Test {
protected:
- ChunkWriter chunkWriter;
- ColumnDesc colDesc;
+ ChunkWriter chunk_writer;
+ ColumnDesc col_desc;
void SetUp() override {
- colDesc.column_name_ = "test_measurement";
- colDesc.type_ = TSDataType::DOUBLE;
- colDesc.encoding_ = TSEncoding::PLAIN;
- colDesc.compression_ = CompressionType::UNCOMPRESSED;
+ col_desc.column_name_ = "test_measurement";
+ col_desc.type_ = TSDataType::DOUBLE;
+ col_desc.encoding_ = TSEncoding::PLAIN;
+ col_desc.compression_ = CompressionType::UNCOMPRESSED;
- ASSERT_EQ(chunkWriter.init(colDesc), E_OK);
+ ASSERT_EQ(chunk_writer.init(col_desc), E_OK);
}
- void TearDown() override { chunkWriter.destroy(); }
+ void TearDown() override { chunk_writer.destroy(); }
};
TEST_F(ChunkWriterTest, InitWithColumnDesc) {
- EXPECT_EQ(chunkWriter.init(colDesc), E_OK);
+ EXPECT_EQ(chunk_writer.init(col_desc), E_OK);
}
TEST_F(ChunkWriterTest, InitWithParameters) {
@@ -59,41 +59,41 @@ TEST_F(ChunkWriterTest, InitWithParameters) {
}
TEST_F(ChunkWriterTest, WriteBoolean) {
- EXPECT_EQ(chunkWriter.write(1234567890, true), E_TYPE_NOT_MATCH);
+ EXPECT_EQ(chunk_writer.write(1234567890, true), E_TYPE_NOT_MATCH);
}
TEST_F(ChunkWriterTest, WriteInt32) {
- EXPECT_EQ(chunkWriter.write(1234567890, int32_t(42)), E_TYPE_NOT_MATCH);
+ EXPECT_EQ(chunk_writer.write(1234567890, int32_t(42)), E_TYPE_NOT_MATCH);
}
TEST_F(ChunkWriterTest, WriteInt64) {
- EXPECT_EQ(chunkWriter.write(1234567890, int64_t(42)), E_TYPE_NOT_MATCH);
+ EXPECT_EQ(chunk_writer.write(1234567890, int64_t(42)), E_TYPE_NOT_MATCH);
}
TEST_F(ChunkWriterTest, WriteFloat) {
- EXPECT_EQ(chunkWriter.write(1234567890, float(42.0)), E_TYPE_NOT_MATCH);
+ EXPECT_EQ(chunk_writer.write(1234567890, float(42.0)), E_TYPE_NOT_MATCH);
}
TEST_F(ChunkWriterTest, WriteDouble) {
- EXPECT_EQ(chunkWriter.write(1234567890, double(42.0)), E_OK);
+ EXPECT_EQ(chunk_writer.write(1234567890, double(42.0)), E_OK);
}
TEST_F(ChunkWriterTest, WriteLargeDataSet) {
for (int i = 0; i < 10000; ++i) {
- chunkWriter.write(i, double(i * 0.1));
+ chunk_writer.write(i, double(i * 0.1));
}
- EXPECT_EQ(chunkWriter.get_chunk_statistic()->count_, 10000);
+ EXPECT_EQ(chunk_writer.get_chunk_statistic()->count_, 10000);
}
TEST_F(ChunkWriterTest, EndEncodeChunk) {
- chunkWriter.write(1234567890, double(42.0));
- EXPECT_EQ(chunkWriter.end_encode_chunk(), E_OK);
- EXPECT_GT(chunkWriter.get_chunk_data().total_size(), 0);
+ chunk_writer.write(1234567890, double(42.0));
+ EXPECT_EQ(chunk_writer.end_encode_chunk(), E_OK);
+ EXPECT_GT(chunk_writer.get_chunk_data().total_size(), 0);
}
TEST_F(ChunkWriterTest, DestroyChunkWriter) {
- chunkWriter.write(1234567890, double(42.0));
- chunkWriter.destroy();
- EXPECT_EQ(chunkWriter.get_chunk_statistic(), nullptr);
- EXPECT_EQ(chunkWriter.get_chunk_data().total_size(), 0);
+ chunk_writer.write(1234567890, double(42.0));
+ chunk_writer.destroy();
+ EXPECT_EQ(chunk_writer.get_chunk_statistic(), nullptr);
+ EXPECT_EQ(chunk_writer.get_chunk_data().total_size(), 0);
}
\ No newline at end of file
diff --git a/cpp/test/writer/page_writer_test.cc b/cpp/test/writer/page_writer_test.cc
index 33c469a5..02189947 100644
--- a/cpp/test/writer/page_writer_test.cc
+++ b/cpp/test/writer/page_writer_test.cc
@@ -22,8 +22,6 @@
#include "common/allocator/byte_stream.h"
#include "common/statistic.h"
-#include "compress/compressor.h"
-#include "encoding/encoder.h"
using namespace storage;
using namespace common;
diff --git a/cpp/test/writer/time_chunk_writer_test.cc b/cpp/test/writer/time_chunk_writer_test.cc
new file mode 100644
index 00000000..0e2b2054
--- /dev/null
+++ b/cpp/test/writer/time_chunk_writer_test.cc
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License a
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include "writer/time_chunk_writer.h"
+
+#include <gtest/gtest.h>
+
+#include "common/allocator/byte_stream.h"
+#include "common/config/config.h"
+#include "common/statistic.h"
+
+using namespace storage;
+using namespace common;
+
+class TimeChunkWriterTest : public ::testing::Test {
+ protected:
+ TimeChunkWriter time_chunk_writer;
+ ColumnDesc col_desc;
+
+ void SetUp() override {
+ col_desc.column_name_ = "time";
+ col_desc.type_ = TSDataType::VECTOR;
+ col_desc.encoding_ = TSEncoding::PLAIN;
+ col_desc.compression_ = CompressionType::UNCOMPRESSED;
+
+ ASSERT_EQ(time_chunk_writer.init(col_desc), E_OK);
+ }
+
+ void TearDown() override { time_chunk_writer.destroy(); }
+};
+
+TEST_F(TimeChunkWriterTest, InitWithColumnDesc) {
+ EXPECT_EQ(time_chunk_writer.init(col_desc), E_OK);
+}
+
+TEST_F(TimeChunkWriterTest, InitWithParameters) {
+ TimeChunkWriter writer;
+ EXPECT_EQ(
+ writer.init("time", TSEncoding::PLAIN, CompressionType::UNCOMPRESSED),
+ E_OK);
+ writer.destroy();
+}
+
+TEST_F(TimeChunkWriterTest, WriteBoolean) {
+ EXPECT_EQ(time_chunk_writer.write(true), E_OK);
+ EXPECT_EQ(time_chunk_writer.write(false), E_OK);
+}
+
+TEST_F(TimeChunkWriterTest, WriteLargeDataSet) {
+ for (int i = 0; i < 10000; ++i) {
+ time_chunk_writer.write(i);
+ }
+ EXPECT_EQ(time_chunk_writer.get_chunk_statistic()->count_, 10000);
+}
+
+TEST_F(TimeChunkWriterTest, EndEncodeChunk) {
+ EXPECT_EQ(time_chunk_writer.write(1234567890), E_OK);
+ EXPECT_EQ(time_chunk_writer.end_encode_chunk(), E_OK);
+ EXPECT_GT(time_chunk_writer.get_chunk_data().total_size(), 0);
+}
+
+TEST_F(TimeChunkWriterTest, DestroyTimeChunkWriter) {
+ time_chunk_writer.write(1234567890);
+ time_chunk_writer.destroy();
+ EXPECT_EQ(time_chunk_writer.get_chunk_statistic(), nullptr);
+ EXPECT_EQ(time_chunk_writer.get_chunk_data().total_size(), 0);
+}
\ No newline at end of file
diff --git a/cpp/test/writer/time_page_writer_test.cc b/cpp/test/writer/time_page_writer_test.cc
new file mode 100644
index 00000000..d0337bee
--- /dev/null
+++ b/cpp/test/writer/time_page_writer_test.cc
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License a
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include "writer/time_page_writer.h"
+
+#include <gtest/gtest.h>
+
+#include "common/allocator/byte_stream.h"
+#include "common/statistic.h"
+#include "encoding/plain_encoder.h"
+
+using namespace storage;
+using namespace common;
+
+class TimePageWriterTest : public ::testing::Test {
+ protected:
+ void SetUp() override {
+ page_writer_ = new TimePageWriter();
+ page_writer_->init(TSEncoding::PLAIN, UNCOMPRESSED);
+ }
+
+ void TearDown() override { delete page_writer_; }
+
+ TimePageWriter* page_writer_;
+};
+
+TEST_F(TimePageWriterTest, WriteSuccess) {
+ TimePageWriter page_writer;
+ page_writer.init(TSEncoding::PLAIN, UNCOMPRESSED);
+ int result = page_writer.write(1234567890);
+ EXPECT_EQ(result, E_OK);
+}
+
+TEST_F(TimePageWriterTest, WriteLargeDataSet) {
+ TimePageWriter page_writer;
+ page_writer.init(TSEncoding::PLAIN, UNCOMPRESSED);
+ for (int i = 0; i < 10000; ++i) {
+ page_writer.write(i);
+ }
+ EXPECT_EQ(page_writer.get_point_numer(), 10000);
+}
+
+TEST_F(TimePageWriterTest, ResetTimePageWriter) {
+ TimePageWriter page_writer;
+ page_writer.init(TSEncoding::PLAIN, UNCOMPRESSED);
+ page_writer.write(1234567890);
+ page_writer.reset();
+ EXPECT_EQ(page_writer.get_point_numer(), 0);
+ EXPECT_EQ(page_writer.get_time_out_stream_size(), 0);
+}
+
+TEST_F(TimePageWriterTest, DestroyTimePageWriter) {
+ TimePageWriter page_writer;
+ page_writer.init(TSEncoding::PLAIN, UNCOMPRESSED);
+ page_writer.write(1234567890);
+ TimeStatistic* stat = (TimeStatistic*)page_writer.get_statistic();
+ EXPECT_NE(stat, nullptr);
+ EXPECT_EQ(stat->count_, 1);
+}
+
+TEST_F(TimePageWriterTest, WritePageHeaderAndData) {
+ EXPECT_EQ(page_writer_->write(1L), common::E_OK);
+ EXPECT_EQ(page_writer_->get_page_memory_size(), 8);
+ EXPECT_EQ(page_writer_->write(2L), common::E_OK);
+ EXPECT_EQ(page_writer_->write(3L), common::E_OK);
+ common::ByteStream byte_stream(1024, common::MOD_DEFAULT);
+ EXPECT_EQ(page_writer_->write_to_chunk(byte_stream, true, true, true),
+ common::E_OK);
+}
\ No newline at end of file
diff --git a/cpp/test/writer/tsfile_writer_test.cc b/cpp/test/writer/tsfile_writer_test.cc
index b934bdce..299575c6 100644
--- a/cpp/test/writer/tsfile_writer_test.cc
+++ b/cpp/test/writer/tsfile_writer_test.cc
@@ -28,6 +28,8 @@
#include "common/tablet.h"
#include "file/tsfile_io_writer.h"
#include "file/write_file.h"
+#include "reader/qds_without_timegenerator.h"
+#include "reader/tsfile_reader.h"
using namespace storage;
using namespace common;
@@ -53,7 +55,7 @@ class TsFileWriterTest : public ::testing::Test {
}
std::string file_name_;
- TsFileWriter* tsfile_writer_ = nullptr;
+ TsFileWriter *tsfile_writer_ = nullptr;
public:
static std::string generate_random_string(int length) {
@@ -74,6 +76,38 @@ class TsFileWriterTest : public ::testing::Test {
return random_string;
}
+
+ static std::string field_to_string(storage::Field *value) {
+ if (value->type_ == common::TEXT) {
+ return std::string(value->value_.sval_);
+ } else {
+ std::stringstream ss;
+ switch (value->type_) {
+ case common::BOOLEAN:
+ ss << (value->value_.bval_ ? "true" : "false");
+ break;
+ case common::INT32:
+ ss << value->value_.ival_;
+ break;
+ case common::INT64:
+ ss << value->value_.lval_;
+ break;
+ case common::FLOAT:
+ ss << value->value_.fval_;
+ break;
+ case common::DOUBLE:
+ ss << value->value_.dval_;
+ break;
+ case common::NULL_TYPE:
+ ss << "NULL";
+ break;
+ default:
+ ASSERT(false);
+ break;
+ }
+ return ss.str();
+ }
+ }
};
TEST_F(TsFileWriterTest, InitWithNullWriteFile) {
@@ -81,6 +115,84 @@ TEST_F(TsFileWriterTest, InitWithNullWriteFile) {
ASSERT_EQ(writer.init(nullptr), E_INVALID_ARG);
}
+TEST_F(TsFileWriterTest, WriteDiffDataType) {
+ std::string device_name = "test_table";
+ common::TSEncoding encoding = common::TSEncoding::PLAIN;
+ common::CompressionType compression_type =
+ common::CompressionType::UNCOMPRESSED;
+ std::vector<std::string> measurement_names = {"level", "num", "bools",
+ "double"};
+ std::vector<common::TSDataType> data_types = {FLOAT, INT64, BOOLEAN,
+ DOUBLE};
+ for (uint32_t i = 0; i < measurement_names.size(); i++) {
+ std::string measurement_name = measurement_names[i];
+ common::TSDataType data_type = data_types[i];
+ tsfile_writer_->register_timeseries(device_name, measurement_name,
+ data_type, encoding,
+ compression_type);
+ }
+
+ int row_num = 1000;
+ for (int i = 0; i < row_num; ++i) {
+ TsRecord record(1622505600000 + i * 100, device_name);
+ for (uint32_t j = 0; j < measurement_names.size(); j++) {
+ std::string measurement_name = measurement_names[j];
+ common::TSDataType data_type = data_types[j];
+ switch (data_type) {
+ case BOOLEAN:
+ record.append_data_point(DataPoint(measurement_name, true));
+ break;
+ case INT64:
+ record.append_data_point(
+ DataPoint(measurement_name, (int64_t)415412));
+ break;
+ case FLOAT:
+ record.append_data_point(
+ DataPoint(measurement_name, (float)1.0));
+ break;
+ case DOUBLE:
+ record.append_data_point(
+ DataPoint(measurement_name, (double)2.0));
+ break;
+ default:
+ break;
+ }
+ }
+ ASSERT_EQ(tsfile_writer_->write_record(record), E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+
+ std::vector<storage::Path> select_list;
+ for (uint32_t i = 0; i < measurement_names.size(); ++i) {
+ std::string measurement_name = measurement_names[i];
+ storage::Path path(device_name, measurement_name);
+ select_list.push_back(path);
+ }
+ storage::QueryExpression *query_expr =
+ storage::QueryExpression::create(select_list, nullptr);
+
+ storage::TsFileReader reader;
+ int ret = reader.open(file_name_);
+ ASSERT_EQ(ret, common::E_OK);
+ storage::QueryDataSet *tmp_qds = nullptr;
+
+ ret = reader.query(query_expr, tmp_qds);
+ auto *qds = (QDSWithoutTimeGenerator *)tmp_qds;
+
+ storage::RowRecord *record;
+ int64_t cur_record_num = 0;
+ do {
+ record = qds->get_next();
+ if (!record) {
+ break;
+ }
+ cur_record_num++;
+ } while (true);
+ EXPECT_EQ(cur_record_num, row_num);
+ reader.destroy_query_data_set(qds);
+}
+
TEST_F(TsFileWriterTest, RegisterTimeSeries) {
std::string device_path = "device1";
std::string measurement_name = "temperature";
@@ -110,7 +222,81 @@ TEST_F(TsFileWriterTest, WriteMultipleRecords) {
DataPoint point(measurement_name, (int32_t)i);
record.append_data_point(point);
ASSERT_EQ(tsfile_writer_->write_record(record), E_OK);
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+}
+
+TEST_F(TsFileWriterTest, WriteMultipleTabletsMultiFlush) {
+ const int device_num = 20;
+ const int measurement_num = 20;
+ int max_tablet_num = 100;
+ std::vector<std::vector<MeasurementSchema>> schema_vecs(
+ device_num, std::vector<MeasurementSchema>(measurement_num));
+ for (int i = 0; i < device_num; i++) {
+ std::string device_name = "test_device" + std::to_string(i);
+ for (int j = 0; j < measurement_num; j++) {
+ std::string measure_name = "measurement" + std::to_string(j);
+ schema_vecs[i][j] =
+ MeasurementSchema(measure_name, common::TSDataType::INT32,
+ common::TSEncoding::PLAIN,
+ common::CompressionType::UNCOMPRESSED);
+ tsfile_writer_->register_timeseries(
+ device_name, measure_name, common::TSDataType::INT32,
+ common::TSEncoding::PLAIN,
+ common::CompressionType::UNCOMPRESSED);
+ }
+ }
+
+ for (int tablet_num = 0; tablet_num < max_tablet_num; tablet_num++) {
+ for (int i = 0; i < device_num; i++) {
+ std::string device_name = "test_device" + std::to_string(i);
+ Tablet tablet(device_name, &schema_vecs[i], 1);
+ tablet.init();
+ for (int j = 0; j < measurement_num; j++) {
+ tablet.set_timestamp(0, 16225600000 + tablet_num * 100);
+ tablet.set_value(0, j, static_cast<int32_t>(tablet_num));
+ }
+ ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+
+ std::vector<storage::Path> select_list;
+ for (int i = 0; i < device_num; i++) {
+ for (int j = 0; j < measurement_num; ++j) {
+ std::string device_name = "test_device" + std::to_string(i);
+ std::string measure_name = "measurement" + std::to_string(j);
+ storage::Path path(device_name, measure_name);
+ select_list.push_back(path);
+ }
+ }
+ storage::QueryExpression *query_expr =
+ storage::QueryExpression::create(select_list, nullptr);
+
+ storage::TsFileReader reader;
+ int ret = reader.open(file_name_);
+ ASSERT_EQ(ret, common::E_OK);
+ storage::QueryDataSet *tmp_qds = nullptr;
+
+ ret = reader.query(query_expr, tmp_qds);
+ auto *qds = (QDSWithoutTimeGenerator *)tmp_qds;
+
+ storage::RowRecord *record;
+ int max_rows = max_tablet_num * 1;
+ for (int cur_row = 0; cur_row < max_rows; cur_row++) {
+ record = qds->get_next();
+ if (!record) {
+ break;
+ }
+ int size = record->get_fields()->size();
+ for (int i = 0; i < size; ++i) {
+ EXPECT_EQ(std::to_string(cur_row),
+ field_to_string(record->get_field(i)));
+ }
}
+ reader.destroy_query_data_set(qds);
}
TEST_F(TsFileWriterTest, WriteMultipleTabletsInt64) {
@@ -175,31 +361,25 @@ TEST_F(TsFileWriterTest, WriteMultipleTabletsDouble) {
for (int i = 0; i < device_num; i++) {
std::string device_name = "test_device" + std::to_string(i);
+ int max_rows = 200;
+ Tablet tablet(device_name, &schema_vec[i], max_rows);
+ tablet.init();
for (int j = 0; j < measurement_num; j++) {
- int max_rows = 200;
- Tablet tablet(device_name, &schema_vec[i], max_rows);
- tablet.init();
for (int row = 0; row < max_rows; row++) {
tablet.set_timestamp(row, 16225600 + row);
}
for (int row = 0; row < max_rows; row++) {
- tablet.set_value(row, j, (double)row + 1.0);
+ tablet.set_value(row, j, static_cast<double>(row) + 1.0);
}
- ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK);
}
+ ASSERT_EQ(tsfile_writer_->write_tablet(tablet), E_OK);
}
ASSERT_EQ(tsfile_writer_->flush(), E_OK);
- tsfile_writer_->close();
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
}
-/*
-// TODO: Flushing without writing after registering a timeseries will cause a
-core
-// dump
-TEST_F(TsFileWriterTest, FlushWithoutWriteAfterRegisterTS) {
- TsFileWriter writer;
- writer.init(file_);
+TEST_F(TsFileWriterTest, FlushWithoutWriteAfterRegisterTS) {
std::string device_path = "device1";
std::string measurement_name = "temperature";
common::TSDataType data_type = common::TSDataType::INT32;
@@ -207,39 +387,209 @@ TEST_F(TsFileWriterTest, FlushWithoutWriteAfterRegisterTS) {
common::CompressionType compression_type =
common::CompressionType::UNCOMPRESSED;
- ASSERT_EQ(writer.register_timeseries(device_path, measurement_name,
- data_type, encoding, compression_type),
+ ASSERT_EQ(tsfile_writer_->register_timeseries(device_path, measurement_name,
+ data_type, encoding,
+ compression_type),
E_OK);
- ASSERT_EQ(writer.flush(), E_OK);
- ASSERT_EQ(writer.close(), E_OK);
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
}
-*/
-TEST_F(TsFileWriterTest, MultiFlush) {
- std::string device_path = "device1";
- std::string measurement_name = "temperature";
+TEST_F(TsFileWriterTest, WriteAlignedTimeseries) {
+ int measurement_num = 100, row_num = 150;
+ std::string device_name = "device";
+ std::vector<std::string> measurement_names;
+ for (int i = 0; i < measurement_num; i++) {
+ measurement_names.emplace_back("temperature" + to_string(i));
+ }
+
common::TSDataType data_type = common::TSDataType::INT32;
common::TSEncoding encoding = common::TSEncoding::PLAIN;
common::CompressionType compression_type =
common::CompressionType::UNCOMPRESSED;
- ASSERT_EQ(tsfile_writer_->register_timeseries(device_path, measurement_name,
- data_type, encoding,
- compression_type),
- E_OK);
- for (int i = 1; i < 2000; i++) {
- TsRecord record(i, device_path);
- DataPoint point(measurement_name, i);
- record.append_data_point(point);
- ASSERT_EQ(tsfile_writer_->write_record(record), E_OK);
+ std::vector<MeasurementSchema *> measurement_schema_vec;
+ for (const auto &measurement_name : measurement_names) {
+ auto *ms = new MeasurementSchema(measurement_name, data_type, encoding,
+ compression_type);
+ measurement_schema_vec.push_back(ms);
}
- ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ tsfile_writer_->register_aligned_timeseries(device_name,
+ measurement_schema_vec);
- for (int i = 2000; i < 4000; i++) {
- TsRecord record(i, device_path);
- DataPoint point(measurement_name, i);
- record.append_data_point(point);
- ASSERT_EQ(tsfile_writer_->write_record(record), E_OK);
+ for (int i = 0; i < row_num; ++i) {
+ TsRecord record(1622505600000 + i * 1000, device_name);
+ for (const auto &measurement_name : measurement_names) {
+ DataPoint point(measurement_name, (int32_t)i);
+ record.append_data_point(point);
+ }
+ ASSERT_EQ(tsfile_writer_->write_record_aligned(record), E_OK);
}
+
ASSERT_EQ(tsfile_writer_->flush(), E_OK);
- tsfile_writer_->close();
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+
+ std::vector<storage::Path> select_list;
+ for (int i = 0; i < measurement_num; ++i) {
+ std::string measurement_name = "temperature" + to_string(i);
+ storage::Path path(device_name, measurement_name);
+ select_list.push_back(path);
+ }
+ storage::QueryExpression *query_expr =
+ storage::QueryExpression::create(select_list, nullptr);
+
+ storage::TsFileReader reader;
+ int ret = reader.open(file_name_);
+ ASSERT_EQ(ret, common::E_OK);
+ storage::QueryDataSet *tmp_qds = nullptr;
+
+ ret = reader.query(query_expr, tmp_qds);
+ auto *qds = (QDSWithoutTimeGenerator *)tmp_qds;
+
+ storage::RowRecord *record;
+ for (int cur_row = 0; cur_row < row_num; cur_row++) {
+ record = qds->get_next();
+ if (!record) {
+ break;
+ }
+ int size = record->get_fields()->size();
+ for (int i = 0; i < size; ++i) {
+ EXPECT_EQ(std::to_string(cur_row),
+ field_to_string(record->get_field(i)));
+ }
+ }
+ reader.destroy_query_data_set(qds);
+}
+
+TEST_F(TsFileWriterTest, WriteAlignedMultiFlush) {
+ int measurement_num = 100, row_num = 100;
+ std::string device_name = "device";
+ std::vector<std::string> measurement_names;
+ for (int i = 0; i < measurement_num; i++) {
+ measurement_names.emplace_back("temperature" + to_string(i));
+ }
+
+ common::TSDataType data_type = common::TSDataType::INT32;
+ common::TSEncoding encoding = common::TSEncoding::PLAIN;
+ common::CompressionType compression_type =
+ common::CompressionType::UNCOMPRESSED;
+ std::vector<MeasurementSchema *> measurement_schema_vec;
+ for (const auto &measurement_name : measurement_names) {
+ auto *ms = new MeasurementSchema(measurement_name, data_type, encoding,
+ compression_type);
+ measurement_schema_vec.push_back(ms);
+ }
+ tsfile_writer_->register_aligned_timeseries(device_name,
+ measurement_schema_vec);
+
+ for (int i = 0; i < row_num; ++i) {
+ TsRecord record(1622505600000 + i * 1000, device_name);
+ for (const auto &measurement_name : measurement_names) {
+ DataPoint point(measurement_name, (int32_t)i);
+ record.append_data_point(point);
+ }
+ ASSERT_EQ(tsfile_writer_->write_record_aligned(record), E_OK);
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ }
+
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+
+ std::vector<storage::Path> select_list;
+ for (int i = 0; i < measurement_num; ++i) {
+ std::string measurement_name = "temperature" + to_string(i);
+ storage::Path path(device_name, measurement_name);
+ select_list.push_back(path);
+ }
+ storage::QueryExpression *query_expr =
+ storage::QueryExpression::create(select_list, nullptr);
+
+ storage::TsFileReader reader;
+ int ret = reader.open(file_name_);
+ ASSERT_EQ(ret, common::E_OK);
+ storage::QueryDataSet *tmp_qds = nullptr;
+
+ ret = reader.query(query_expr, tmp_qds);
+ auto *qds = (QDSWithoutTimeGenerator *)tmp_qds;
+
+ storage::RowRecord *record;
+ for (int cur_row = 0; cur_row < row_num; cur_row++) {
+ record = qds->get_next();
+ if (!record) {
+ break;
+ }
+ int size = record->get_fields()->size();
+ for (int i = 0; i < size; ++i) {
+ EXPECT_EQ(std::to_string(cur_row),
+ field_to_string(record->get_field(i)));
+ }
+ }
+ reader.destroy_query_data_set(qds);
}
+
+TEST_F(TsFileWriterTest, WriteAlignedPartialData) {
+ int measurement_num = 100, row_num = 200;
+ std::string device_name = "device";
+ std::vector<std::string> measurement_names;
+ for (int i = 0; i < measurement_num; i++) {
+ measurement_names.emplace_back("temperature" + to_string(i));
+ }
+
+ common::TSDataType data_type = common::TSDataType::INT32;
+ common::TSEncoding encoding = common::TSEncoding::PLAIN;
+ common::CompressionType compression_type =
+ common::CompressionType::UNCOMPRESSED;
+ std::vector<MeasurementSchema *> measurement_schema_vec;
+ for (const auto &measurement_name : measurement_names) {
+ auto *ms = new MeasurementSchema(measurement_name, data_type, encoding,
+ compression_type);
+ measurement_schema_vec.push_back(ms);
+ }
+ tsfile_writer_->register_aligned_timeseries(device_name,
+ measurement_schema_vec);
+
+ for (int i = 0; i < row_num; ++i) {
+ TsRecord record(1622505600000 + i * 1000, device_name);
+ for (const auto &measurement_name : measurement_names) {
+ DataPoint point(measurement_name, (int32_t)i);
+ if (i % 2 == 0) {
+ point.isnull = true;
+ }
+ record.append_data_point(point);
+ }
+ ASSERT_EQ(tsfile_writer_->write_record_aligned(record), E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+
+ std::vector<storage::Path> select_list;
+ for (int i = 0; i < measurement_num; ++i) {
+ std::string measurement_name = "temperature" + to_string(i);
+ storage::Path path(device_name, measurement_name);
+ select_list.push_back(path);
+ }
+ storage::QueryExpression *query_expr =
+ storage::QueryExpression::create(select_list, nullptr);
+
+ storage::TsFileReader reader;
+ int ret = reader.open(file_name_);
+ ASSERT_EQ(ret, common::E_OK);
+ storage::QueryDataSet *tmp_qds = nullptr;
+
+ ret = reader.query(query_expr, tmp_qds);
+ auto *qds = (QDSWithoutTimeGenerator *)tmp_qds;
+
+ storage::RowRecord *record;
+ int64_t cur_row = 1;
+ do {
+ record = qds->get_next();
+ if (!record) {
+ break;
+ }
+ int size = record->get_fields()->size();
+ for (int i = 0; i < size; ++i) {
+ EXPECT_EQ(std::to_string(cur_row),
+ field_to_string(record->get_field(i)));
+ }
+ cur_row += 2;
+ } while (true);
+ reader.destroy_query_data_set(qds);
+}
\ No newline at end of file
diff --git a/cpp/test/writer/value_chunk_writer_test.cc b/cpp/test/writer/value_chunk_writer_test.cc
new file mode 100644
index 00000000..bc6075fe
--- /dev/null
+++ b/cpp/test/writer/value_chunk_writer_test.cc
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License a
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include "writer/value_chunk_writer.h"
+
+#include <gtest/gtest.h>
+
+#include "common/allocator/byte_stream.h"
+#include "common/config/config.h"
+#include "common/statistic.h"
+
+using namespace storage;
+using namespace common;
+
+class ValueChunkWriterTest : public ::testing::Test {
+ protected:
+ ValueChunkWriter value_chunk_writer;
+ ColumnDesc col_desc;
+
+ void SetUp() override {
+ col_desc.column_name_ = "test_measurement";
+ col_desc.type_ = TSDataType::DOUBLE;
+ col_desc.encoding_ = TSEncoding::PLAIN;
+ col_desc.compression_ = CompressionType::UNCOMPRESSED;
+
+ ASSERT_EQ(value_chunk_writer.init(col_desc), E_OK);
+ }
+
+ void TearDown() override { value_chunk_writer.destroy(); }
+};
+
+TEST_F(ValueChunkWriterTest, InitWithColumnDesc) {
+ EXPECT_EQ(value_chunk_writer.init(col_desc), E_OK);
+}
+
+TEST_F(ValueChunkWriterTest, InitWithParameters) {
+ ValueChunkWriter writer;
+ EXPECT_EQ(writer.init("test_measurement", TSDataType::DOUBLE,
+ TSEncoding::PLAIN, CompressionType::UNCOMPRESSED),
+ E_OK);
+ writer.destroy();
+}
+
+TEST_F(ValueChunkWriterTest, WriteBoolean) {
+ EXPECT_EQ(value_chunk_writer.write(1234567890, true, false),
+ E_TYPE_NOT_MATCH);
+}
+
+TEST_F(ValueChunkWriterTest, WriteInt32) {
+ EXPECT_EQ(value_chunk_writer.write(1234567890, int32_t(42), false),
+ E_TYPE_NOT_MATCH);
+}
+
+TEST_F(ValueChunkWriterTest, WriteInt64) {
+ EXPECT_EQ(value_chunk_writer.write(1234567890, int64_t(42), false),
+ E_TYPE_NOT_MATCH);
+}
+
+TEST_F(ValueChunkWriterTest, WriteFloat) {
+ EXPECT_EQ(value_chunk_writer.write(1234567890, float(42.0), false),
+ E_TYPE_NOT_MATCH);
+}
+
+TEST_F(ValueChunkWriterTest, WriteDouble) {
+ EXPECT_EQ(value_chunk_writer.write(1234567890, double(42.0), false), E_OK);
+}
+
+TEST_F(ValueChunkWriterTest, WriteLargeDataSet) {
+ for (int i = 0; i < 10000; ++i) {
+ value_chunk_writer.write(i, double(i * 0.1), false);
+ }
+ EXPECT_EQ(value_chunk_writer.get_chunk_statistic()->count_, 10000);
+}
+
+TEST_F(ValueChunkWriterTest, EndEncodeChunk) {
+ value_chunk_writer.write(1234567890, double(42.0), false);
+ EXPECT_EQ(value_chunk_writer.end_encode_chunk(), E_OK);
+ EXPECT_GT(value_chunk_writer.get_chunk_data().total_size(), 0);
+}
+
+TEST_F(ValueChunkWriterTest, DestroyChunkWriter) {
+ value_chunk_writer.write(1234567890, double(42.0), false);
+ value_chunk_writer.destroy();
+ EXPECT_EQ(value_chunk_writer.get_chunk_statistic(), nullptr);
+ EXPECT_EQ(value_chunk_writer.get_chunk_data().total_size(), 0);
+}
\ No newline at end of file
diff --git a/cpp/test/writer/value_page_writer_test.cc b/cpp/test/writer/value_page_writer_test.cc
new file mode 100644
index 00000000..e88712b9
--- /dev/null
+++ b/cpp/test/writer/value_page_writer_test.cc
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License a
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include "writer/value_page_writer.h"
+
+#include <gtest/gtest.h>
+
+#include "common/allocator/byte_stream.h"
+#include "common/statistic.h"
+
+using namespace storage;
+using namespace common;
+
+class ValuePageWriterTest : public ::testing::Test {};
+
+TEST_F(ValuePageWriterTest, WriteBooleanSuccess) {
+ ValuePageWriter value_page_writer;
+ value_page_writer.init(TSDataType::BOOLEAN, TSEncoding::PLAIN,
+ UNCOMPRESSED);
+ int result = value_page_writer.write(1234567890, true, false);
+ EXPECT_EQ(result, E_OK);
+}
+
+TEST_F(ValuePageWriterTest, WriteInt32Success) {
+ ValuePageWriter value_page_writer;
+ value_page_writer.init(TSDataType::INT32, TSEncoding::PLAIN, UNCOMPRESSED);
+ int result = value_page_writer.write(1234567890, int32_t(42), false);
+ EXPECT_EQ(result, E_OK);
+}
+
+TEST_F(ValuePageWriterTest, WriteInt64Success) {
+ ValuePageWriter value_page_writer;
+ value_page_writer.init(TSDataType::INT64, TSEncoding::PLAIN, UNCOMPRESSED);
+ int result = value_page_writer.write(1234567890, int64_t(42), false);
+ EXPECT_EQ(result, E_OK);
+}
+
+TEST_F(ValuePageWriterTest, WriteFloatSuccess) {
+ ValuePageWriter value_page_writer;
+ value_page_writer.init(TSDataType::FLOAT, TSEncoding::PLAIN, UNCOMPRESSED);
+ int result = value_page_writer.write(1234567890, float(42.0), false);
+ EXPECT_EQ(result, E_OK);
+}
+
+TEST_F(ValuePageWriterTest, WriteDoubleSuccess) {
+ ValuePageWriter value_page_writer;
+ value_page_writer.init(TSDataType::DOUBLE, TSEncoding::PLAIN, UNCOMPRESSED);
+ int result = value_page_writer.write(1234567890, double(42.0), false);
+ EXPECT_EQ(result, E_OK);
+}
+
+TEST_F(ValuePageWriterTest, WriteLargeDataSet) {
+ ValuePageWriter value_page_writer;
+ value_page_writer.init(TSDataType::DOUBLE, TSEncoding::PLAIN, UNCOMPRESSED);
+ for (int i = 0; i < 10000; ++i) {
+ value_page_writer.write(i, double(i * 0.1), false);
+ }
+ EXPECT_EQ(value_page_writer.get_point_numer(), 10000);
+}
+
+TEST_F(ValuePageWriterTest, ResetValuePageWriter) {
+ ValuePageWriter value_page_writer;
+ value_page_writer.init(TSDataType::INT64, TSEncoding::PLAIN, UNCOMPRESSED);
+ value_page_writer.write(1234567890, int64_t(42), false);
+ value_page_writer.reset();
+ EXPECT_EQ(value_page_writer.get_point_numer(), 0);
+ EXPECT_EQ(value_page_writer.get_col_notnull_bitmap_out_stream_size(), 0);
+}
+
+TEST_F(ValuePageWriterTest, DestroyValuePageWriter) {
+ ValuePageWriter value_page_writer;
+ value_page_writer.init(TSDataType::INT64, TSEncoding::PLAIN, UNCOMPRESSED);
+ value_page_writer.write(1234567890, int64_t(42), false);
+ Int64Statistic* stat = (Int64Statistic*)value_page_writer.get_statistic();
+ EXPECT_TRUE(stat);
+ EXPECT_EQ(stat->count_, 1);
+}
+
+TEST_F(ValuePageWriterTest, WritePageHeaderAndData) {
+ ValuePageWriter value_page_writer;
+ value_page_writer.init(TSDataType::INT64, TSEncoding::PLAIN, UNCOMPRESSED);
+ EXPECT_EQ(value_page_writer.write(1234567890, (int64_t)1, false),
+ common::E_OK);
+ EXPECT_EQ(value_page_writer.get_page_memory_size(), 8);
+ EXPECT_EQ(value_page_writer.write(1234567891, (int64_t)2, false),
+ common::E_OK);
+ EXPECT_EQ(value_page_writer.write(1234567892, (int64_t)3, false),
+ common::E_OK);
+ common::ByteStream byte_stream(1024, common::MOD_DEFAULT);
+ EXPECT_EQ(value_page_writer.write_to_chunk(byte_stream, true, true, true),
+ common::E_OK);
+}
(tsfile) 03/03: change version
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch iotdb
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 999a6a217e37548915c3c06b7dc79033b4890f85
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Thu Aug 29 09:16:32 2024 +0800
change version
---
cpp/pom.xml | 2 +-
java/common/pom.xml | 2 +-
java/examples/pom.xml | 4 ++--
java/pom.xml | 4 ++--
java/tools/pom.xml | 6 +++---
java/tsfile/pom.xml | 4 ++--
pom.xml | 2 +-
python/pom.xml | 2 +-
8 files changed, 13 insertions(+), 13 deletions(-)
diff --git a/cpp/pom.xml b/cpp/pom.xml
index f5a98d9a..56bf94e2 100644
--- a/cpp/pom.xml
+++ b/cpp/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-parent</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
</parent>
<artifactId>tsfile-cpp</artifactId>
<packaging>pom</packaging>
diff --git a/java/common/pom.xml b/java/common/pom.xml
index 3c23f331..82b44cee 100644
--- a/java/common/pom.xml
+++ b/java/common/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-java</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
</parent>
<artifactId>common</artifactId>
<name>TsFile: Java: Common</name>
diff --git a/java/examples/pom.xml b/java/examples/pom.xml
index 8671742f..f34bbd28 100644
--- a/java/examples/pom.xml
+++ b/java/examples/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-java</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
</parent>
<artifactId>examples</artifactId>
<packaging>pom</packaging>
@@ -38,7 +38,7 @@
<dependency>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
</dependency>
</dependencies>
<build>
diff --git a/java/pom.xml b/java/pom.xml
index de0b0c3c..094585de 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -24,10 +24,10 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-parent</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
</parent>
<artifactId>tsfile-java</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
<packaging>pom</packaging>
<name>TsFile: Java</name>
<modules>
diff --git a/java/tools/pom.xml b/java/tools/pom.xml
index 82745fcc..0e77a1ab 100644
--- a/java/tools/pom.xml
+++ b/java/tools/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-java</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
</parent>
<artifactId>tools</artifactId>
<name>TsFile: Java: Tools</name>
@@ -32,7 +32,7 @@
<dependency>
<groupId>org.apache.tsfile</groupId>
<artifactId>common</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>commons-cli</groupId>
@@ -50,7 +50,7 @@
<dependency>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
diff --git a/java/tsfile/pom.xml b/java/tsfile/pom.xml
index daed5fbb..ffe1eb8f 100644
--- a/java/tsfile/pom.xml
+++ b/java/tsfile/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-java</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
</parent>
<artifactId>tsfile</artifactId>
<name>TsFile: Java: TsFile</name>
@@ -38,7 +38,7 @@
<dependency>
<groupId>org.apache.tsfile</groupId>
<artifactId>common</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.github.luben</groupId>
diff --git a/pom.xml b/pom.xml
index 6fbedfba..87863651 100644
--- a/pom.xml
+++ b/pom.xml
@@ -28,7 +28,7 @@
</parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-parent</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Apache TsFile Project Parent POM</name>
<properties>
diff --git a/python/pom.xml b/python/pom.xml
index f228492e..b02d1d4a 100644
--- a/python/pom.xml
+++ b/python/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.tsfile</groupId>
<artifactId>tsfile-parent</artifactId>
- <version>1.2.0-20a94fb3-SNAPSHOT</version>
+ <version>1.2.0-efc8f71d-SNAPSHOT</version>
</parent>
<artifactId>tsfile-python</artifactId>
<packaging>pom</packaging>
(tsfile) 02/03: Fix allSatisfy bug in InFilter
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch iotdb
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit efc8f71d8f903a666a622cbb71b81e4dc6a32ea0
Author: Jackie Tien <ja...@gmail.com>
AuthorDate: Thu Aug 29 09:06:44 2024 +0800
Fix allSatisfy bug in InFilter
---
java/tsfile/src/main/codegen/templates/FilterOperatorsTemplate.ftl | 5 +++++
1 file changed, 5 insertions(+)
diff --git a/java/tsfile/src/main/codegen/templates/FilterOperatorsTemplate.ftl b/java/tsfile/src/main/codegen/templates/FilterOperatorsTemplate.ftl
index 8bc34671..b6f7a4c0 100644
--- a/java/tsfile/src/main/codegen/templates/FilterOperatorsTemplate.ftl
+++ b/java/tsfile/src/main/codegen/templates/FilterOperatorsTemplate.ftl
@@ -1035,6 +1035,11 @@ public final class ${className} {
return true;
}
+ // has null value, just return false
+ if (metadata.hasNullValue(measurementIndex)) {
+ return false;
+ }
+
// All values are same
if (statistics.isPresent()) {
Statistics<? extends Serializable> stat = statistics.get();