diff --git a/cpp/src/common/schema.h b/cpp/src/common/schema.h index feb2eb799..c10527c35 100644 --- a/cpp/src/common/schema.h +++ b/cpp/src/common/schema.h @@ -103,6 +103,7 @@ struct MeasurementSchema { if (ret == common::E_OK) { if (RET_FAIL(common::SerializationUtil::write_ui32(props_.size(), out))) { + } else { for (const auto& prop : props_) { if (RET_FAIL(common::SerializationUtil::write_str( prop.first, out))) { diff --git a/cpp/src/reader/block/single_device_tsblock_reader.cc b/cpp/src/reader/block/single_device_tsblock_reader.cc index 20431c481..9d6c63fba 100644 --- a/cpp/src/reader/block/single_device_tsblock_reader.cc +++ b/cpp/src/reader/block/single_device_tsblock_reader.cc @@ -20,6 +20,7 @@ #include "single_device_tsblock_reader.h" #include +#include #include #include @@ -230,7 +231,7 @@ int SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task, bool used_multi = false; std::set multi_names; { - bool can_multi = !time_series_indexs.empty(); + bool can_multi = field_filter == nullptr && !time_series_indexs.empty(); auto& meas_cols = device_query_task->get_column_mapping()->get_measurement_columns(); for (const auto& ts_idx : time_series_indexs) { @@ -365,7 +366,8 @@ int SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task, } // Detect aligned fast path: every field column comes from an aligned chunk. - if (!field_column_contexts_.empty() && dense_row_count_ >= 0 && + if (field_filter_ == nullptr && !field_column_contexts_.empty() && + dense_row_count_ >= 0 && aligned_col_count_ == field_column_contexts_.size()) { all_aligned_ = true; aligned_vec_.reserve(field_column_contexts_.size()); @@ -636,8 +638,22 @@ int SingleDeviceTsBlockReader::has_next_aligned(bool& result_has_next) { int SingleDeviceTsBlockReader::fill_measurements( std::vector& column_contexts) { int ret = common::E_OK; - if (field_filter_ == - nullptr /*TODO: || field_filter_->satisfy(column_contexts)*/) { + bool row_satisfies = true; + if (field_filter_ != nullptr) { + for (auto* column_context : column_contexts) { + auto* single_context = + dynamic_cast(column_context); + if (single_context == nullptr) return common::E_NOT_SUPPORT; + bool column_satisfies = false; + int filter_ret = single_context->satisfy_field_filter( + field_filter_, next_time_, column_satisfies); + if (RET_FAIL(filter_ret)) return filter_ret; + row_satisfies = row_satisfies && column_satisfies; + if (!row_satisfies) break; + } + } + + if (row_satisfies) { row_appender_->add_row(); if (!col_appenders_[time_column_index_]->add_row()) { assert(false); @@ -652,11 +668,8 @@ int SingleDeviceTsBlockReader::fill_measurements( col_appenders_[time_in_query_index]->append( (const char*)&next_time_, sizeof(next_time_)); } - for (auto& column_context : column_contexts) { + for (auto* column_context : column_contexts) { column_context->fill_into(col_appenders_); - if (RET_FAIL(advance_column(column_context))) { - break; - } } // Align all columns, filling with nulls where data is missing. @@ -674,6 +687,11 @@ int SingleDeviceTsBlockReader::fill_measurements( } } } + + for (auto* column_context : column_contexts) { + ret = advance_column(column_context); + if (RET_FAIL(ret)) break; + } return ret; } @@ -896,6 +914,43 @@ int SingleMeasurementColumnContext::get_current_value(char*& value, return common::E_OK; } +int SingleMeasurementColumnContext::satisfy_field_filter( + Filter* field_filter, int64_t time, bool& satisfies) { + char* value = nullptr; + uint32_t len = 0; + int ret = get_current_value(value, len); + if (ret != common::E_OK) return ret; + if (value == nullptr) { + satisfies = false; + return common::E_OK; + } + + switch (value_iter_->get_data_type()) { + case common::INT32: { + int32_t typed_value; + if (len != sizeof(typed_value)) return common::E_INVALID_ARG; + std::memcpy(&typed_value, value, sizeof(typed_value)); + satisfies = field_filter->satisfy(time, typed_value); + return common::E_OK; + } + case common::INT64: { + int64_t typed_value; + if (len != sizeof(typed_value)) return common::E_INVALID_ARG; + std::memcpy(&typed_value, value, sizeof(typed_value)); + satisfies = field_filter->satisfy(time, typed_value); + return common::E_OK; + } + case common::STRING: + case common::TEXT: + case common::BLOB: + satisfies = field_filter->satisfy(time, + common::String(value, len)); + return common::E_OK; + default: + return common::E_NOT_SUPPORT; + } +} + int SingleMeasurementColumnContext::move_iter() { int ret = common::E_OK; time_iter_->next(); diff --git a/cpp/src/reader/block/single_device_tsblock_reader.h b/cpp/src/reader/block/single_device_tsblock_reader.h index 4bf0cca0f..bb73a103d 100644 --- a/cpp/src/reader/block/single_device_tsblock_reader.h +++ b/cpp/src/reader/block/single_device_tsblock_reader.h @@ -167,6 +167,8 @@ class SingleMeasurementColumnContext : public MeasurementColumnContext { int get_next_tsblock(bool alloc_mem) override; int get_current_time(int64_t& time) override; int get_current_value(char*& value, uint32_t& len) override; + int satisfy_field_filter(Filter* field_filter, int64_t time, + bool& satisfies); int move_iter() override; uint32_t available_rows() const override; int bulk_copy_into(std::vector& col_appenders, diff --git a/cpp/src/writer/chunk_writer.cc b/cpp/src/writer/chunk_writer.cc index acdb4951d..5b55a0d40 100644 --- a/cpp/src/writer/chunk_writer.cc +++ b/cpp/src/writer/chunk_writer.cc @@ -31,13 +31,15 @@ int ChunkWriter::init(const ColumnSchema& col_schema) { } int ChunkWriter::init(const std::string& measurement_name, TSDataType data_type, - TSEncoding encoding, CompressionType compression_type) { + TSEncoding encoding, CompressionType compression_type, + int max_point_number) { 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))) { + page_writer_.init(data_type, encoding, compression_type, + max_point_number))) { } else if (IS_NULL((first_page_statistic_ = StatisticFactory::alloc_statistic(data_type)))) { ret = E_OOM; diff --git a/cpp/src/writer/chunk_writer.h b/cpp/src/writer/chunk_writer.h index 3d0d6dd32..11c600165 100644 --- a/cpp/src/writer/chunk_writer.h +++ b/cpp/src/writer/chunk_writer.h @@ -58,7 +58,8 @@ class ChunkWriter { int init(const common::ColumnSchema& col_schema); int init(const std::string& measurement_name, common::TSDataType data_type, common::TSEncoding encoding, - common::CompressionType compression_type); + common::CompressionType compression_type, + int max_point_number = -1); void reset(); void destroy(); diff --git a/cpp/src/writer/page_writer.cc b/cpp/src/writer/page_writer.cc index eebe5b400..30b2eef06 100644 --- a/cpp/src/writer/page_writer.cc +++ b/cpp/src/writer/page_writer.cc @@ -23,6 +23,7 @@ #include "common/logger/elog.h" #include "compress/compressor_factory.h" #include "encoding/encoder_factory.h" +#include "encoding/ts2diff_encoder.h" using namespace common; @@ -75,7 +76,7 @@ int PageData::init(ByteStream& time_bs, ByteStream& value_bs, /* ================ PageWriter ================ */ int PageWriter::init(TSDataType data_type, TSEncoding encoding, - CompressionType compression) { + CompressionType compression, int max_point_number) { int ret = E_OK; data_type_ = data_type; if (nullptr == (time_encoder_ = EncoderFactory::alloc_time_encoder())) { @@ -83,12 +84,24 @@ int PageWriter::init(TSDataType data_type, TSEncoding encoding, } else if (nullptr == (value_encoder_ = EncoderFactory::alloc_value_encoder( encoding, data_type))) { ret = E_OOM; - } else if (nullptr == - (statistic_ = StatisticFactory::alloc_statistic(data_type))) { - ret = E_OOM; - } else if (nullptr == (compressor_ = CompressorFactory::alloc_compressor( - compression))) { - ret = E_OOM; + } else { + if (encoding == TS_2DIFF && max_point_number >= 0) { + if (data_type == FLOAT) { + static_cast(value_encoder_) + ->set_max_point_number(max_point_number); + } else if (data_type == DOUBLE) { + static_cast(value_encoder_) + ->set_max_point_number(max_point_number); + } + } + if (nullptr == + (statistic_ = StatisticFactory::alloc_statistic(data_type))) { + ret = E_OOM; + } else if (nullptr == + (compressor_ = CompressorFactory::alloc_compressor( + compression))) { + ret = E_OOM; + } } if (ret != E_OK) { if (time_encoder_ != nullptr) { diff --git a/cpp/src/writer/page_writer.h b/cpp/src/writer/page_writer.h index 47c958913..8ae7efbb7 100644 --- a/cpp/src/writer/page_writer.h +++ b/cpp/src/writer/page_writer.h @@ -103,7 +103,7 @@ class PageWriter { compressor_(nullptr), is_inited_(false) {} int init(common::TSDataType data_type, common::TSEncoding encoding, - common::CompressionType compression); + common::CompressionType compression, int max_point_number = -1); ~PageWriter() { destroy(); } // reset statistic_, time_out_stream_, value_out_stream_ void reset(); diff --git a/cpp/src/writer/tsfile_writer.cc b/cpp/src/writer/tsfile_writer.cc index ed3e09be7..3de047403 100644 --- a/cpp/src/writer/tsfile_writer.cc +++ b/cpp/src/writer/tsfile_writer.cc @@ -20,6 +20,7 @@ #include "tsfile_writer.h" #include +#include #include #include #include @@ -533,9 +534,14 @@ int TsFileWriter::do_check_schema( MeasurementSchema* ms = ms_iter->second; if (IS_NULL(ms->chunk_writer_)) { ms->chunk_writer_ = new ChunkWriter; - ret = ms->chunk_writer_->init(ms->measurement_name_, - ms->data_type_, ms->encoding_, - ms->compression_type_); + int max_point_number = -1; + auto property = ms->props_.find("max_point_number"); + if (property != ms->props_.end()) { + max_point_number = std::atoi(property->second.c_str()); + } + ret = ms->chunk_writer_->init( + ms->measurement_name_, ms->data_type_, ms->encoding_, + ms->compression_type_, max_point_number); if (IS_SUCC(ret)) { chunk_writers.push_back(ms->chunk_writer_); } else { @@ -567,9 +573,15 @@ int TsFileWriter::ensure_aligned_value_chunk_writer( if (IS_NULL(value_chunk_writer)) { return E_OOM; } + int max_point_number = -1; + auto property = measurement_schema->props_.find("max_point_number"); + if (property != measurement_schema->props_.end()) { + max_point_number = std::atoi(property->second.c_str()); + } int ret = value_chunk_writer->init( measurement_schema->measurement_name_, measurement_schema->data_type_, - measurement_schema->encoding_, measurement_schema->compression_type_); + measurement_schema->encoding_, measurement_schema->compression_type_, + max_point_number); if (RET_FAIL(ret)) { delete value_chunk_writer; return (ret == E_OOM) ? ret : common::E_INVALID_ARG; @@ -694,9 +706,14 @@ int TsFileWriter::do_check_schema_table( MeasurementSchema* ms = ms_iter->second; if (IS_NULL(ms->value_chunk_writer_)) { ms->value_chunk_writer_ = new ValueChunkWriter; + int max_point_number = -1; + auto property = ms->props_.find("max_point_number"); + if (property != ms->props_.end()) { + max_point_number = std::atoi(property->second.c_str()); + } ret = ms->value_chunk_writer_->init( ms->measurement_name_, ms->data_type_, ms->encoding_, - ms->compression_type_); + ms->compression_type_, max_point_number); if (IS_SUCC(ret)) { value_chunk_writers.push_back(ms->value_chunk_writer_); } else { diff --git a/cpp/src/writer/value_chunk_writer.cc b/cpp/src/writer/value_chunk_writer.cc index 182471c71..1dd2b2117 100644 --- a/cpp/src/writer/value_chunk_writer.cc +++ b/cpp/src/writer/value_chunk_writer.cc @@ -32,13 +32,15 @@ int ValueChunkWriter::init(const ColumnSchema& col_schema) { int ValueChunkWriter::init(const std::string& measurement_name, TSDataType data_type, TSEncoding encoding, - CompressionType compression_type) { + CompressionType compression_type, + int max_point_number) { int ret = E_OK; chunk_statistic_ = StatisticFactory::alloc_statistic(data_type); if (chunk_statistic_ == nullptr) { return E_OOM; } else if (RET_FAIL(value_page_writer_.init(data_type, encoding, - compression_type))) { + compression_type, + max_point_number))) { } else if (IS_NULL((first_page_statistic_ = StatisticFactory::alloc_statistic(data_type)))) { ret = E_OOM; diff --git a/cpp/src/writer/value_chunk_writer.h b/cpp/src/writer/value_chunk_writer.h index 88871c1a9..de28ba3c8 100644 --- a/cpp/src/writer/value_chunk_writer.h +++ b/cpp/src/writer/value_chunk_writer.h @@ -58,7 +58,8 @@ class ValueChunkWriter { int init(const common::ColumnSchema& col_schema); int init(const std::string& measurement_name, common::TSDataType data_type, common::TSEncoding encoding, - common::CompressionType compression_type); + common::CompressionType compression_type, + int max_point_number = -1); void reset(); void destroy(); diff --git a/cpp/src/writer/value_page_writer.cc b/cpp/src/writer/value_page_writer.cc index 7e9b09db7..3832f8d43 100644 --- a/cpp/src/writer/value_page_writer.cc +++ b/cpp/src/writer/value_page_writer.cc @@ -23,6 +23,7 @@ #include "common/logger/elog.h" #include "compress/compressor_factory.h" #include "encoding/encoder_factory.h" +#include "encoding/ts2diff_encoder.h" using namespace common; @@ -86,17 +87,30 @@ int ValuePageData::init(ByteStream& col_notnull_bitmap_bs, ByteStream& value_bs, } int ValuePageWriter::init(TSDataType data_type, TSEncoding encoding, - CompressionType compression) { + CompressionType compression, int max_point_number) { int ret = E_OK; data_type_ = 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))) { - ret = E_OOM; - } else if (nullptr == (compressor_ = CompressorFactory::alloc_compressor( - compression))) { + } else { + if (encoding == TS_2DIFF && max_point_number >= 0) { + if (data_type == FLOAT) { + static_cast(value_encoder_) + ->set_max_point_number(max_point_number); + } else if (data_type == DOUBLE) { + static_cast(value_encoder_) + ->set_max_point_number(max_point_number); + } + } + if (nullptr == + (statistic_ = StatisticFactory::alloc_statistic(data_type))) { + ret = E_OOM; + } + } + if (ret == E_OK && nullptr == + (compressor_ = CompressorFactory::alloc_compressor( + compression))) { ret = E_OOM; } if (ret != E_OK) { diff --git a/cpp/src/writer/value_page_writer.h b/cpp/src/writer/value_page_writer.h index 61bf59564..0f280a9d5 100644 --- a/cpp/src/writer/value_page_writer.h +++ b/cpp/src/writer/value_page_writer.h @@ -99,7 +99,7 @@ class ValuePageWriter { size_(0) {} ~ValuePageWriter() { destroy(); } int init(common::TSDataType data_type, common::TSEncoding encoding, - common::CompressionType compression); + common::CompressionType compression, int max_point_number = -1); void reset(); void destroy(); diff --git a/cpp/test/common/schema_test.cc b/cpp/test/common/schema_test.cc index 4308bc4a9..f7c0f210a 100644 --- a/cpp/test/common/schema_test.cc +++ b/cpp/test/common/schema_test.cc @@ -52,6 +52,23 @@ TEST(MeasurementSchemaTest, JavaCppGap) { } #endif +TEST(MeasurementSchemaTest, SerializesProperties) { + MeasurementSchema original("temperature", common::FLOAT, + common::TS_2DIFF, common::UNCOMPRESSED); + original.props_["max_point_number"] = "0"; + + common::ByteStream encoded(1024, common::MOD_DEFAULT); + ASSERT_EQ(common::E_OK, original.serialize_to(encoded)); + std::vector bytes(encoded.total_size()); + common::copy_bs_to_buf(encoded, bytes.data(), bytes.size()); + + common::ByteStream input(1024, common::MOD_DEFAULT); + input.wrap_from(bytes.data(), bytes.size()); + MeasurementSchema decoded; + ASSERT_EQ(common::E_OK, decoded.deserialize_from(input)); + EXPECT_EQ(original.props_, decoded.props_); +} + TEST(MeasurementSchemaTest, ParameterizedConstructor) { MeasurementSchema schema("test_measurement", common::TSDataType::BOOLEAN, common::TSEncoding::RLE, diff --git a/cpp/test/reader/table_view/tsfile_reader_table_test.cc b/cpp/test/reader/table_view/tsfile_reader_table_test.cc index dfbff0576..42a35b829 100644 --- a/cpp/test/reader/table_view/tsfile_reader_table_test.cc +++ b/cpp/test/reader/table_view/tsfile_reader_table_test.cc @@ -212,8 +212,62 @@ class TsFileTableReaderTest : public ::testing::Test { } }; +class ConstantFieldFilter final : public Filter { + public: + ConstantFieldFilter(int* calls, bool accept) + : calls_(calls), accept_(accept) {} + bool satisfy(int64_t, int64_t) override { + ++*calls_; + return accept_; + } + bool satisfy(int64_t, common::String) override { + ++*calls_; + return accept_; + } + + private: + int* calls_; + bool accept_; +}; + TEST_F(TsFileTableReaderTest, TableModelQuery) { test_table_model_query(); } +TEST_F(TsFileTableReaderTest, FieldFilterControlsRowsAndIsInvoked) { + auto table_schema = gen_table_schema(0); + TsFileTableWriter writer(&write_file_, table_schema); + auto tablet = gen_tablet(table_schema, 0, 1, 1); + ASSERT_EQ(writer.write_table(tablet), common::E_OK); + ASSERT_EQ(writer.flush(), common::E_OK); + ASSERT_EQ(writer.close(), common::E_OK); + + LocalRandomAccessReadFile read_file; + ASSERT_EQ(read_file.open(file_name_), common::E_OK); + TableQueryExecutor executor(&read_file); + for (int mode : {0, 1, 2}) { + int filter_calls = 0; + Filter* field_filter = mode == 0 + ? nullptr + : new ConstantFieldFilter(&filter_calls, + mode == 1); + ResultSet* result = nullptr; + ASSERT_EQ(executor.query(table_schema->get_table_name(), {"s0"}, + nullptr, nullptr, field_filter, result), + common::E_OK); + ASSERT_NE(result, nullptr); + int rows = 0; + bool has_next = false; + int ret = common::E_OK; + while ((ret = result->next(has_next)) == common::E_OK && has_next) { + ++rows; + } + EXPECT_EQ(ret, common::E_OK); + EXPECT_EQ(filter_calls, mode == 0 ? 0 : 1); + EXPECT_EQ(rows, mode == 2 ? 0 : 1); + executor.destroy_query_data_set(result); + } + delete table_schema; +} + class TsFileTableReaderTimeFilterTest : public TsFileTableReaderTest, public ::testing::WithParamInterface { diff --git a/cpp/test/writer/tsfile_writer_test.cc b/cpp/test/writer/tsfile_writer_test.cc index 1727fa5b8..7a78ee0e7 100644 --- a/cpp/test/writer/tsfile_writer_test.cc +++ b/cpp/test/writer/tsfile_writer_test.cc @@ -285,6 +285,37 @@ TEST_F(TsFileWriterTest, RegisterTimeSeries) { ASSERT_EQ(tsfile_writer_->close(), E_OK); } +TEST_F(TsFileWriterTest, AppliesMaxPointNumberPropertyToTs2DiffWriter) { + const std::string device = "root.precision.d"; + auto* schema = new MeasurementSchema( + "v", TSDataType::FLOAT, TSEncoding::TS_2DIFF, + CompressionType::UNCOMPRESSED); + schema->props_["max_point_number"] = "0"; + tsfile_writer_->set_generate_table_schema(false); + std::vector schemas = {schema}; + ASSERT_EQ(tsfile_writer_->register_timeseries(device, schemas), E_OK); + + TsRecord record(1, device); + record.add_point("v", 1.23456f); + ASSERT_EQ(tsfile_writer_->write_record(record), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + TsFileReader reader; + ASSERT_EQ(reader.open(file_name_), E_OK); + ResultSet* result = nullptr; + std::vector select_list = {device + ".v"}; + ASSERT_EQ(reader.query(select_list, 1, 2, result), E_OK); + auto* qds = static_cast(result); + bool has_next = false; + ASSERT_EQ(qds->next(has_next), E_OK); + ASSERT_TRUE(has_next); + EXPECT_FLOAT_EQ(qds->get_value(2), 1.0f); + ASSERT_EQ(qds->next(has_next), E_OK); + EXPECT_FALSE(has_next); + reader.destroy_query_data_set(qds); + ASSERT_EQ(reader.close(), E_OK); +} + TEST_F(TsFileWriterTest, MultiFlushWriteAndRead) { std::string device_path = "device1"; std::string measurement_name = "temperature";