Copilot commented on code in PR #2201:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2201#discussion_r3474537714
##########
thirdparty/lmdb/add-cmake-file.patch:
##########
@@ -0,0 +1,29 @@
+lmdb only comes with a Makefile which only works on Linux so we create our own
CMakeLists.txt file to be platform independent
+
+diff -rupN orig/libraries/liblmdb/CMakeLists.txt
patched/libraries/liblmdb/CMakeLists.txt
Review Comment:
This patch file begins with a free-form prose line, which GNU `patch`
commonly treats as invalid input ("Only garbage was found in the patch input")
when applying with `patch -p1`. Since `cmake/LMDB.cmake` applies this via
`patch`, the patch should start directly with the diff header.
##########
extensions/lmdb/LmdbStream.cpp:
##########
@@ -0,0 +1,116 @@
+/**
+ *
+ * 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 "LmdbStream.h"
+
+#include <string>
+#include <utility>
+
Review Comment:
`LmdbStream.cpp` uses `std::min` and `std::memcpy` but does not include
`<algorithm>` / `<cstring>`. This can cause non-portable builds depending on
transitive includes.
##########
extensions/lmdb/LmdbStream.cpp:
##########
@@ -0,0 +1,116 @@
+/**
+ *
+ * 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 "LmdbStream.h"
+
+#include <string>
+#include <utility>
+
+#include "io/validation.h"
+
+namespace org::apache::nifi::minifi::io {
+
+LmdbStream::LmdbStream(std::string path, MDB_env* lmdb_env, MDB_dbi*
lmdb_handle, bool write_enable)
+ : BaseStreamImpl(),
+ path_(std::move(path)),
+ write_enable_(write_enable),
+ lmdb_env_(lmdb_env),
+ lmdb_handle_(lmdb_handle),
+ exists_(loadValue()) {}
+
+bool LmdbStream::loadValue() {
+ MDB_val key{path_.size(), const_cast<char*>(path_.data())};
+ MDB_val value{};
+
+ MDB_txn* txn = nullptr;
+ if (const int rc = mdb_txn_begin(lmdb_env_, nullptr, MDB_RDONLY, &txn); rc
!= MDB_SUCCESS) {
+ logger_->log_error("Failed to begin LMDB read transaction in loadValue:
{}", mdb_strerror(rc));
+ return false;
+ }
+ auto guard = gsl::finally([txn] { mdb_txn_abort(txn); });
+
+ const auto rc = mdb_get(txn, *lmdb_handle_, &key, &value);
+ if (rc == MDB_SUCCESS) {
+ value_ = std::string(static_cast<char*>(value.mv_data), value.mv_size);
+ return true;
+ } else if (rc != MDB_NOTFOUND) {
+ logger_->log_error("Failed to get value from LMDB database: {}",
mdb_strerror(rc));
+ }
+ return false;
+}
+
+void LmdbStream::close() {
+ commit();
+}
+
+bool LmdbStream::commit() {
+ if (!write_enable_ || !dirty_) { return false; }
+ dirty_ = false;
+
+ MDB_txn* txn = nullptr;
+ auto rc = mdb_txn_begin(lmdb_env_, nullptr, 0, &txn);
+ if (rc != MDB_SUCCESS) {
+ logger_->log_error("Failed to begin LMDB transaction in close: {}",
mdb_strerror(rc));
+ return false;
+ }
+
+ MDB_val key{path_.size(), const_cast<char*>(path_.data())};
+ MDB_val val{value_.size(), const_cast<char*>(value_.data())};
+ rc = mdb_put(txn, *lmdb_handle_, &key, &val, 0);
+ if (rc != MDB_SUCCESS) {
+ logger_->log_error("Failed to put value in LMDB database during close:
{}", mdb_strerror(rc));
+ mdb_txn_abort(txn);
+ return false;
+ }
+
+ rc = mdb_txn_commit(txn);
+ if (rc != MDB_SUCCESS) {
+ logger_->log_error("Failed to commit LMDB transaction during close: {}",
mdb_strerror(rc));
+ return false;
+ }
+ return true;
+}
Review Comment:
`LmdbStream::commit()` clears `dirty_` before the LMDB transaction is
begun/committed. If `mdb_txn_begin`, `mdb_put`, or `mdb_txn_commit` fails, the
buffered data remains in `value_` but `dirty_` stays false, so later
`close()`/`commit()` calls will silently skip retrying the write.
##########
extensions/lmdb/LmdbStream.cpp:
##########
@@ -0,0 +1,116 @@
+/**
+ *
+ * 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 "LmdbStream.h"
+
+#include <string>
+#include <utility>
+
+#include "io/validation.h"
+
+namespace org::apache::nifi::minifi::io {
+
+LmdbStream::LmdbStream(std::string path, MDB_env* lmdb_env, MDB_dbi*
lmdb_handle, bool write_enable)
+ : BaseStreamImpl(),
+ path_(std::move(path)),
+ write_enable_(write_enable),
+ lmdb_env_(lmdb_env),
+ lmdb_handle_(lmdb_handle),
+ exists_(loadValue()) {}
+
+bool LmdbStream::loadValue() {
+ MDB_val key{path_.size(), const_cast<char*>(path_.data())};
+ MDB_val value{};
+
+ MDB_txn* txn = nullptr;
+ if (const int rc = mdb_txn_begin(lmdb_env_, nullptr, MDB_RDONLY, &txn); rc
!= MDB_SUCCESS) {
+ logger_->log_error("Failed to begin LMDB read transaction in loadValue:
{}", mdb_strerror(rc));
+ return false;
+ }
+ auto guard = gsl::finally([txn] { mdb_txn_abort(txn); });
+
+ const auto rc = mdb_get(txn, *lmdb_handle_, &key, &value);
+ if (rc == MDB_SUCCESS) {
+ value_ = std::string(static_cast<char*>(value.mv_data), value.mv_size);
+ return true;
+ } else if (rc != MDB_NOTFOUND) {
+ logger_->log_error("Failed to get value from LMDB database: {}",
mdb_strerror(rc));
+ }
+ return false;
+}
+
+void LmdbStream::close() {
+ commit();
+}
+
+bool LmdbStream::commit() {
+ if (!write_enable_ || !dirty_) { return false; }
+ dirty_ = false;
+
+ MDB_txn* txn = nullptr;
+ auto rc = mdb_txn_begin(lmdb_env_, nullptr, 0, &txn);
+ if (rc != MDB_SUCCESS) {
+ logger_->log_error("Failed to begin LMDB transaction in close: {}",
mdb_strerror(rc));
+ return false;
+ }
+
+ MDB_val key{path_.size(), const_cast<char*>(path_.data())};
+ MDB_val val{value_.size(), const_cast<char*>(value_.data())};
+ rc = mdb_put(txn, *lmdb_handle_, &key, &val, 0);
+ if (rc != MDB_SUCCESS) {
+ logger_->log_error("Failed to put value in LMDB database during close:
{}", mdb_strerror(rc));
+ mdb_txn_abort(txn);
+ return false;
+ }
+
+ rc = mdb_txn_commit(txn);
+ if (rc != MDB_SUCCESS) {
+ logger_->log_error("Failed to commit LMDB transaction during close: {}",
mdb_strerror(rc));
+ return false;
+ }
+ return true;
+}
+
+void LmdbStream::seek(size_t offset) {
+ offset_ = offset;
+}
+
+size_t LmdbStream::tell() const {
+ return offset_;
+}
+
+size_t LmdbStream::write(const uint8_t* value, size_t size) {
+ if (!write_enable_) { return STREAM_ERROR; }
+ if (size != 0 && IsNullOrEmpty(value)) { return STREAM_ERROR; }
+ value_.append(reinterpret_cast<const char*>(value), size);
+ dirty_ = true;
Review Comment:
`LmdbStream::write()` buffers the entire value in a `std::string` and
appends on each call. For large flow files written in chunks this can cause
high peak memory usage and quadratic-time growth due to repeated realloc/copy.
If this repository is intended for large content, consider a design that writes
directly to LMDB (e.g., keep a write txn open and use `MDB_RESERVE`) to avoid
full buffering and repeated copies.
##########
extensions/lmdb/LmdbContentRepository.cpp:
##########
@@ -0,0 +1,283 @@
+/**
+ * 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 "LmdbContentRepository.h"
+
+#include <filesystem>
+#include <iterator>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "LmdbStream.h"
+#include "core/Resource.h"
+#include "lmdb.h"
+#include "minifi-cpp/Exception.h"
+#include "minifi-cpp/utils/gsl.h"
+#include "utils/Locations.h"
+
+namespace org::apache::nifi::minifi::core::repository {
+
+LmdbContentRepository::Session::Session(std::shared_ptr<ContentRepository>
repository) : BufferedContentSession(std::move(repository)) {}
+
+void LmdbContentRepository::Session::commit() {
+ auto lmdb_content_repository =
std::dynamic_pointer_cast<LmdbContentRepository>(repository_);
+ if (!lmdb_content_repository) { throw Exception(REPOSITORY_EXCEPTION,
"Session's repository is not an LmdbContentRepository"); }
+
+ const auto writeResource = [&lmdb_content_repository](const
std::shared_ptr<ResourceClaim>& resource_claim, const
std::shared_ptr<io::BaseStream>& stream, bool is_append) {
+ auto outStream = lmdb_content_repository->write(*resource_claim,
is_append);
+ if (outStream == nullptr) { throw Exception(REPOSITORY_EXCEPTION,
"Couldn't open the underlying resource for write: " +
resource_claim->getContentFullPath()); }
+ const auto size = stream->size();
+ if (outStream->write(stream->getBuffer()) != size) {
+ throw Exception(REPOSITORY_EXCEPTION, "Failed to write " +
std::string(is_append ? "appended" : "new") + " resource: " +
resource_claim->getContentFullPath());
+ }
+ auto lmdb_out_stream =
std::dynamic_pointer_cast<io::LmdbStream>(outStream);
+ if (lmdb_out_stream == nullptr) { throw Exception(REPOSITORY_EXCEPTION,
"Couldn't cast output stream to LmdbStream for commit: " +
resource_claim->getContentFullPath()); }
+ if (!lmdb_out_stream->commit()) { throw Exception(REPOSITORY_EXCEPTION,
"Failed to commit " + std::string(is_append ? "appended" : "new") + " resource:
" + resource_claim->getContentFullPath()); }
+ };
+
+ for (const auto& resource : managed_resources_) {
+ writeResource(resource.first, resource.second, false);
+ }
+
+ for (const auto& resource : append_state_) {
+ writeResource(resource.first, resource.second.stream, true);
+ }
+
+ managed_resources_.clear();
+ append_state_.clear();
+}
+
+bool LmdbContentRepository::initialize(const
std::shared_ptr<minifi::Configure>& configuration) {
+ if (const int rc = mdb_env_create(&lmdb_env_)) {
+ logger_->log_error("Failed to create LMDB environment: {}",
mdb_strerror(rc));
+ return false;
+ }
+
+ // Reserve virtual address space for the DB file (max size it can grow to)
+ const auto max_db_size =
configuration->get(Configure::nifi_content_repository_lmdb_max_db_size) |
utils::andThen([](auto max_db_size_str) -> std::optional<uint64_t> {
+ if (max_db_size_str.empty()) { return std::nullopt; }
+ return parsing::parseDataSize(max_db_size_str) |
utils::orThrow(fmt::format("{} was set to invalid value: '{}'",
Configure::nifi_content_repository_lmdb_max_db_size, max_db_size_str));
+ }) | utils::orElse([] {
+ // Default to 10 GB if the property is not set
+ return std::make_optional<uint64_t>(10ULL * 1024 * 1024 * 1024);
+ });
+
+ if (!max_db_size) {
+ logger_->log_error("Invalid max DB size configuration for LMDB Content
Repository");
+ mdb_env_close(lmdb_env_);
+ lmdb_env_ = nullptr;
+ return false;
+ }
+
+ logger_->log_info("Setting LMDB max DB size to {} bytes", *max_db_size);
+ mdb_env_set_mapsize(lmdb_env_, gsl::narrow<size_t>(*max_db_size));
Review Comment:
The return value of `mdb_env_set_mapsize` is ignored. If setting the map
size fails (e.g., invalid size or platform limitation), initialization will
proceed and later operations may fail in hard-to-debug ways.
##########
extensions/lmdb/CMakeLists.txt:
##########
@@ -0,0 +1,37 @@
+#
+# 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.
+#
+
+if (NOT (ENABLE_ALL OR ENABLE_LMDB))
+ return()
+endif()
+
+include(LMDB)
+
+include(${CMAKE_SOURCE_DIR}/extensions/ExtensionHeader.txt)
+
+file(GLOB SOURCES "*.cpp")
+
+add_minifi_library(minifi-lmdb SHARED ${SOURCES})
+
+target_include_directories(minifi-lmdb PUBLIC ${LMDB_INCLUDE_DIR})
+target_link_libraries(minifi-lmdb PUBLIC lmdb)
+target_link_libraries(minifi-lmdb PUBLIC minifi-api minifi-extension-framework)
+target_link_libraries(minifi-lmdb PRIVATE $<LINK_ONLY:core-minifi>)
+
+register_extension(minifi-lmdb "LMDB" LMDB "This Enables persistent
provenance, flowfile, and content repositories using LMDB"
"extensions/lmdb/tests")
Review Comment:
The extension description claims it enables provenance and flowfile
repositories too, but this PR appears to implement only an LMDB-backed
*content* repository. This description is user-visible and should match the
actual capabilities to avoid confusion.
##########
thirdparty/lmdb/add-cmake-file.patch:
##########
@@ -0,0 +1,29 @@
+lmdb only comes with a Makefile which only works on Linux so we create our own
CMakeLists.txt file to be platform independent
+
+diff -rupN orig/libraries/liblmdb/CMakeLists.txt
patched/libraries/liblmdb/CMakeLists.txt
+--- orig/libraries/liblmdb/CMakeLists.txt 2026-06-12 13:02:00.868592927
+0200
++++ patched/libraries/liblmdb/CMakeLists.txt 2026-06-12 11:51:51.369241534
+0200
+@@ -0,0 +1,23 @@
++cmake_minimum_required(VERSION 3.25)
++project(lmdb C)
++
++if(MSVC)
++ add_compile_options(/W3 /WX-)
++else()
++ add_compile_options(-W -Wall -Wno-unused-parameter -Wbad-function-cast
-Wuninitialized -O2 -g)
Review Comment:
The injected LMDB CMakeLists forces optimization/debug flags via
`add_compile_options(... -O2 -g)`, which overrides the parent project's build
type and toolchain flags for this third-party dependency. This can lead to
surprising debug/release mixes and makes builds harder to reproduce.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
