MINIFICPP-2845 - Monotonic provenance event identifiers, iterations support, refactor

cmake/FetchSQLite.cmake

@@ -0,0 +1,29 @@
+# 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.
+
+include(FetchContent)
+
+FetchContent_Declare(
+    sqlite3
+    URL "https://www.sqlite.org/2026/sqlite-amalgamation-3530200.zip"
+    URL_HASH "SHA256=8a310d0a16c7a90cacd4c884e70faa51c902afed2a89f63aaa0126ab83558a32"
+)
+
+FetchContent_MakeAvailable(sqlite3)
+
+add_library(sqlite3 STATIC ${sqlite3_SOURCE_DIR}/sqlite3.c)
+target_include_directories(sqlite3 PUBLIC ${sqlite3_SOURCE_DIR})

core-framework/common/src/utils/Id.cpp

@@ -43,6 +43,18 @@ bool Identifier::isNil() const {
   return *this == Identifier{};
 }
 
+Identifier& Identifier::operator++() {
+  for (int byte_idx = 15; byte_idx >= 0 && ++data_[byte_idx] == 0; --byte_idx) {}
+  return *this;
+}
+
+Identifier Identifier::operator++(int) {
+  auto result = *this;
+  ++*this;
+  return result;
+}
+
+
 bool Identifier::operator!=(const Identifier& other) const {
   return !(*this == other);
 }

core-framework/include/core/Repository.h

@@ -106,12 +106,6 @@ class RepositoryImpl : public core::CoreComponentImpl, public core::RepositoryMe
     return false;
   }
 
-  bool getElements(std::vector<std::shared_ptr<core::SerializableComponent>>& /*store*/, size_t& /*max_size*/) override {
-    return true;
-  }
-
-  bool storeElement(const std::shared_ptr<core::SerializableComponent>& element) override;
-
   void loadComponent(const std::shared_ptr<core::ContentRepository>& /*content_repo*/) override {
   }
 

core-framework/src/core/Repository.cpp

@@ -27,20 +27,4 @@ bool RepositoryImpl::Delete(std::vector<std::shared_ptr<core::SerializableCompon
   return found;
 }
 
-bool RepositoryImpl::storeElement(const std::shared_ptr<core::SerializableComponent>& element) {
-  if (!element) {
-    return false;
-  }
-
-  org::apache::nifi::minifi::io::BufferStream stream;
-
-  element->serialize(stream);
-
-  if (!Put(element->getUUIDStr(), reinterpret_cast<const uint8_t*>(stream.getBuffer().data()), stream.size())) {
-    logger_->log_error("NiFi Provenance Store event {} size {} fail", element->getUUIDStr(), stream.size());
-    return false;
-  }
-  return true;
-}
-
 }  // namespace org::apache::nifi::minifi::core

extensions/grafana-loki/PushGrafanaLokiGrpc.cpp

@@ -116,7 +116,7 @@ std::expected<void, std::string> PushGrafanaLokiGrpc::submitRequest(const std::v
 
   for (const auto& flow_file : batched_flow_files) {
     logproto::EntryAdapter *entry = stream->add_entries();
-    auto timestamp_str = std::to_string(flow_file->getlineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1));
+    auto timestamp_str = std::to_string(flow_file->getLineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1));
     auto timestamp_nanos = std::stoll(timestamp_str);
     *entry->mutable_timestamp() = google::protobuf::util::TimeUtil::NanosecondsToTimestamp(timestamp_nanos);
 

extensions/grafana-loki/PushGrafanaLokiREST.cpp

@@ -133,7 +133,7 @@ std::string PushGrafanaLokiREST::createLokiJson(const std::vector<std::shared_pt
   for (const auto& flow_file : batched_flow_files) {
     rapidjson::Value log_line(rapidjson::kArrayType);
 
-    auto timestamp_str = std::to_string(flow_file->getlineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1));
+    auto timestamp_str = std::to_string(flow_file->getLineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1));
     rapidjson::Value timestamp;
     timestamp.SetString(timestamp_str.c_str(), gsl::narrow<rapidjson::SizeType>(timestamp_str.length()), allocator);
     rapidjson::Value log_line_value;