Fix C++ schema properties, TS_2DIFF precision, and field filters

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))) {

cpp/src/reader/block/single_device_tsblock_reader.cc

@@ -20,6 +20,7 @@
 #include "single_device_tsblock_reader.h"
 
 #include <algorithm>
+#include <cstring>
 #include <iostream>
 #include <set>
 
@@ -230,7 +231,7 @@ int SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task,
     bool used_multi = false;
     std::set<std::string> 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<MeasurementColumnContext*>& 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<SingleMeasurementColumnContext*>(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();

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<common::ColAppender*>& col_appenders,

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;

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();
 

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,20 +76,32 @@ 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())) {
         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))) {
-        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<FloatTS2DIFFEncoder*>(value_encoder_)
+                    ->set_max_point_number(max_point_number);
+            } else if (data_type == DOUBLE) {
+                static_cast<DoubleTS2DIFFEncoder*>(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) {

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();

cpp/src/writer/tsfile_writer.cc

@@ -20,6 +20,7 @@
 #include "tsfile_writer.h"
 
 #include <chrono>
+#include <cstdlib>
 #include <iomanip>
 #include <map>
 #include <set>
@@ -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 {

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;

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();
 

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<FloatTS2DIFFEncoder*>(value_encoder_)
+                    ->set_max_point_number(max_point_number);
+            } else if (data_type == DOUBLE) {
+                static_cast<DoubleTS2DIFFEncoder*>(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) {

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();
 

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<char> 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,