started on the writing path

Author: normen662

fdb-extensions/src/main/java/com/apple/foundationdb/async/MoreAsyncUtil.java

@@ -42,9 +42,13 @@
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.Function;
+import java.util.function.IntFunction;
+import java.util.function.IntPredicate;
+import java.util.function.IntUnaryOperator;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 
@@ -1051,6 +1055,24 @@ public static CompletableFuture<Void> swallowException(@Nonnull CompletableFutur
         return result;
     }
 
+    public static CompletableFuture<Void> forLoop(int startI, @Nonnull final IntPredicate conditionPredicate,
+                                                  @Nonnull final IntUnaryOperator stepFunction,
+                                                  @Nonnull final IntFunction<CompletableFuture<Void>> body,
+                                                  @Nonnull final Executor executor) {
+        final AtomicInteger loopVariableAtomic = new AtomicInteger(startI);
+        return AsyncUtil.whileTrue(() -> {
+            final int loopVariable = loopVariableAtomic.get();
+            if (!conditionPredicate.test(loopVariable)) {
+                return AsyncUtil.READY_FALSE;
+            }
+            return body.apply(loopVariable)
+                    .thenApply(ignored -> {
+                        loopVariableAtomic.set(stepFunction.applyAsInt(loopVariable));
+                        return true;
+                    });
+        }, executor);
+    }
+
     /**
      * A {@code Boolean} function that is always true.
      * @param <T> the type of the (ignored) argument to the function

fdb-extensions/src/main/java/com/apple/foundationdb/async/hnsw/AbstractStorageAdapter.java

@@ -21,17 +21,12 @@
 package com.apple.foundationdb.async.hnsw;
 
 import com.apple.foundationdb.ReadTransaction;
-import com.apple.foundationdb.Transaction;
 import com.apple.foundationdb.subspace.Subspace;
 import com.apple.foundationdb.tuple.Tuple;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
-import java.math.BigInteger;
-import java.util.List;
-import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
-import java.util.function.Function;
 
 /**
  * Implementations and attributes common to all concrete implementations of {@link StorageAdapter}.
@@ -53,8 +48,6 @@ abstract class AbstractStorageAdapter implements StorageAdapter {
     private final Subspace dataSubspace;
 
     protected AbstractStorageAdapter(@Nonnull final HNSW.Config config, @Nonnull final Subspace subspace,
-                                     @Nonnull final Subspace nodeSlotIndexSubspace,
-                                     @Nonnull final Function<HNSW.Point, BigInteger> hilbertValueFunction,
                                      @Nonnull final OnWriteListener onWriteListener,
                                      @Nonnull final OnReadListener onReadListener) {
         this.config = config;
@@ -108,79 +101,16 @@ public OnReadListener getOnReadListener() {
         return onReadListener;
     }
 
-    @Override
-    public void writeNodes(@Nonnull final Transaction transaction, @Nonnull final List<? extends Node> nodes) {
-        for (final Node node : nodes) {
-            writeNode(transaction, node);
-        }
-    }
-
-    protected void writeNode(@Nonnull final Transaction transaction, @Nonnull final Node node) {
-        final Node.ChangeSet changeSet = node.getChangeSet();
-        if (changeSet == null) {
-            return;
-        }
-
-        changeSet.apply(transaction);
-        getOnWriteListener().onNodeWritten(node);
-    }
-
-    @Nonnull
-    public byte[] packWithSubspace(final byte[] key) {
-        return getSubspace().pack(key);
-    }
-
-    @Nonnull
-    public byte[] packWithSubspace(final Tuple tuple) {
-        return getSubspace().pack(tuple);
-    }
-
-    @Nonnull
-    @Override
-    public CompletableFuture<Node> scanNodeIndexAndFetchNode(@Nonnull final ReadTransaction transaction,
-                                                             final int level,
-                                                             @Nonnull final BigInteger hilbertValue,
-                                                             @Nonnull final Tuple key,
-                                                             final boolean isInsertUpdate) {
-        Objects.requireNonNull(nodeSlotIndexAdapter);
-        return nodeSlotIndexAdapter.scanIndexForNodeId(transaction, level, hilbertValue, key, isInsertUpdate)
-                .thenCompose(nodeId -> nodeId == null
-                                       ? CompletableFuture.completedFuture(null)
-                                       : fetchNode(transaction, nodeId));
-    }
-
-    @Override
-    public void insertIntoNodeIndexIfNecessary(@Nonnull final Transaction transaction, final int level,
-                                               @Nonnull final NodeSlot nodeSlot) {
-        if (!getConfig().isUseNodeSlotIndex() || !(nodeSlot instanceof ChildSlot)) {
-            return;
-        }
-
-        Objects.requireNonNull(nodeSlotIndexAdapter);
-        nodeSlotIndexAdapter.writeChildSlot(transaction, level, (ChildSlot)nodeSlot);
-    }
-
-    @Override
-    public void deleteFromNodeIndexIfNecessary(@Nonnull final Transaction transaction, final int level,
-                                               @Nonnull final NodeSlot nodeSlot) {
-        if (!getConfig().isUseNodeSlotIndex() || !(nodeSlot instanceof ChildSlot)) {
-            return;
-        }
-
-        Objects.requireNonNull(nodeSlotIndexAdapter);
-        nodeSlotIndexAdapter.clearChildSlot(transaction, level, (ChildSlot)nodeSlot);
-    }
-
     @Nonnull
     @Override
-    public <N extends NodeReference> CompletableFuture<Node<N>> fetchNode(@Nonnull final Node.NodeCreator<N> creator,
+    public <N extends NodeReference> CompletableFuture<Node<N>> fetchNode(@Nonnull final NodeFactory<N> nodeFactory,
                                                                           @Nonnull final ReadTransaction readTransaction,
                                                                           int layer, @Nonnull Tuple primaryKey) {
-        return fetchNodeInternal(creator, readTransaction, layer, primaryKey).thenApply(this::checkNode);
+        return fetchNodeInternal(nodeFactory, readTransaction, layer, primaryKey).thenApply(this::checkNode);
     }
 
     @Nonnull
-    protected abstract <N extends NodeReference> CompletableFuture<Node<N>> fetchNodeInternal(@Nonnull Node.NodeCreator<N> creator,
+    protected abstract <N extends NodeReference> CompletableFuture<Node<N>> fetchNodeInternal(@Nonnull NodeFactory<N> nodeFactory,
                                                                                               @Nonnull ReadTransaction readTransaction,
                                                                                               int layer, @Nonnull Tuple primaryKey);
 
@@ -197,16 +127,4 @@ protected abstract <N extends NodeReference> CompletableFuture<Node<N>> fetchNod
     private <N extends NodeReference> Node<N> checkNode(@Nullable final Node<N> node) {
         return node;
     }
-
-    @Nonnull
-    abstract <S extends NodeSlot, N extends AbstractNode<S, N>> AbstractChangeSet<S, N>
-            newInsertChangeSet(@Nonnull N node, int level, @Nonnull List<S> insertedSlots);
-
-    @Nonnull
-    abstract <S extends NodeSlot, N extends AbstractNode<S, N>> AbstractChangeSet<S, N>
-            newUpdateChangeSet(@Nonnull N node, int level, @Nonnull S originalSlot, @Nonnull S updatedSlot);
-
-    @Nonnull
-    abstract <S extends NodeSlot, N extends AbstractNode<S, N>> AbstractChangeSet<S, N>
-            newDeleteChangeSet(@Nonnull N node, int level, @Nonnull List<S> deletedSlots);
 }