sashapolo commented on a change in pull request #285: URL: https://github.com/apache/ignite-3/pull/285#discussion_r693048114
########## File path: modules/rocksdb-common/src/main/java/org/apache/ignite/internal/rocksdb/RocksUtils.java ########## @@ -0,0 +1,157 @@ +/* + * 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. + */ + +package org.apache.ignite.internal.rocksdb; + +import java.nio.file.Path; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; +import org.apache.ignite.lang.IgniteInternalException; +import org.rocksdb.EnvOptions; +import org.rocksdb.Options; +import org.rocksdb.ReadOptions; +import org.rocksdb.RocksDBException; +import org.rocksdb.RocksIterator; +import org.rocksdb.Snapshot; +import org.rocksdb.SstFileWriter; + +/** + * RocksDB utility functions. + */ +public class RocksUtils { + /** + * Creates an SST file for the column family. + * + * @param columnFamily Column family. + * @param snapshot Point-in-time snapshot. + * @param path Directory to put the SST file in. + * @param snapshotExecutor Snapshot executor. + * @return Future representing pending completion of the operation. + */ + public static CompletableFuture<Void> createSstFile( Review comment: I would suggest making this method synchronous and dispatch it onto executors on the caller's side ########## File path: modules/rocksdb-common/src/main/java/org/apache/ignite/internal/rocksdb/RocksUtils.java ########## @@ -0,0 +1,157 @@ +/* + * 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. + */ + +package org.apache.ignite.internal.rocksdb; + +import java.nio.file.Path; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; +import org.apache.ignite.lang.IgniteInternalException; +import org.rocksdb.EnvOptions; +import org.rocksdb.Options; +import org.rocksdb.ReadOptions; +import org.rocksdb.RocksDBException; +import org.rocksdb.RocksIterator; +import org.rocksdb.Snapshot; +import org.rocksdb.SstFileWriter; + +/** + * RocksDB utility functions. + */ +public class RocksUtils { + /** + * Creates an SST file for the column family. + * + * @param columnFamily Column family. + * @param snapshot Point-in-time snapshot. + * @param path Directory to put the SST file in. + * @param snapshotExecutor Snapshot executor. + * @return Future representing pending completion of the operation. + */ + public static CompletableFuture<Void> createSstFile( + ColumnFamily columnFamily, + Snapshot snapshot, + Path path, + Executor snapshotExecutor + ) { + return CompletableFuture.runAsync(() -> { + try ( + EnvOptions envOptions = new EnvOptions(); + Options options = new Options(); + ReadOptions readOptions = new ReadOptions().setSnapshot(snapshot); + RocksIterator it = columnFamily.newIterator(readOptions); + SstFileWriter sstFileWriter = new SstFileWriter(envOptions, options) + ) { + Path sstFile = path.resolve(columnFamily.name()); + + sstFileWriter.open(sstFile.toString()); + + it.seekToFirst(); + + forEach(it, sstFileWriter::put); + + sstFileWriter.finish(); + } + catch (Throwable t) { + throw new IgniteInternalException("Failed to write snapshot: " + t.getMessage(), t); + } + }, snapshotExecutor); + } + + /** + * Iterates over the given iterator passing key-value pairs to the given consumer and + * checks the iterator's status afterwards. + * + * @param iterator Iterator. + * @param consumer Consumer of key-value pairs. + * @throws RocksDBException If failed. + */ + public static void forEach(RocksIterator iterator, RocksBiConsumer consumer) throws RocksDBException { + for (; iterator.isValid(); iterator.next()) + consumer.accept(iterator.key(), iterator.value()); + + checkIterator(iterator); + } + + /** + * Iterates over the given iterator testing key-value pairs with the given predicate and checks + * the iterator's status afterwards. + * + * @param iterator Iterator. + * @param consumer Consumer of key-value pairs. + * @return {@code true} if a matching key-value pair has been found, {@code false} otherwise. + * @throws RocksDBException If failed. + */ + public static boolean find(RocksIterator iterator, RocksBiPredicate consumer) throws RocksDBException { + for (; iterator.isValid(); iterator.next()) { + boolean result = consumer.test(iterator.key(), iterator.value()); + + if (result) + return true; + } + + checkIterator(iterator); + + return false; + } + + /** + * Checks the status of the iterator and throws an exception if it is not correct. + * + * @param it RocksDB iterator. + * @throws IgniteInternalException if the iterator has an incorrect status. + */ + public static void checkIterator(RocksIterator it) { + try { + it.status(); + } + catch (RocksDBException e) { + throw new IgniteInternalException(e); + } + } + + /** + * BiConsumer that can throw {@link RocksDBException}. + */ + @FunctionalInterface + public interface RocksBiConsumer { Review comment: let's extract these classes to the upper level -- 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]
