Add C++ and Python TsFile properties support (#897)

* Add C++ and Python TsFile properties support

* Fix TsFile properties metadata compatibility

* Fix Windows TsFile binary file mode

* Preserve legacy empty BloomFilter metadata

Author: ColinLeeo

cpp/README-zh.md

@@ -188,3 +188,22 @@ bash build.sh
 ```
 
 即可在 `./examples/build` 目录下生成可执行文件。
+
+### 文件级 Properties
+
+`TsFileWriter` 和 `TsFileTableWriter` 可以在 writer 打开期间新增或覆盖二进制
+property。传入的数据会立即复制，调用 `flush()` 后仍可继续修改；文件关闭后不能修改。
+
+```cpp
+std::vector<uint8_t> value = {0x01, 0x00, 0xFF};
+writer.add_tsfile_property("binary-property", value);
+
+// nullptr 且长度为 0 表示 null；空 vector 表示非 null 的零长度值。
+writer.add_tsfile_property("null-property", nullptr, 0);
+writer.add_tsfile_property("empty-property", std::vector<uint8_t>());
+
+storage::TsFileProperties properties = reader.get_tsfile_properties();
+```
+
+Property value 本身不保存数据类型。整数、浮点数或结构体应由应用使用明确、可跨语言的
+字节编码进行转换。

cpp/README.md

@@ -203,3 +203,23 @@ By default, parallel write is enabled when the machine has more than one CPU cor
 ## Use TsFile
 
 You can find examples on how to read and write data in `demo_read.cpp` and `demo_write.cpp` located under `./examples/cpp_examples`. There are also examples under `./examples/c_examples` on how to use a C-style API to read and write data in a C environment. The examples will be built automatically when you run the main build command.
+
+### File-level properties
+
+`TsFileWriter` and `TsFileTableWriter` can add or replace binary properties
+while the writer is open. Values are copied immediately and may still be
+changed after `flush()`; a closed file cannot be modified.
+
+```cpp
+std::vector<uint8_t> value = {0x01, 0x00, 0xFF};
+writer.add_tsfile_property("binary-property", value);
+
+// nullptr with length 0 is null; an empty vector is a non-null empty value.
+writer.add_tsfile_property("null-property", nullptr, 0);
+writer.add_tsfile_property("empty-property", std::vector<uint8_t>());
+
+storage::TsFileProperties properties = reader.get_tsfile_properties();
+```
+
+Property values do not store a data type. Applications should define their own
+portable byte encoding for integers, floating-point values, or structures.

cpp/src/common/tsfile_common.cc

@@ -20,6 +20,7 @@
 #include "common/tsfile_common.h"
 
 #include <algorithm>
+#include <limits>
 #include <map>
 
 #include "common/logger/elog.h"
@@ -180,37 +181,101 @@ int TSMIterator::get_next(std::shared_ptr<IDeviceID>& ret_device_id,
     return ret;
 }
 
-int TsFileMeta::serialize_to(common::ByteStream& out) {
+int TsFileMeta::serialize_to(common::ByteStream& out,
+                             int32_t& serialized_size) {
+    serialized_size = 0;
+    const size_t max_property_size =
+        static_cast<size_t>(std::numeric_limits<int32_t>::max());
+    if (tsfile_properties_.size() > max_property_size) {
+        return common::E_OUT_OF_RANGE;
+    }
+    for (const auto& tsfile_property : tsfile_properties_) {
+        if (tsfile_property.first.size() > max_property_size ||
+            (!tsfile_property.second.is_null &&
+             tsfile_property.second.value.size() > max_property_size)) {
+            return common::E_OUT_OF_RANGE;
+        }
+    }
+
+    int ret = common::E_OK;
     auto start_idx = out.total_size();
-    common::SerializationUtil::write_var_uint(
-        table_metadata_index_node_map_.size(), out);
+    if (RET_FAIL(common::SerializationUtil::write_var_uint(
+            table_metadata_index_node_map_.size(), out))) {
+        return ret;
+    }
     for (auto& idx_nodes_iter : table_metadata_index_node_map_) {
-        common::SerializationUtil::write_var_str(idx_nodes_iter.first, out);
-        idx_nodes_iter.second->serialize_to(out);
+        if (RET_FAIL(common::SerializationUtil::write_var_str(
+                idx_nodes_iter.first, out))) {
+            return ret;
+        } else if (RET_FAIL(idx_nodes_iter.second->serialize_to(out))) {
+            return ret;
+        }
     }
 
-    common::SerializationUtil::write_var_uint(table_schemas_.size(), out);
+    if (RET_FAIL(common::SerializationUtil::write_var_uint(
+            table_schemas_.size(), out))) {
+        return ret;
+    }
     for (auto& table_schema_iter : table_schemas_) {
-        common::SerializationUtil::write_var_str(table_schema_iter.first, out);
-        table_schema_iter.second->serialize_to(out);
+        if (RET_FAIL(common::SerializationUtil::write_var_str(
+                table_schema_iter.first, out))) {
+            return ret;
+        } else if (RET_FAIL(table_schema_iter.second->serialize_to(out))) {
+            return ret;
+        }
     }
 
-    common::SerializationUtil::write_i64(meta_offset_, out);
+    if (RET_FAIL(common::SerializationUtil::write_i64(meta_offset_, out))) {
+        return ret;
+    }
 
     if (bloom_filter_ != nullptr) {
-        bloom_filter_->serialize_to(out);
+        if (RET_FAIL(bloom_filter_->serialize_to(out))) {
+            return ret;
+        }
     } else {
-        common::SerializationUtil::write_ui8(0, out);
+        if (RET_FAIL(common::SerializationUtil::write_ui8(0, out))) {
+            return ret;
+        }
     }
 
-    common::SerializationUtil::write_var_int(tsfile_properties_.size(), out);
+    if (RET_FAIL(common::SerializationUtil::write_var_int(
+            static_cast<int32_t>(tsfile_properties_.size()), out))) {
+        return ret;
+    }
     for (const auto& tsfile_property : tsfile_properties_) {
-        common::SerializationUtil::write_var_str(tsfile_property.first, out);
-        common::SerializationUtil::write_var_char_ptr(tsfile_property.second,
-                                                      out);
+        if (RET_FAIL(common::SerializationUtil::write_var_str(
+                tsfile_property.first, out))) {
+            return ret;
+        }
+        const TsFilePropertyValue& value = tsfile_property.second;
+        if (value.is_null) {
+            if (RET_FAIL(common::SerializationUtil::write_var_int(
+                    NO_STR_TO_READ, out))) {
+                return ret;
+            }
+        } else {
+            if (RET_FAIL(common::SerializationUtil::write_var_int(
+                    static_cast<int32_t>(value.value.size()), out))) {
+                return ret;
+            }
+            if (!value.value.empty()) {
+                if (RET_FAIL(out.write_buf(
+                        value.value.data(),
+                        static_cast<uint32_t>(value.value.size())))) {
+                    return ret;
+                }
+            }
+        }
     }
 
-    return out.total_size() - start_idx;
+    const uint64_t total_size = out.total_size() - start_idx;
+    if (total_size >
+        static_cast<uint64_t>(std::numeric_limits<int32_t>::max())) {
+        return common::E_OUT_OF_RANGE;
+    }
+    serialized_size = static_cast<int32_t>(total_size);
+    return common::E_OK;
 }
 
 int TsFileMeta::deserialize_from(common::ByteStream& in) {
@@ -251,15 +316,67 @@ int TsFileMeta::deserialize_from(common::ByteStream& in) {
 
     common::SerializationUtil::read_i64(meta_offset_, in);
 
-    bloom_filter_->deserialize_from(in);
+    if (RET_FAIL(bloom_filter_->deserialize_from(in))) {
+        return ret;
+    }
 
     int32_t tsfile_properties_size = 0;
-    common::SerializationUtil::read_var_int(tsfile_properties_size, in);
+    if (RET_FAIL(common::SerializationUtil::read_var_int(tsfile_properties_size,
+                                                         in))) {
+        return ret;
+    }
+    if (tsfile_properties_size < 0) {
+        return common::E_TSFILE_CORRUPTED;
+    }
     for (int i = 0; i < tsfile_properties_size; i++) {
-        std::string key, *value;
-        common::SerializationUtil::read_var_str(key, in);
-        common::SerializationUtil::read_var_char_ptr(value, in);
-        tsfile_properties_.emplace(key, value);
+        std::string key;
+        int32_t key_len = 0;
+        int32_t value_len = 0;
+        if (RET_FAIL(common::SerializationUtil::read_var_int(key_len, in))) {
+            return ret;
+        } else if (key_len < 0) {
+            return common::E_TSFILE_CORRUPTED;
+        }
+        if (static_cast<uint64_t>(key_len) > in.remaining_size()) {
+            return common::E_TSFILE_CORRUPTED;
+        }
+        key.resize(static_cast<size_t>(key_len));
+        if (key_len > 0) {
+            uint32_t read_len = 0;
+            if (RET_FAIL(in.read_buf(reinterpret_cast<uint8_t*>(&key[0]),
+                                     static_cast<uint32_t>(key_len),
+                                     read_len))) {
+                return ret;
+            } else if (read_len != static_cast<uint32_t>(key_len)) {
+                return common::E_BUF_NOT_ENOUGH;
+            }
+        }
+        if (RET_FAIL(common::SerializationUtil::read_var_int(value_len, in))) {
+            return ret;
+        }
+
+        TsFilePropertyValue value;
+        if (value_len == NO_STR_TO_READ) {
+            value.is_null = true;
+        } else if (value_len < 0) {
+            return common::E_TSFILE_CORRUPTED;
+        } else {
+            if (static_cast<uint64_t>(value_len) > in.remaining_size()) {
+                return common::E_TSFILE_CORRUPTED;
+            }
+            value.is_null = false;
+            value.value.resize(static_cast<size_t>(value_len));
+            if (value_len > 0) {
+                uint32_t read_len = 0;
+                if (RET_FAIL(
+                        in.read_buf(value.value.data(), value_len, read_len))) {
+                    return ret;
+                } else if (read_len != static_cast<uint32_t>(value_len)) {
+                    return common::E_BUF_NOT_ENOUGH;
+                }
+            }
+        }
+        tsfile_properties_.emplace(key, std::move(value));
     }
     return ret;
 }
@@ -375,4 +492,4 @@ int MetaIndexNode::binary_search_children(const String &name,
 }
 #endif
 
-}  // end namespace storage
+}  // end namespace storage

cpp/src/common/tsfile_common.h

@@ -1126,13 +1126,35 @@ struct MetaIndexNode {
 
 class TableSchema;
 
+struct TsFilePropertyValue {
+    /** A default-constructed property represents a null value. */
+    TsFilePropertyValue() : is_null(true), value() {}
+
+    /** A vector, including an empty vector, represents a non-null value. */
+    explicit TsFilePropertyValue(const std::vector<uint8_t>& value)
+        : is_null(false), value(value) {}
+
+    /** nullptr represents null; a non-null pointer with length 0 is empty. */
+    TsFilePropertyValue(const uint8_t* data, uint32_t value_len)
+        : is_null(data == nullptr), value() {
+        if (data != nullptr && value_len > 0) {
+            value.assign(data, data + value_len);
+        }
+    }
+
+    bool is_null;
+    std::vector<uint8_t> value;
+};
+
+using TsFileProperties = std::unordered_map<std::string, TsFilePropertyValue>;
+
 struct TsFileMeta {
     typedef std::map<std::shared_ptr<IDeviceID>, std::shared_ptr<MetaIndexNode>,
                      IDeviceIDComparator>
         DeviceNodeMap;
     std::map<std::string, std::shared_ptr<MetaIndexNode>>
         table_metadata_index_node_map_;
-    std::unordered_map<std::string, std::string*> tsfile_properties_;
+    TsFileProperties tsfile_properties_;
     typedef std::unordered_map<std::string, std::shared_ptr<TableSchema>>
         TableSchemasMap;
     TableSchemasMap table_schemas_;
@@ -1170,18 +1192,12 @@ struct TsFileMeta {
         if (bloom_filter_ != nullptr) {
             bloom_filter_->destroy();
         }
-        for (auto properties : tsfile_properties_) {
-            if (properties.second != nullptr) {
-                delete properties.second;
-                properties.second = nullptr;
-            }
-        }
         tsfile_properties_.clear();
         table_metadata_index_node_map_.clear();
         table_schemas_.clear();
     }
 
-    int serialize_to(common::ByteStream& out);
+    int serialize_to(common::ByteStream& out, int32_t& serialized_size);
 
     int deserialize_from(common::ByteStream& in);