Persist: introduce deleteWithReferenced(Obj)

versioned/storage/batching/src/main/java/org/projectnessie/versioned/storage/batching/BatchingPersistImpl.java

@@ -329,6 +329,11 @@ public void deleteObjs(@Nonnull @javax.annotation.Nonnull ObjId[] ids) {
     }
   }
 
+  @Override
+  public boolean deleteWithReferenced(@Nonnull Obj obj) {
+    throw new UnsupportedOperationException();
+  }
+
   @Override
   public boolean deleteConditional(@Nonnull UpdateableObj obj) {
     throw new UnsupportedOperationException();

versioned/storage/bigtable/src/main/java/org/projectnessie/versioned/storage/bigtable/BigTablePersist.java

@@ -566,6 +566,20 @@ public void upsertObjs(@Nonnull Obj[] objs) throws ObjTooLargeException {
     }
   }
 
+  @Override
+  public boolean deleteWithReferenced(@Nonnull Obj obj) {
+    Filter condition =
+        FILTERS
+            .chain()
+            .filter(FILTERS.qualifier().exactMatch(QUALIFIER_OBJ_REFERENCED))
+            .filter(FILTERS.value().exactMatch(copyFromUtf8(Long.toString(obj.referenced()))));
+
+    ConditionalRowMutation conditionalRowMutation =
+        conditionalRowMutation(obj, condition, Mutation.create().deleteRow());
+
+    return backend.client().checkAndMutateRow(conditionalRowMutation);
+  }
+
   @Override
   public boolean deleteConditional(@Nonnull UpdateableObj obj) {
     ConditionalRowMutation conditionalRowMutation =
@@ -591,9 +605,6 @@ public boolean updateConditional(@Nonnull UpdateableObj expected, @Nonnull Updat
   @Nonnull
   private ConditionalRowMutation mutationForConditional(
       @Nonnull UpdateableObj obj, Mutation mutation) {
-    checkArgument(obj.id() != null, "Obj to store must have a non-null ID");
-    ByteString key = dbKey(obj.id());
-
     Filters.ChainFilter objTypeFilter =
         FILTERS
             .chain()
@@ -607,6 +618,15 @@ private ConditionalRowMutation mutationForConditional(
     Filter condition =
         FILTERS.condition(objTypeFilter).then(objVersionFilter).otherwise(FILTERS.block());
 
+    return conditionalRowMutation(obj, condition, mutation);
+  }
+
+  @Nonnull
+  private ConditionalRowMutation conditionalRowMutation(
+      @Nonnull Obj obj, @Nonnull Filter condition, @Nonnull Mutation mutation) {
+    checkArgument(obj.id() != null, "Obj to store must have a non-null ID");
+    ByteString key = dbKey(obj.id());
+
     return ConditionalRowMutation.create(backend.tableObjsId, key)
         .condition(condition)
         .then(mutation);

versioned/storage/cache/src/main/java/org/projectnessie/versioned/storage/cache/CachingPersistImpl.java

@@ -302,6 +302,15 @@ public void deleteObjs(@Nonnull ObjId[] ids) {
     }
   }
 
+  @Override
+  public boolean deleteWithReferenced(@Nonnull Obj obj) {
+    try {
+      return persist.deleteWithReferenced(obj);
+    } finally {
+      cache.remove(obj.id());
+    }
+  }
+
   @Override
   public boolean deleteConditional(@Nonnull UpdateableObj obj) {
     try {

versioned/storage/cassandra/src/main/java/org/projectnessie/versioned/storage/cassandra/CassandraConstants.java

@@ -81,6 +81,17 @@ public final class CassandraConstants {
           + COL_OBJ_VERS
           + "=?";
 
+  static final String DELETE_OBJ_REFERENCED =
+      "DELETE FROM %s."
+          + TABLE_OBJS
+          + " WHERE "
+          + COL_REPO_ID
+          + "=? AND "
+          + COL_OBJ_ID
+          + "=? IF "
+          + COL_OBJ_REFERENCED
+          + "=?";
+
   public static final String INSERT_OBJ_PREFIX =
       "INSERT INTO %s."
           + TABLE_OBJS
@@ -124,7 +135,10 @@ public final class CassandraConstants {
           + " AND "
           + COL_OBJ_ID
           + "=:"
-          + COL_OBJ_ID;
+          + COL_OBJ_ID
+          // IF EXISTS is necessary to prevent writing just the referenced timestamp after an object
+          // has been deleted.
+          + " IF EXISTS";
 
   static final Set<CqlColumn> COLS_OBJS_ALL =
       Stream.concat(

versioned/storage/cassandra/src/main/java/org/projectnessie/versioned/storage/cassandra/CassandraPersist.java

@@ -28,6 +28,7 @@
 import static org.projectnessie.versioned.storage.cassandra.CassandraConstants.COL_REPO_ID;
 import static org.projectnessie.versioned.storage.cassandra.CassandraConstants.DELETE_OBJ;
 import static org.projectnessie.versioned.storage.cassandra.CassandraConstants.DELETE_OBJ_CONDITIONAL;
+import static org.projectnessie.versioned.storage.cassandra.CassandraConstants.DELETE_OBJ_REFERENCED;
 import static org.projectnessie.versioned.storage.cassandra.CassandraConstants.EXPECTED_SUFFIX;
 import static org.projectnessie.versioned.storage.cassandra.CassandraConstants.FETCH_OBJ_TYPE;
 import static org.projectnessie.versioned.storage.cassandra.CassandraConstants.FIND_OBJS;
@@ -342,6 +343,18 @@ public void upsertObjs(@Nonnull Obj[] objs) throws ObjTooLargeException {
     persistObjs(objs, referenced, true);
   }
 
+  @Override
+  public boolean deleteWithReferenced(@Nonnull Obj obj) {
+    BoundStatement stmt =
+        backend.buildStatement(
+            DELETE_OBJ_REFERENCED,
+            false,
+            config.repositoryId(),
+            serializeObjId(obj.id()),
+            obj.referenced());
+    return backend.executeCas(stmt);
+  }
+
   @Override
   public boolean deleteConditional(@Nonnull UpdateableObj obj) {
     BoundStatement stmt =

versioned/storage/cassandra2/src/main/java/org/projectnessie/versioned/storage/cassandra2/Cassandra2Constants.java

@@ -115,6 +115,17 @@ public final class Cassandra2Constants {
           + COL_OBJ_VERS
           + "=?";
 
+  static final String DELETE_OBJ_REFERENCED =
+      "DELETE FROM %s."
+          + TABLE_OBJS
+          + " WHERE "
+          + COL_REPO_ID
+          + "=? AND "
+          + COL_OBJ_ID
+          + "=? IF "
+          + COL_OBJ_REFERENCED
+          + "=?";
+
   static final String CREATE_TABLE_OBJS =
       "CREATE TABLE %s."
           + TABLE_OBJS
@@ -376,7 +387,10 @@ public final class Cassandra2Constants {
           + " AND "
           + COL_OBJ_ID
           + "=:"
-          + COL_OBJ_ID;
+          + COL_OBJ_ID
+          // IF EXISTS is necessary to prevent writing just the referenced timestamp after an object
+          // has been deleted.
+          + " IF EXISTS";
 
   private Cassandra2Constants() {}
 }

versioned/storage/cassandra2/src/main/java/org/projectnessie/versioned/storage/cassandra2/Cassandra2Persist.java

@@ -29,6 +29,7 @@
 import static org.projectnessie.versioned.storage.cassandra2.Cassandra2Constants.COL_REPO_ID;
 import static org.projectnessie.versioned.storage.cassandra2.Cassandra2Constants.DELETE_OBJ;
 import static org.projectnessie.versioned.storage.cassandra2.Cassandra2Constants.DELETE_OBJ_CONDITIONAL;
+import static org.projectnessie.versioned.storage.cassandra2.Cassandra2Constants.DELETE_OBJ_REFERENCED;
 import static org.projectnessie.versioned.storage.cassandra2.Cassandra2Constants.EXPECTED_SUFFIX;
 import static org.projectnessie.versioned.storage.cassandra2.Cassandra2Constants.FETCH_OBJ_TYPE;
 import static org.projectnessie.versioned.storage.cassandra2.Cassandra2Constants.FIND_OBJS;
@@ -342,6 +343,18 @@ public void upsertObjs(@Nonnull Obj[] objs) throws ObjTooLargeException {
     persistObjs(objs, referenced, true);
   }
 
+  @Override
+  public boolean deleteWithReferenced(@Nonnull Obj obj) {
+    BoundStatement stmt =
+        backend.buildStatement(
+            DELETE_OBJ_REFERENCED,
+            false,
+            config.repositoryId(),
+            serializeObjId(obj.id()),
+            obj.referenced());
+    return backend.executeCas(stmt);
+  }
+
   @Override
   public boolean deleteConditional(@Nonnull UpdateableObj obj) {
     BoundStatement stmt =

versioned/storage/common-tests/src/main/java/org/projectnessie/versioned/storage/commontests/AbstractBasePersistTests.java

@@ -821,6 +821,27 @@ public void referencedLifecycleForUpdateable() throws Exception {
     soft.assertThat(updated.referenced()).isGreaterThan(upserted.referenced());
   }
 
+  @ParameterizedTest
+  @MethodSource("allObjectTypeSamples")
+  public void referencedDelete(Obj obj) throws Exception {
+    assumeThat(persist.isCaching())
+        .describedAs("'referenced' not tested against a caching Persist")
+        .isFalse();
+
+    soft.assertThat(persist.storeObj(obj)).isTrue();
+    Obj stored = persist.fetchObj(obj.id());
+    Thread.sleep(0, 1000);
+    persist.storeObj(stored);
+    Obj stored2 = persist.fetchObj(obj.id());
+    soft.assertThat(stored.referenced()).isNotEqualTo(stored2.referenced());
+
+    soft.assertThat(persist.deleteWithReferenced(stored)).isFalse();
+    soft.assertThatCode(() -> persist.fetchObj(obj.id())).doesNotThrowAnyException();
+    soft.assertThat(persist.deleteWithReferenced(stored2)).isTrue();
+    soft.assertThatExceptionOfType(ObjNotFoundException.class)
+        .isThrownBy(() -> persist.fetchObj(obj.id()));
+  }
+
   @ParameterizedTest
   @MethodSource("allObjectTypeSamples")
   public void singleObjectCreateDelete(Obj obj) throws Exception {

versioned/storage/common/src/main/java/org/projectnessie/versioned/storage/common/persist/ObservingPersist.java

@@ -271,6 +271,14 @@ public void upsertObjs(@Nonnull Obj[] objs) throws ObjTooLargeException {
     delegate.upsertObjs(objs);
   }
 
+  @WithSpan
+  @Override
+  @Counted(PREFIX)
+  @Timed(value = PREFIX, histogram = true)
+  public boolean deleteWithReferenced(@Nonnull Obj obj) {
+    return delegate.deleteWithReferenced(obj);
+  }
+
   @WithSpan
   @Override
   @Counted(PREFIX)

versioned/storage/common/src/main/java/org/projectnessie/versioned/storage/common/persist/Persist.java

@@ -378,6 +378,14 @@ boolean storeObj(@Nonnull Obj obj, boolean ignoreSoftSizeRestrictions)
    */
   void deleteObjs(@Nonnull ObjId[] ids);
 
+  /**
+   * Deletes the given object, if its {@link Obj#referenced()} value is equal to the persisted
+   * object's value. Since a caching {@link Persist} does not guarantee that the {@link
+   * Obj#referenced()} value is up-to-date, callers must ensure that they read the uncached object
+   * state, for example via a {@link #scanAllObjects(Set)}.
+   */
+  boolean deleteWithReferenced(@Nonnull Obj obj);
+
   /**
    * Deletes the object, if the current state in the database is equal to the given state, comparing
    * the {@link UpdateableObj#versionToken()}.

versioned/storage/dynamodb/src/main/java/org/projectnessie/versioned/storage/dynamodb/DynamoDBPersist.java

@@ -545,6 +545,28 @@ public void upsertObjs(@Nonnull Obj[] objs) throws ObjTooLargeException {
     }
   }
 
+  @Override
+  public boolean deleteWithReferenced(@Nonnull Obj obj) {
+    ObjId id = obj.id();
+
+    Map<String, ExpectedAttributeValue> expectedValues =
+        Map.of(
+            COL_OBJ_REFERENCED,
+            ExpectedAttributeValue.builder().value(fromS(Long.toString(obj.referenced()))).build());
+
+    try {
+      backend
+          .client()
+          .deleteItem(
+              b -> b.tableName(backend.tableObjs).key(objKeyMap(id)).expected(expectedValues));
+      return true;
+    } catch (ConditionalCheckFailedException checkFailedException) {
+      return false;
+    } catch (RuntimeException e) {
+      throw unhandledException(e);
+    }
+  }
+
   @Override
   public boolean deleteConditional(@Nonnull UpdateableObj obj) {
     ObjId id = obj.id();

versioned/storage/dynamodb2/src/main/java/org/projectnessie/versioned/storage/dynamodb2/DynamoDB2Persist.java

@@ -546,6 +546,28 @@ public void upsertObjs(@Nonnull Obj[] objs) throws ObjTooLargeException {
     }
   }
 
+  @Override
+  public boolean deleteWithReferenced(@Nonnull Obj obj) {
+    ObjId id = obj.id();
+
+    Map<String, ExpectedAttributeValue> expectedValues =
+        Map.of(
+            COL_OBJ_REFERENCED,
+            ExpectedAttributeValue.builder().value(fromS(Long.toString(obj.referenced()))).build());
+
+    try {
+      backend
+          .client()
+          .deleteItem(
+              b -> b.tableName(backend.tableObjs).key(objKeyMap(id)).expected(expectedValues));
+      return true;
+    } catch (ConditionalCheckFailedException checkFailedException) {
+      return false;
+    } catch (RuntimeException e) {
+      throw unhandledException(e);
+    }
+  }
+
   @Override
   public boolean deleteConditional(@Nonnull UpdateableObj obj) {
     ObjId id = obj.id();

versioned/storage/inmemory/src/main/java/org/projectnessie/versioned/storage/inmemory/InmemoryPersist.java

@@ -287,6 +287,24 @@ public void upsertObjs(@Nonnull Obj[] objs) throws ObjTooLargeException {
     }
   }
 
+  @Override
+  public boolean deleteWithReferenced(@Nonnull Obj obj) {
+    AtomicBoolean result = new AtomicBoolean();
+    inmemory.objects.compute(
+        compositeKey(obj.id()),
+        (k, v) -> {
+          if (v == null) {
+            // not present
+            return null;
+          } else if (v.referenced() != obj.referenced()) {
+            return v;
+          }
+          result.set(true);
+          return null;
+        });
+    return result.get();
+  }
+
   @Override
   public boolean deleteConditional(@Nonnull UpdateableObj obj) {
     return updateDeleteConditional(obj, null);

versioned/storage/jdbc/src/main/java/org/projectnessie/versioned/storage/jdbc/AbstractJdbcPersist.java

@@ -31,6 +31,7 @@
 import static org.projectnessie.versioned.storage.jdbc.SqlConstants.COL_REPO_ID;
 import static org.projectnessie.versioned.storage.jdbc.SqlConstants.DELETE_OBJ;
 import static org.projectnessie.versioned.storage.jdbc.SqlConstants.DELETE_OBJ_CONDITIONAL;
+import static org.projectnessie.versioned.storage.jdbc.SqlConstants.DELETE_OBJ_REFERENCED;
 import static org.projectnessie.versioned.storage.jdbc.SqlConstants.FETCH_OBJ_TYPE;
 import static org.projectnessie.versioned.storage.jdbc.SqlConstants.FIND_OBJS;
 import static org.projectnessie.versioned.storage.jdbc.SqlConstants.FIND_OBJS_TYPED;
@@ -442,6 +443,17 @@ protected final Void updateObjs(@Nonnull Connection conn, @Nonnull Obj[] objs)
     return null;
   }
 
+  protected final boolean deleteWithReferenced(@Nonnull Connection conn, @Nonnull Obj obj) {
+    try (PreparedStatement ps = conn.prepareStatement(DELETE_OBJ_REFERENCED)) {
+      ps.setString(1, config.repositoryId());
+      serializeObjId(ps, 2, obj.id(), databaseSpecific);
+      ps.setLong(3, obj.referenced());
+      return ps.executeUpdate() == 1;
+    } catch (SQLException e) {
+      throw unhandledSQLException(e);
+    }
+  }
+
   protected final boolean deleteConditional(@Nonnull Connection conn, @Nonnull UpdateableObj obj) {
     try (PreparedStatement ps = conn.prepareStatement(DELETE_OBJ_CONDITIONAL)) {
       ps.setString(1, config.repositoryId());

versioned/storage/jdbc/src/main/java/org/projectnessie/versioned/storage/jdbc/JdbcPersist.java

@@ -210,6 +210,11 @@ public void upsertObjs(@Nonnull Obj[] objs) throws ObjTooLargeException {
     withConnectionException(false, conn -> super.updateObjs(conn, objs));
   }
 
+  @Override
+  public boolean deleteWithReferenced(@Nonnull Obj obj) {
+    return withConnectionException(false, conn -> super.deleteWithReferenced(conn, obj));
+  }
+
   @Override
   public boolean deleteConditional(@Nonnull UpdateableObj obj) {
     return withConnectionException(false, conn -> super.deleteConditional(conn, obj));

versioned/storage/jdbc/src/main/java/org/projectnessie/versioned/storage/jdbc/SqlConstants.java

@@ -63,6 +63,16 @@ final class SqlConstants {
           + "=? AND "
           + COL_OBJ_ID
           + "=?";
+  static final String DELETE_OBJ_REFERENCED =
+      "DELETE FROM "
+          + TABLE_OBJS
+          + " WHERE "
+          + COL_REPO_ID
+          + "=? AND "
+          + COL_OBJ_ID
+          + "=? AND "
+          + COL_OBJ_REFERENCED
+          + "=?";
 
   static final String COL_REFS_NAME = "ref_name";
   static final String COL_REFS_POINTER = "pointer";