feat(clp-s)!: Add Timestamp column type to replaceDateString column type; Bump the archive version to 0.5.0. (#1788)

Co-authored-by: Jack Luo <jack.luo@yscope.com>
Co-authored-by: Junhao Liao <junhao.liao@yscope.com>
Co-authored-by: Lin Zhihao <59785146+LinZhihao-723@users.noreply.github.com>

Author: 4 people

components/core/cmake/Options/options.cmake

@@ -132,6 +132,7 @@ function(validate_clp_binaries_dependencies)
         CLP_BUILD_CLP_S_SEARCH
         CLP_BUILD_CLP_S_SEARCH_AST
         CLP_BUILD_CLP_S_SEARCH_KQL
+        CLP_BUILD_CLP_S_TIMESTAMP_PARSER
     )
 endfunction()
 
@@ -207,6 +208,7 @@ function(validate_clp_s_archivereader_dependencies)
         CLP_BUILD_CLP_STRING_UTILS
         CLP_BUILD_CLP_S_CLP_DEPENDENCIES
         CLP_BUILD_CLP_S_IO
+        CLP_BUILD_CLP_S_TIMESTAMP_PARSER
         CLP_BUILD_CLP_S_TIMESTAMPPATTERN
     )
 endfunction()
@@ -228,6 +230,7 @@ function(validate_clp_s_archivewriter_dependencies)
     validate_clp_dependencies_for_target(CLP_BUILD_CLP_S_ARCHIVEWRITER
         CLP_BUILD_CLP_S_CLP_DEPENDENCIES
         CLP_BUILD_CLP_S_IO
+        CLP_BUILD_CLP_S_TIMESTAMP_PARSER
         CLP_BUILD_CLP_S_TIMESTAMPPATTERN
     )
 endfunction()

components/core/src/clp_s/ArchiveReader.cpp

