Address RocksDBStore review comments

versioned/tiered/rocksdb/src/main/java/com/dremio/nessie/versioned/store/rocksdb/Function.java

@@ -27,8 +27,16 @@
  */
 @Immutable
 abstract class Function {
-  static final String EQUALS = "equals";
-  static final String SIZE = "size";
+  /**
+   * An enum encapsulating.
+   */
+  enum Operator {
+    // An operator comparing the equality of entities.
+    EQUALS,
+
+    // An operator comparing the size of entities.
+    SIZE
+  }
 
   /**
    * Compares for equality with a provided Function object.
@@ -56,7 +64,7 @@ public int hashCode() {
     return Objects.hash(getOperator(), getPath(), getValue());
   }
 
-  abstract String getOperator();
+  abstract Operator getOperator();
 
   abstract ExpressionPath getPath();
 
@@ -73,7 +81,7 @@ ExpressionPath.NameSegment getRootPathAsNameSegment() {
    */
   boolean isRootNameSegmentChildlessAndEquals() {
     return !getRootPathAsNameSegment().getChild().isPresent()
-      && getOperator().equals(Function.EQUALS);
+      && getOperator().equals(Operator.EQUALS);
   }
 
   /**

versioned/tiered/rocksdb/src/main/java/com/dremio/nessie/versioned/store/rocksdb/RocksBaseValue.java

@@ -87,7 +87,7 @@ Entity toEntity(Stream<Id> idStream, int position) {
    */
   boolean evaluateStream(Function function, Stream<Id> stream) {
     // EQUALS will either compare a specified position or the whole stream as a List.
-    if (function.getOperator().equals(Function.EQUALS)) {
+    if (function.getOperator().equals(Function.Operator.EQUALS)) {
       final ExpressionPath.PathSegment pathSegment = function.getPath().getRoot().getChild().orElse(null);
       if (pathSegment == null) {
         return toEntity(stream).equals(function.getValue());
@@ -97,7 +97,7 @@ boolean evaluateStream(Function function, Stream<Id> stream) {
           return toEntity(stream, position).equals(function.getValue());
         }
       }
-    } else if (function.getOperator().equals(Function.SIZE)) {
+    } else if (function.getOperator().equals(Function.Operator.SIZE)) {
       return (stream.count() == function.getValue().getNumber());
     }
 

versioned/tiered/rocksdb/src/main/java/com/dremio/nessie/versioned/store/rocksdb/RocksDBStore.java

@@ -171,7 +171,7 @@ public <C extends BaseValue<C>> boolean putIfAbsent(SaveOp<C> saveOp) {
     final ColumnFamilyHandle columnFamilyHandle = getColumnFamilyHandle(saveOp.getType());
     try (final Transaction transaction = rocksDB.beginTransaction(new WriteOptions())) {
       // Get exclusive access to key if it exists.
-      final byte[] buffer = transaction.getForUpdate(new ReadOptions(), columnFamilyHandle, saveOp.getId().toBytes(), true);
+      final byte[] buffer = getForUpdate(transaction, columnFamilyHandle, saveOp.getId());
       if (null == buffer) {
         transaction.put(columnFamilyHandle, saveOp.getId().toBytes(), RocksSerDe.serializeWithConsumer(saveOp));
         transaction.commit();
@@ -189,19 +189,9 @@ public <C extends BaseValue<C>> void put(SaveOp<C> saveOp, Optional<ConditionExp
     final ColumnFamilyHandle columnFamilyHandle = getColumnFamilyHandle(saveOp.getType());
 
     try (final Transaction transaction = rocksDB.beginTransaction(new WriteOptions())) {
-      if (condition.isPresent()) {
-        final byte[] buffer = transaction.getForUpdate(new ReadOptions(), columnFamilyHandle, saveOp.getId().toBytes(), true);
-        if (null == buffer) {
-          throw new ConditionFailedException("Unable to load item with ID: " + saveOp.getId());
-        }
-        final RocksBaseValue<C> consumer = RocksSerDe.getConsumer(saveOp.getType());
-        RocksSerDe.deserializeToConsumer(saveOp.getType(), buffer, consumer);
-
-        if (!consumer.evaluate(translate(condition.get()))) {
-          throw new ConditionFailedException("Condition failed during put operation");
-        }
+      if (!isConditionExpressionValid(transaction, columnFamilyHandle, saveOp.getId(), saveOp.getType(), condition, "load")) {
+        throw new ConditionFailedException("Condition failed during put operation");
       }
-
       transaction.put(columnFamilyHandle, saveOp.getId().toBytes(), RocksSerDe.serializeWithConsumer(saveOp));
       transaction.commit();
     } catch (RocksDBException e) {
@@ -214,24 +204,10 @@ public <C extends BaseValue<C>> boolean delete(ValueType<C> type, Id id, Optiona
     final ColumnFamilyHandle columnFamilyHandle = getColumnFamilyHandle(type);
 
     try (final Transaction transaction = rocksDB.beginTransaction(new WriteOptions())) {
-      final byte[] value = transaction.getForUpdate(new ReadOptions(), columnFamilyHandle, id.toBytes(), true);
-
-      if (null == value) {
-        throw new NotFoundException("No value was found for the given id to delete.");
+      if (!isConditionExpressionValid(transaction, columnFamilyHandle, id, type, condition, "delete")) {
+        LOGGER.debug("Condition failed during delete operation.");
+        return false;
       }
-
-      if (condition.isPresent()) {
-        final RocksBaseValue<C> consumer = RocksSerDe.getConsumer(type);
-        // TODO: Does the critical section need to be locked?
-        // TODO: Critical Section - evaluating the condition expression and deleting the entity.
-        // Check if condition expression is valid.
-        RocksSerDe.deserializeToConsumer(type, value, consumer);
-        if (!(consumer.evaluate(translate(condition.get())))) {
-          LOGGER.debug("Condition failed during delete operation.");
-          return false;
-        }
-      }
-
       transaction.delete(columnFamilyHandle, id.toBytes());
       transaction.commit();
       return true;
@@ -336,6 +312,34 @@ static List<Function> translate(ConditionExpression conditionExpression) {
       .collect(Collectors.toList());
   }
 
+  private <C extends BaseValue<C>> boolean isConditionExpressionValid(Transaction transaction, ColumnFamilyHandle columnFamilyHandle,
+                                                                      Id id, ValueType type, Optional<ConditionExpression> condition,
+                                                                      String operation) throws RocksDBException {
+    if (condition.isPresent()) {
+      final byte[] value = getAndCheckValue(transaction, columnFamilyHandle, id, operation);
+      final RocksBaseValue<C> consumer = RocksSerDe.getConsumer(type);
+      RocksSerDe.deserializeToConsumer(type, value, consumer);
+      if (!(consumer.evaluate(translate(condition.get())))) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private byte[] getAndCheckValue(Transaction transaction, ColumnFamilyHandle columnFamilyHandle,
+                                  Id id, String operation) throws RocksDBException {
+    final byte[] value = getForUpdate(transaction, columnFamilyHandle, id);
+    if (null == value) {
+      throw new ConditionFailedException("Unable to " + operation + " item with ID: " + id);
+    }
+    return value;
+  }
+
+  private byte[] getForUpdate(Transaction transaction, ColumnFamilyHandle columnFamilyHandle, Id id)
+      throws RocksDBException {
+    return transaction.getForUpdate(new ReadOptions(), columnFamilyHandle, id.toBytes(), true);
+  }
+
   private <C extends BaseValue<C>> ColumnFamilyHandle getColumnFamilyHandle(ValueType<C> valueType) {
     final ColumnFamilyHandle columnFamilyHandle = valueTypeToColumnFamily.get(valueType);
     if (null == columnFamilyHandle) {

versioned/tiered/rocksdb/src/main/java/com/dremio/nessie/versioned/store/rocksdb/RocksDBValueVisitor.java

@@ -68,12 +68,12 @@ public Function visit(ExpressionFunction value) {
       case EQUALS:
         // Special case SIZE, as the object representation is not contained in one level of ExpressionFunction.
         if (isSize(arguments.get(0))) {
-          return ImmutableFunction.builder().operator(Function.SIZE)
+          return ImmutableFunction.builder().operator(Function.Operator.SIZE)
             .path(arguments.get(0).getFunction().getArguments().get(0).accept(EXPRESSION_PATH_VALUE_VISITOR))
             .value(arguments.get(1).getValue()).build();
         }
 
-        return ImmutableFunction.builder().operator(Function.EQUALS)
+        return ImmutableFunction.builder().operator(Function.Operator.EQUALS)
           .path(arguments.get(0).accept(EXPRESSION_PATH_VALUE_VISITOR))
           .value(arguments.get(1).getValue()).build();
       default:

versioned/tiered/rocksdb/src/main/java/com/dremio/nessie/versioned/store/rocksdb/RocksFragment.java

@@ -79,9 +79,9 @@ public boolean evaluateFunction(Function function) {
         if (function.getRootPathAsNameSegment().getChild().isPresent()) {
           return false;
         }
-        if (function.getOperator().equals(Function.EQUALS)) {
+        if (function.getOperator().equals(Function.Operator.EQUALS)) {
           return keysAsEntityList(keys).equals(function.getValue());
-        } else if (function.getOperator().equals(Function.SIZE)) {
+        } else if (function.getOperator().equals(Function.Operator.SIZE)) {
           return (keys.count() == function.getValue().getNumber());
         } else {
           return false;

versioned/tiered/rocksdb/src/main/java/com/dremio/nessie/versioned/store/rocksdb/RocksL1.java

@@ -109,7 +109,7 @@ public boolean evaluateFunction(Function function) {
       case KEY_LIST:
         return false;
       case INCREMENTAL_KEY_LIST:
-        if (!nameSegment.getChild().isPresent() || !function.getOperator().equals(Function.EQUALS)) {
+        if (!nameSegment.getChild().isPresent() || !function.getOperator().equals(Function.Operator.EQUALS)) {
           return false;
         }
         final String childName = nameSegment.getChild().get().asName().getName();

versioned/tiered/rocksdb/src/main/java/com/dremio/nessie/versioned/store/rocksdb/RocksRef.java

@@ -15,8 +15,6 @@
  */
 package com.dremio.nessie.versioned.store.rocksdb;
 
-import static com.dremio.nessie.versioned.store.rocksdb.Function.SIZE;
-
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Consumer;
@@ -141,10 +139,10 @@ private boolean evaluateTag(Function function) {
         return (function.isRootNameSegmentChildlessAndEquals()
           && type.equals(Type.getType(function.getValue().getString())));
       case NAME:
-        return (function.getOperator().equals(Function.EQUALS)
+        return (function.getOperator().equals(Function.Operator.EQUALS)
           && name.equals(function.getValue().getString()));
       case COMMIT:
-        return (function.getOperator().equals(Function.EQUALS)
+        return (function.getOperator().equals(Function.Operator.EQUALS)
           && commit.toEntity().equals(function.getValue()));
       default:
         return false;