arpadboda commented on a change in pull request #1071: URL: https://github.com/apache/nifi-minifi-cpp/pull/1071#discussion_r644741254
########## File path: extensions/rocksdb-repos/database/OpenRocksDb.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 "OpenRocksDb.h" +#include "ColumnHandle.h" +#include "RocksDbInstance.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace internal { + +OpenRocksDb::OpenRocksDb(RocksDbInstance& db, gsl::not_null<std::shared_ptr<rocksdb::DB>> impl, gsl::not_null<std::shared_ptr<ColumnHandle>> column) + : db_(&db), impl_(std::move(impl)), column_(std::move(column)) {} + +rocksdb::Status OpenRocksDb::Put(const rocksdb::WriteOptions& options, const rocksdb::Slice& key, const rocksdb::Slice& value) { + rocksdb::Status result = impl_->Put(options, column_->handle.get(), key, value); + if (result == rocksdb::Status::NoSpace()) { + db_->invalidate(); + } + return result; +} + +rocksdb::Status OpenRocksDb::Get(const rocksdb::ReadOptions& options, const rocksdb::Slice& key, std::string* value) { + rocksdb::Status result = impl_->Get(options, column_->handle.get(), key, value); + if (result == rocksdb::Status::NoSpace()) { + db_->invalidate(); + } + return result; +} + +std::vector<rocksdb::Status> OpenRocksDb::MultiGet(const rocksdb::ReadOptions& options, const std::vector<rocksdb::Slice>& keys, std::vector<std::string>* values) { + std::vector<rocksdb::Status> results = impl_->MultiGet( + options, std::vector<rocksdb::ColumnFamilyHandle*>(keys.size(), column_->handle.get()), keys, values); + for (const auto& result : results) { + if (result == rocksdb::Status::NoSpace()) { Review comment: How can we get no space when we try to get? On the other side, is that the only error to trigger breaking? ########## File path: extensions/rocksdb-repos/controllers/RocksDbPersistableKeyValueStoreService.cpp ########## @@ -58,18 +58,18 @@ void RocksDbPersistableKeyValueStoreService::onEnable() { } db_.reset(); - rocksdb::Options options; - options.create_if_missing = true; - options.use_direct_io_for_flush_and_compaction = true; - options.use_direct_reads = true; + auto db_opts = [] (internal::Writable<rocksdb::DBOptions>& db_opts) { + db_opts.set(&rocksdb::DBOptions::create_if_missing, true); + db_opts.set(&rocksdb::DBOptions::use_direct_io_for_flush_and_compaction, true); + db_opts.set(&rocksdb::DBOptions::use_direct_reads, true); + }; // Use the same buffer settings as the FlowFileRepository - options.write_buffer_size = 8 << 20; - options.max_write_buffer_number = 20; - options.min_write_buffer_number_to_merge = 1; - if (!always_persist_) { - options.manual_wal_flush = true; - } - db_ = utils::make_unique<minifi::internal::RocksDatabase>(options, directory_); + auto cf_opts = [] (minifi::internal::Writable<rocksdb::ColumnFamilyOptions>& cf_opts) { + cf_opts.set(&rocksdb::ColumnFamilyOptions::write_buffer_size, 8ULL << 20U); + cf_opts.set<int>(&rocksdb::ColumnFamilyOptions::max_write_buffer_number, 20); Review comment: Is this the cf used to store the state of processors? In case it is, I would reduce the numbers here, that definitely doesn't need such huge buffers. ########## File path: extensions/rocksdb-repos/database/RocksDatabase.cpp ########## @@ -0,0 +1,91 @@ +/** + * + * 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 "RocksDatabase.h" +#include "core/logging/LoggerConfiguration.h" +#include "utils/StringUtils.h" +#include "RocksDbInstance.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace internal { + +std::shared_ptr<core::logging::Logger> RocksDatabase::logger_ = core::logging::LoggerFactory<RocksDatabase>::getLogger(); + +std::unique_ptr<RocksDatabase> RocksDatabase::create(const DBOptionsPatch& db_options_patch, const ColumnFamilyOptionsPatch& cf_options_patch, const std::string& uri, RocksDbMode mode) { + const std::string scheme = "minifidb://"; + + logger_->log_trace("Acquiring database handle '%s'", uri); + std::string db_path = uri; + std::string db_column = "default"; + if (utils::StringUtils::startsWith(uri, scheme)) { + const std::string path = uri.substr(scheme.length()); + logger_->log_trace("RocksDB scheme is detected in '%s'", uri); + // last segment is treated as the column name + std::string::size_type pos = path.find_last_of('/'); + if (pos == std::string::npos) { + pos = path.find_last_of('\\'); + } + if (pos == std::string::npos) { + logger_->log_error("Couldn't detect the column name in '%s'", uri); + return nullptr; + } + db_path = path.substr(0, pos); + db_column = path.substr(pos + 1); + logger_->log_trace("Using column '%s' in rocksdb database '%s'", db_column, db_path); Review comment: I think these might be more verbose than trace as they are only printed once and can be useful. ########## File path: extensions/rocksdb-repos/database/OpenRocksDb.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 "OpenRocksDb.h" +#include "ColumnHandle.h" +#include "RocksDbInstance.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace internal { + +OpenRocksDb::OpenRocksDb(RocksDbInstance& db, gsl::not_null<std::shared_ptr<rocksdb::DB>> impl, gsl::not_null<std::shared_ptr<ColumnHandle>> column) + : db_(&db), impl_(std::move(impl)), column_(std::move(column)) {} + +rocksdb::Status OpenRocksDb::Put(const rocksdb::WriteOptions& options, const rocksdb::Slice& key, const rocksdb::Slice& value) { + rocksdb::Status result = impl_->Put(options, column_->handle.get(), key, value); + if (result == rocksdb::Status::NoSpace()) { + db_->invalidate(); + } + return result; +} + +rocksdb::Status OpenRocksDb::Get(const rocksdb::ReadOptions& options, const rocksdb::Slice& key, std::string* value) { + rocksdb::Status result = impl_->Get(options, column_->handle.get(), key, value); + if (result == rocksdb::Status::NoSpace()) { + db_->invalidate(); + } + return result; +} + +std::vector<rocksdb::Status> OpenRocksDb::MultiGet(const rocksdb::ReadOptions& options, const std::vector<rocksdb::Slice>& keys, std::vector<std::string>* values) { + std::vector<rocksdb::Status> results = impl_->MultiGet( + options, std::vector<rocksdb::ColumnFamilyHandle*>(keys.size(), column_->handle.get()), keys, values); + for (const auto& result : results) { + if (result == rocksdb::Status::NoSpace()) { + db_->invalidate(); + break; + } + } + return results; +} + +rocksdb::Status OpenRocksDb::Write(const rocksdb::WriteOptions& options, internal::WriteBatch* updates) { + rocksdb::Status result = impl_->Write(options, &updates->impl_); + if (result == rocksdb::Status::NoSpace()) { Review comment: Idea: maybe this should be moved to a HandleResult(result) func or something like that, so we don't need to copy-paste and easier to be extended with different results. ########## File path: extensions/rocksdb-repos/DatabaseContentRepository.cpp ########## @@ -41,14 +42,17 @@ bool DatabaseContentRepository::initialize(const std::shared_ptr<minifi::Configu } else { directory_ = configuration->getHome() + "/dbcontentrepository"; } - rocksdb::Options options; - options.create_if_missing = true; - options.use_direct_io_for_flush_and_compaction = true; - options.use_direct_reads = true; - options.merge_operator = std::make_shared<StringAppender>(); - options.error_if_exists = false; - options.max_successive_merges = 0; - db_ = utils::make_unique<minifi::internal::RocksDatabase>(options, directory_); + auto db_opts = [] (internal::Writable<rocksdb::DBOptions>& db_opts) { Review comment: Nitpicking: as these are going to be functions we call later, I would prefer them to be named accordingly. "set_db_opts" for eg. -- 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. For queries about this service, please contact Infrastructure at: [email protected]
