v4-write-process

cpp/src/common/allocator/my_string.h

@@ -40,6 +40,9 @@ struct String {
         len_ = 0;
         buf_ = nullptr;
     }
+    std::string to_string() {
+        return std::string(buf_, len_);
+    }
     FORCE_INLINE int dup_from(const std::string &str, common::PageArena &pa) {
         len_ = str.size();
         if (UNLIKELY(len_ == 0)) {

cpp/src/common/config/config.h

@@ -19,7 +19,7 @@
 #ifndef COMMON_CONFIG_CONFIG_H
 #define COMMON_CONFIG_CONFIG_H
 
-#include <stdint.h>
+#include <cstdint>
 
 #include "common/mutex/mutex.h"
 #include "utils/db_utils.h"
@@ -46,9 +46,11 @@ typedef struct ConfigValue {
     CompressionType time_compress_type_;
     int32_t chunk_group_size_threshold_;
     int32_t record_count_for_next_mem_check_;
+    bool encrypt_flag_ = false;
 } ConfigValue;
 
 extern void init_config_value();
+extern TSEncoding get_value_encoder(TSDataType data_type);
 
 // In the future, configuration items need to be dynamically adjusted according
 // to the level

cpp/src/common/db_common.h

@@ -34,6 +34,7 @@ enum TSDataType {
     DOUBLE = 4,
     TEXT = 5,
     VECTOR = 6,
+    STRING = 11,
     NULL_TYPE = 254,
     INVALID_DATATYPE = 255
 };

cpp/src/common/device_id.h

@@ -0,0 +1,212 @@
+/*
+ * 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 COMMON_DEVICE_ID_H
+#define COMMON_DEVICE_ID_H
+
+#include <algorithm>
+#include <cstdint>
+#include <cstring>
+#include <functional>
+#include <iostream>
+#include <memory>
+#include <numeric>
+#include <sstream>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+#include "common/allocator/byte_stream.h"
+#include "utils/errno_define.h"
+
+class IDeviceID {
+   public:
+    virtual ~IDeviceID() = default;
+    virtual int serialize(common::ByteStream& write_stream) { return 0; }
+    virtual std::string get_table_name() { return ""; }
+    virtual int segment_num() { return 0; }
+    virtual std::vector<std::string> get_segments() const { return {}; }
+    virtual std::string get_device_name() const { return ""; };
+    virtual bool operator<(const IDeviceID& other) { return 0; }
+    virtual bool operator==(const IDeviceID& other) { return false; }
+    virtual bool operator!=(const IDeviceID& other) { return false; }
+};
+
+struct IDeviceIDComparator {
+    bool operator()(const std::shared_ptr<IDeviceID>& lhs,
+                    const std::shared_ptr<IDeviceID>& rhs) const {
+        return *lhs < *rhs;
+    }
+};
+
+class StringArrayDeviceID : public IDeviceID {
+   public:
+    explicit StringArrayDeviceID(const std::vector<std::string>& segments)
+        : segments_(formalize(segments)) {}
+
+    explicit StringArrayDeviceID(const std::string& device_id_string)
+        : segments_(split_device_id_string(device_id_string)) {}
+
+    ~StringArrayDeviceID() override = default;
+
+    std::string get_device_name() const override {
+        return std::accumulate(std::next(segments_.begin()), segments_.end(),
+                               segments_.front(),
+                               [](std::string a, const std::string& b) {
+                                   return std::move(a) + "." + b;
+                               });
+    };
+
+    int serialize(common::ByteStream& write_stream) override {
+        int ret = common::E_OK;
+        if (RET_FAIL(common::SerializationUtil::write_var_int(segment_num(),
+                                                              write_stream))) {
+            return ret;
+        }
+        for (const auto& segment : segments_) {
+            if (RET_FAIL(common::SerializationUtil::write_var_int(
+                    segment.size(), write_stream))) {
+                return ret;
+            } else if (RET_FAIL(write_stream.write_buf(segment.c_str(),
+                                                       segment.size()))) {
+                return ret;
+            }
+        }
+        return ret;
+    }
+
+    std::string get_table_name() override {
+        return segments_.empty() ? "" : segments_[0];
+    }
+
+    int segment_num() override { return static_cast<int>(segments_.size()); }
+
+    std::vector<std::string> get_segments() const override { return segments_; }
+
+    virtual bool operator<(const IDeviceID& other) override {
+        auto other_segments = other.get_segments();
+        return std::lexicographical_compare(segments_.begin(), segments_.end(),
+                                            other_segments.begin(),
+                                            other_segments.end());
+    }
+
+    bool operator==(const IDeviceID& other) override {
+        auto other_segments = other.get_segments();
+        return (segments_.size() == other_segments.size()) &&
+               std::equal(segments_.begin(), segments_.end(),
+                          other_segments.begin());
+    }
+
+    bool operator!=(const IDeviceID& other) override {
+        return !(*this == other);
+    }
+
+   private:
+    std::vector<std::string> segments_;
+
+    static std::vector<std::string> formalize(
+        const std::vector<std::string>& segments) {
+        auto it =
+            std::find_if(segments.rbegin(), segments.rend(),
+                         [](const std::string& seg) { return !seg.empty(); });
+        return std::vector<std::string>(segments.begin(), it.base());
+    }
+
+    static std::vector<std::string> split_device_id_string(
+        std::basic_string<char> device_id_string) {
+        std::vector<std::string> splits;
+        std::istringstream stream(device_id_string);
+        std::string segment;
+        while (std::getline(stream, segment, '.')) {
+            splits.push_back(segment);
+        }
+        return splits;
+    }
+};
+
+class PlainDeviceID : public IDeviceID {
+   public:
+    explicit PlainDeviceID(const std::string& deviceID)
+        : device_id_(deviceID), tableName_(), segments_() {}
+
+    ~PlainDeviceID() override = default;
+
+    bool operator==(const IDeviceID& other) override {
+        return device_id_ == other.get_device_name();
+    }
+
+    bool operator!=(const IDeviceID& other) override {
+        return device_id_ != other.get_device_name();
+    }
+
+    int serialize(common::ByteStream& write_stream) override {
+        int ret = common::E_OK;
+        if (RET_FAIL(common::SerializationUtil::write_var_int(device_id_.size(),
+                                                              write_stream))) {
+            return ret;
+        } else if (RET_FAIL(write_stream.write_buf(device_id_.c_str(),
+                                                   device_id_.size()))) {
+            return ret;
+        }
+        return ret;
+    }
+
+    std::string get_device_name() const override { return device_id_; };
+
+    std::string get_table_name() override {
+        if (!tableName_.empty()) {
+            return tableName_;
+        }
+
+        size_t lastSeparatorPos = device_id_.find_last_of('.');
+        if (lastSeparatorPos == std::string::npos) {
+            tableName_ = device_id_;  // Use entire deviceID as tableName
+        } else {
+            tableName_ = device_id_.substr(0, lastSeparatorPos);
+        }
+        return tableName_;
+    }
+
+    int segment_num() override {
+        if (!segments_.empty()) {
+            return static_cast<int>(segments_.size());
+        }
+        split_segments();
+        return static_cast<int>(segments_.size());
+    }
+
+    bool operator<(const IDeviceID& other) override {
+        return device_id_ < other.get_device_name();
+    }
+
+   private:
+    std::string device_id_;
+    std::string tableName_;
+    std::vector<std::string> segments_;
+
+    void split_segments() {
+        std::istringstream stream(device_id_);
+        std::string segment;
+        while (std::getline(stream, segment, '.')) {
+            segments_.push_back(segment);
+        }
+    }
+};
+
+#endif

cpp/src/common/global.cc

@@ -46,6 +46,32 @@ void init_config_value() {
     g_config_value_.time_compress_type_ = LZ4;
 }
 
+extern TSEncoding get_value_encoder(TSDataType data_type) {
+    switch (data_type) {
+        case BOOLEAN:
+            return TSEncoding::RLE;
+        case INT32:
+            return TSEncoding::TS_2DIFF;
+        case INT64:
+            return TSEncoding::TS_2DIFF;
+        case FLOAT:
+            return TSEncoding::GORILLA;
+        case DOUBLE:
+            return TSEncoding::GORILLA;
+        case TEXT:
+            return TSEncoding::PLAIN;
+        case VECTOR:
+            break;
+        case NULL_TYPE:
+            break;
+        case INVALID_DATATYPE:
+            break;
+        default:
+            break;
+    }
+    return TSEncoding::PLAIN;
+}
+
 void config_set_page_max_point_count(uint32_t page_max_ponint_count) {
     g_config_value_.page_writer_max_point_num_ = page_max_ponint_count;
 }