@@ -212,8 +212,12 @@ BaseColumnReader* ArchiveReader::append_reader_column(SchemaReader& reader, int3
         case NodeType::UnstructuredArray:
             column_reader = new ClpStringColumnReader(column_id, m_var_dict, m_array_dict, true);
             break;
-        case NodeType::DateString:
-            column_reader = new DateStringColumnReader(column_id, get_timestamp_dictionary());
+        case NodeType::DeprecatedDateString:
+            column_reader
+                    = new DeprecatedDateStringColumnReader(column_id, get_timestamp_dictionary());
+            break;
+        case NodeType::Timestamp:
+            column_reader = new TimestampColumnReader(column_id, get_timestamp_dictionary());
             break;
         // No need to push columns without associated object readers into the SchemaReader.
         case NodeType::Metadata:
@@ -268,10 +272,11 @@ void ArchiveReader::append_unordered_reader_columns(
             case NodeType::Boolean:
                 column_reader = new BooleanColumnReader(column_id);
                 break;
-            // UnstructuredArray and DateString currently aren't supported as part of any unordered
-            // object, so we disregard them here
+            // UnstructuredArray, DeprecatedDateString, and Timestamp currently aren't supported as
+            // part of any unordered object, so we disregard them here
             case NodeType::UnstructuredArray:
-            case NodeType::DateString:
+            case NodeType::DeprecatedDateString:
+            case NodeType::Timestamp:
             // No need to push columns without associated object readers into the SchemaReader.
             case NodeType::StructuredArray:
             case NodeType::Object:

components/core/src/clp_s/ArchiveReader.hpp

@@ -151,6 +151,14 @@ class ArchiveReader {
      */
     bool has_log_order() { return m_log_event_idx_column_id >= 0; }
 
+    /**
+     * @return Whether this archive can contain columns with the deprecated DateString timestamp
+     * format.
+     */
+    [[nodiscard]] auto has_deprecated_timestamp_format() const -> bool {
+        return get_header().has_deprecated_timestamp_format();
+    }
+
 private:
     /**
      * Initializes a schema reader passed by reference to become a reader for a given schema.

components/core/src/clp_s/ArchiveReaderAdaptor.cpp

@@ -75,7 +75,10 @@ ArchiveReaderAdaptor::try_read_archive_file_info(ZstdDecompressor& decompressor,
 
 ErrorCode
 ArchiveReaderAdaptor::try_read_timestamp_dictionary(ZstdDecompressor& decompressor, size_t size) {
-    return m_timestamp_dictionary->read(decompressor);
+    return m_timestamp_dictionary->read(
+            decompressor,
+            m_archive_header.has_deprecated_timestamp_format()
+    );
 }
 
 ErrorCode ArchiveReaderAdaptor::try_read_archive_info(ZstdDecompressor& decompressor, size_t size) {

components/core/src/clp_s/ArchiveWriter.cpp

@@ -333,12 +333,13 @@ void ArchiveWriter::initialize_schema_writer(SchemaWriter* writer, Schema const&
                         std::make_unique<ClpStringColumnWriter>(id, m_var_dict, m_array_dict)
                 );
                 break;
-            case NodeType::DateString:
-                writer->append_column(std::make_unique<DateStringColumnWriter>(id));
-                break;
             case NodeType::DeltaInteger:
                 writer->append_column(std::make_unique<DeltaEncodedInt64ColumnWriter>(id));
                 break;
+            case NodeType::Timestamp:
+                writer->append_column(std::make_unique<TimestampColumnWriter>(id));
+                break;
+            case NodeType::DeprecatedDateString:
             case NodeType::Metadata:
             case NodeType::NullValue:
             case NodeType::Object:

components/core/src/clp_s/ArchiveWriter.hpp

@@ -193,34 +193,49 @@ class ArchiveWriter {
     int32_t add_schema(Schema const& schema) { return m_schema_map.add_schema(schema); }
 
     /**
-     * Ingests a timestamp entry from a string
+     * Ingests a timestamp entry from a string.
      * @param key
      * @param node_id
      * @param timestamp
-     * @param pattern_id
-     * @return the epoch time corresponding to the string timestamp
+     * @param is_json_literal
+     * @return Forwards `TimestampDictionaryWriter::ingest_string_timestamp`'s return values.
      */
-    epochtime_t ingest_timestamp_entry(
+    [[nodiscard]] auto ingest_string_timestamp(
             std::string_view key,
             int32_t node_id,
             std::string_view timestamp,
-            uint64_t& pattern_id
-    ) {
-        return m_timestamp_dict.ingest_entry(key, node_id, timestamp, pattern_id);
+            bool is_json_literal
+    ) -> std::pair<epochtime_t, uint64_t> {
+        return m_timestamp_dict.ingest_string_timestamp(key, node_id, timestamp, is_json_literal);
     }
 
     /**
-     * Ingests a timestamp entry from a number
-     * @param column_key
+     * Ingests a numeric JSON entry.
+     * @param key
      * @param node_id
      * @param timestamp
+     * @return Forwards `TimestampDictionaryWriter::ingest_numeric_json_timestamp`'s return values.
      */
-    void ingest_timestamp_entry(std::string_view key, int32_t node_id, double timestamp) {
-        m_timestamp_dict.ingest_entry(key, node_id, timestamp);
+    [[nodiscard]] auto
+    ingest_numeric_json_timestamp(std::string_view key, int32_t node_id, std::string_view timestamp)
+            -> std::pair<epochtime_t, uint64_t> {
+        return m_timestamp_dict.ingest_numeric_json_timestamp(key, node_id, timestamp);
     }
 
-    void ingest_timestamp_entry(std::string_view key, int32_t node_id, int64_t timestamp) {
-        m_timestamp_dict.ingest_entry(key, node_id, timestamp);
+    /**
+     * Ingests an unknown precision epoch timestamp.
+     * @param key
+     * @param node_id
+     * @param timestamp
+     * @return Forwards `TimestampDictionaryWriter::ingest_unknown_precision_epoch_timestamp`'s
+     * return values.
+     */
+    [[nodiscard]] auto ingest_unknown_precision_epoch_timestamp(
+            std::string_view key,
+            int32_t node_id,
+            int64_t timestamp
+    ) -> std::pair<epochtime_t, uint64_t> {
+        return m_timestamp_dict.ingest_unknown_precision_epoch_timestamp(key, node_id, timestamp);
     }
 
     /**

components/core/src/clp_s/CMakeLists.txt

@@ -281,13 +281,14 @@ if(CLP_BUILD_CLP_S_ARCHIVEWRITER)
                 absl::flat_hash_map
                 clp_s::clp_dependencies
                 clp_s::io
+                clp_s::timestamp_parser
+                clp_s::timestamp_pattern
                 msgpack-cxx
                 nlohmann_json::nlohmann_json
                 simdjson::simdjson
                 ystdlib::error_handling
                 PRIVATE
                 Boost::url
-                clp_s::timestamp_pattern
                 ${CURL_LIBRARIES}
                 fmt::fmt
                 spdlog::spdlog
@@ -345,13 +346,14 @@ if(CLP_BUILD_CLP_S_ARCHIVEREADER)
                 absl::flat_hash_map
                 clp::string_utils
                 clp_s::io
+                clp_s::timestamp_parser
+                clp_s::timestamp_pattern
                 msgpack-cxx
                 nlohmann_json::nlohmann_json
                 ystdlib::error_handling
                 PRIVATE
                 Boost::url
                 clp_s::clp_dependencies
-                clp_s::timestamp_pattern
                 ${CURL_LIBRARIES}
                 fmt::fmt
                 spdlog::spdlog

components/core/src/clp_s/ColumnReader.cpp

@@ -226,31 +226,31 @@ int64_t VariableStringColumnReader::get_variable_id(uint64_t cur_message) {
     return m_variables[cur_message];
 }
 
-void DateStringColumnReader::load(BufferViewReader& reader, uint64_t num_messages) {
+void DeprecatedDateStringColumnReader::load(BufferViewReader& reader, uint64_t num_messages) {
     m_timestamps = reader.read_unaligned_span<int64_t>(num_messages);
     m_timestamp_encodings = reader.read_unaligned_span<int64_t>(num_messages);
 }
 
-std::variant<int64_t, double, std::string, uint8_t> DateStringColumnReader::extract_value(
+std::variant<int64_t, double, std::string, uint8_t> DeprecatedDateStringColumnReader::extract_value(
         uint64_t cur_message
 ) {
-    return m_timestamp_dict->get_string_encoding(
+    return m_timestamp_dict->get_deprecated_timestamp_string_encoding(
             m_timestamps[cur_message],
             m_timestamp_encodings[cur_message]
     );
 }
 
-void DateStringColumnReader::extract_string_value_into_buffer(
+void DeprecatedDateStringColumnReader::extract_string_value_into_buffer(
         uint64_t cur_message,
         std::string& buffer
 ) {
-    buffer.append(m_timestamp_dict->get_string_encoding(
+    buffer.append(m_timestamp_dict->get_deprecated_timestamp_string_encoding(
             m_timestamps[cur_message],
             m_timestamp_encodings[cur_message]
     ));
 }
 
-epochtime_t DateStringColumnReader::get_encoded_time(uint64_t cur_message) {
+epochtime_t DeprecatedDateStringColumnReader::get_encoded_time(uint64_t cur_message) {
     return m_timestamps[cur_message];
 }
 

components/core/src/clp_s/ColumnReader.hpp

@@ -327,20 +327,23 @@ class VariableStringColumnReader : public BaseColumnReader {
     UnalignedMemSpan<uint64_t> m_variables;
 };
 
-class DateStringColumnReader : public BaseColumnReader {
+class DeprecatedDateStringColumnReader : public BaseColumnReader {
 public:
     // Constructor
-    DateStringColumnReader(int32_t id, std::shared_ptr<TimestampDictionaryReader> timestamp_dict)
+    DeprecatedDateStringColumnReader(
+            int32_t id,
+            std::shared_ptr<TimestampDictionaryReader> timestamp_dict
+    )
             : BaseColumnReader(id),
               m_timestamp_dict(std::move(timestamp_dict)) {}
 
     // Destructor
-    ~DateStringColumnReader() override = default;
+    ~DeprecatedDateStringColumnReader() override = default;
 
     // Methods inherited from BaseColumnReader
     void load(BufferViewReader& reader, uint64_t num_messages) override;
 
-    NodeType get_type() override { return NodeType::DateString; }
+    NodeType get_type() override { return NodeType::DeprecatedDateString; }
 
     std::variant<int64_t, double, std::string, uint8_t> extract_value(
             uint64_t cur_message

components/core/src/clp_s/ColumnWriter.cpp

@@ -165,21 +165,6 @@ void VariableStringColumnWriter::store(ZstdCompressor& compressor) {
     compressor.write(reinterpret_cast<char const*>(m_var_dict_ids.data()), size);
 }
 
-size_t DateStringColumnWriter::add_value(ParsedMessage::variable_t& value) {
-    auto encoded_timestamp = std::get<std::pair<uint64_t, epochtime_t>>(value);
-    m_timestamps.push_back(encoded_timestamp.second);
-    m_timestamp_encodings.push_back(encoded_timestamp.first);
-    return 2 * sizeof(int64_t);
-    ;
-}
-
-void DateStringColumnWriter::store(ZstdCompressor& compressor) {
-    size_t timestamps_size = m_timestamps.size() * sizeof(int64_t);
-    compressor.write(reinterpret_cast<char const*>(m_timestamps.data()), timestamps_size);
-    size_t encodings_size = m_timestamp_encodings.size() * sizeof(int64_t);
-    compressor.write(reinterpret_cast<char const*>(m_timestamp_encodings.data()), encodings_size);
-}
-
 auto TimestampColumnWriter::add_value(ParsedMessage::variable_t& value) -> size_t {
     auto const [timestamp, encoding] = std::get<std::pair<epochtime_t, uint64_t>>(value);
     auto const encoded_timestamp_size{m_timestamps.add_value(timestamp)};