MINIFICPP-2849 Implement LMDB based content repository#2201
Conversation
| 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); |
There was a problem hiding this comment.
LMDB does not have an append function when writing a value like RocksDB's Merge function, so instead of rereading the original value, appending to it, then writing back the new value, all the writes are buffered until the stream is closed, that's when the actual write and commit happens. Currently all content repository streams are used either for write-only or read-only use cases, so there should be no use case where reads and writes are mixed. This should be addressed in a separate PR to change the content repository interface to use separate OutputStream and InputStream types for reads and writes to enforce this, which would also result in separate LmdbInputStream and LmdbOutputStream types (same for RocksDB).
|
|
||
| if (WIN32) | ||
| get_directory_property(MINIFI_SAVED_COMPILE_DEFS COMPILE_DEFINITIONS) | ||
| remove_definitions(-DWIN32_LEAN_AND_MEAN) |
There was a problem hiding this comment.
can you explain why you did this?
There was a problem hiding this comment.
LMDB fails to compile on Windows when WIN32_LEAN_AND_MEAN is defined, which is automatically added to the compile definitions in CMakeLists.txt so it is included in all thirdparties used with FetchContent, so it needs to be removed separately for LMDB on Windows.
There was a problem hiding this comment.
I'd like to make absolutely sure that we only undefine WIN32_LEAN_AND_MEAN while compiling LMDB, or find another workaround. If you look up what it does, it prevents windows.h from including a bunch of additional headers that usually end up unused, and may end up conflicting with other headers, while also increasing compile times. Alternatively, we could look up the transitively included headers that LMDB relies on, and include them explicitly with target_compile_options(target PRIVATE -include foo.h)
Ideally Microsoft would've made the lightweight header the default, with the option to opt in to additional features, but for historical reasons, the heavyweight header is the default, and you can opt out of the extra features. (bloat)
There was a problem hiding this comment.
When I checked this according to my understanding this change should only impact the lmdb subdirectory, so it should not impact any other compilation. Anyway I can still check if there is another workaround.
There was a problem hiding this comment.
There were 2 issues:
- Defining WIN32_LEAN_AND_MEAN removed
winternl.hneeded forNTSTATUSusage, that needed to be added explicitly - After including
winternl.hthe function pointer names inmdb.cclashed with the Windows symbols, those needed to be renamed
Added patch in: 52d93b2
There was a problem hiding this comment.
Pull request overview
Implements an LMDB-backed Content Repository extension for MiNiFi C++ (as an alternative to the RocksDB-based DatabaseContentRepository), including build integration, configuration plumbing, and dedicated unit tests.
Changes:
- Adds a new
minifi-lmdbextension implementingLmdbContentRepositoryandLmdbStream, plus LMDB-focused unit tests. - Wires LMDB into configuration (new
nifi.content.repository.lmdb.max.db.sizeproperty) and improves user-facing error reporting/docs. - Introduces a FetchContent-based LMDB third-party build with patching, and enables LMDB in CI/default extensions.
Reviewed changes
Copilot reviewed 22 out of 22 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| thirdparty/lmdb/fix-windows-symbols.patch | Patches upstream LMDB to avoid Windows NT API symbol collisions. |
| thirdparty/lmdb/add-cmake-file.patch | Adds a CMakeLists.txt to LMDB upstream sources for cross-platform builds. |
| minifi-api/include/minifi-cpp/properties/Configuration.h | Adds config key for LMDB max DB size. |
| libminifi/test/libtest/unit/TestBase.h | Adds virtual destructor to TestController for safer polymorphic use. |
| libminifi/src/core/RepositoryFactory.cpp | Adds clearer error log when LMDB extension is missing. |
| libminifi/src/Configuration.cpp | Registers LMDB max DB size property validator. |
| extensions/rocksdb-repos/tests/ContentSessionTests.cpp | Renames member and fixes destructor override for test controller. |
| extensions/lmdb/tests/LmdbStreamTests.cpp | Adds unit tests for LmdbStream read/write/commit behavior. |
| extensions/lmdb/tests/LmdbContentSessionTests.cpp | Adds session semantics tests for LMDB content repository. |
| extensions/lmdb/tests/LmdbContentRepositoryTests.cpp | Adds repository init/exists/read/remove/orphan tests for LMDB. |
| extensions/lmdb/tests/CMakeLists.txt | Adds CMake rules to build/register LMDB unit tests. |
| extensions/lmdb/LmdbStream.h | Introduces LMDB-backed stream abstraction. |
| extensions/lmdb/LmdbStream.cpp | Implements LMDB stream read/write/commit logic. |
| extensions/lmdb/LmdbContentRepository.h | Declares LMDB-backed ContentRepository implementation and session type. |
| extensions/lmdb/LmdbContentRepository.cpp | Implements LMDB-backed content repository lifecycle, GC/orphaning, stats. |
| extensions/lmdb/CMakeLists.txt | Adds extension build and registration for minifi-lmdb. |
| CONFIGURE.md | Documents LMDB content repository option and caveats. |
| conf/minifi.properties.in | Adds commented LMDB max DB size property template. |
| CMakeLists.txt | Adds minifi-lmdb to default enabled extensions list. |
| cmake/MiNiFiOptions.cmake | Adds ENABLE_LMDB option. |
| cmake/LMDB.cmake | FetchContent integration for building LMDB and applying local patches. |
| .github/workflows/ci.yml | Enables LMDB in CI build matrices. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
|
||
| namespace { | ||
|
|
||
| std::shared_ptr<io::OutputStream> operator<<(std::shared_ptr<io::OutputStream> stream, const std::string& str) { |
There was a problem hiding this comment.
nitpick, but stream should be const & (also in >>)
| if (const int rc = mdb_env_create(&lmdb_env_)) { | ||
| throw std::runtime_error("Failed to create LMDB environment: " + std::string(mdb_strerror(rc))); | ||
| } | ||
| if (const int rc = mdb_env_set_mapsize(lmdb_env_, 100ULL * 1024 * 1024); rc != MDB_SUCCESS) { | ||
| throw std::runtime_error("Failed to set LMDB map size: " + std::string(mdb_strerror(rc))); | ||
| } | ||
|
|
||
| if (const int rc = mdb_env_open(lmdb_env_, db_path_.c_str(), MDB_NOTLS, 0664)) { | ||
| throw std::runtime_error("Failed to open LMDB environment " + db_path_ + ": " + std::string(mdb_strerror(rc))); | ||
| } |
There was a problem hiding this comment.
I would add ; rc != MDB_SUCCESS in lines 29 and 36, too
| 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()); } |
There was a problem hiding this comment.
should we check this before the write()?
| 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()); } | ||
| }; |
There was a problem hiding this comment.
Very minor nitpick, but these throw Exception... lines look too long to me for a single-line format. I would put all of them on separate lines as you did at lines 46-48.
| } | ||
|
|
||
| bool LmdbContentRepository::initialize(const std::shared_ptr<minifi::Configure>& configuration) { | ||
| if (const int rc = mdb_env_create(&lmdb_env_)) { |
There was a problem hiding this comment.
I would add the explicit comparison here:
| if (const int rc = mdb_env_create(&lmdb_env_)) { | |
| if (const int rc = mdb_env_create(&lmdb_env_); rc != MDB_SUCCESS) { |
| 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; | ||
| } |
There was a problem hiding this comment.
I don't think max_db_size can be nullopt here: if the property was unset or empty, then we are using the default value, and if it is an invalid number, then we throw earlier. Maybe we should set max_db_size to nullopt instead of throwing?
| while (rc == MDB_SUCCESS) { | ||
| std::string key_string = std::string(static_cast<char*>(key.mv_data), key.mv_size); | ||
|
|
||
| std::lock_guard<std::mutex> lock(count_map_mutex_); |
There was a problem hiding this comment.
Should we lock the count_map_mutex_ for the whole duration of the iteration? Allowing modifications to count_map_ while we are checking the orphans makes it difficult to reason about what is happening.
|
|
||
| class LmdbStream : public io::BaseStreamImpl { | ||
| public: | ||
| explicit LmdbStream(std::string path, MDB_env* lmdb_env, MDB_dbi* lmdb_handle, bool write_enable = false); |
There was a problem hiding this comment.
I can't see any place where we use the default, and it's better not to have it, so I would remove it:
| explicit LmdbStream(std::string path, MDB_env* lmdb_env, MDB_dbi* lmdb_handle, bool write_enable = false); | |
| explicit LmdbStream(std::string path, MDB_env* lmdb_env, MDB_dbi* lmdb_handle, bool write_enable); |
| bool LmdbStream::commit() { | ||
| if (!write_enable_ || !dirty_) { return 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; | ||
| } | ||
|
|
||
| dirty_ = false; | ||
| return true; | ||
| } |
There was a problem hiding this comment.
nitpick: the error logs should say "commit" instead of "close"
|
|
||
| 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; } |
There was a problem hiding this comment.
value is just a pointer, so IsNullOrEmpty is just a null check; I would change this line to
| if (size != 0 && IsNullOrEmpty(value)) { return STREAM_ERROR; } | |
| if (size != 0 && value == nullptr) { return STREAM_ERROR; } |
to make this clearer
https://issues.apache.org/jira/browse/MINIFICPP-2849
Thank you for submitting a contribution to Apache NiFi - MiNiFi C++.
In order to streamline the review of the contribution we ask you to ensure the following steps have been taken:
For all changes:
Is there a JIRA ticket associated with this PR? Is it referenced in the commit message?
Does your PR title start with MINIFICPP-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
Has your PR been rebased against the latest commit within the target branch (typically main)?
Is your initial contribution a single, squashed commit?
For code changes:
For documentation related changes:
Note:
Please ensure that once the PR is submitted, you check GitHub Actions CI results for build issues and submit an update to your PR as soon as possible.