insert almost works

Author: normen662

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

@@ -23,6 +23,8 @@
 import com.apple.foundationdb.annotation.API;
 import com.apple.foundationdb.util.LoggableException;
 import com.google.common.base.Suppliers;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 import javax.annotation.Nonnull;
@@ -33,6 +35,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.NoSuchElementException;
+import java.util.Objects;
 import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -1055,7 +1058,7 @@ public static CompletableFuture<Void> swallowException(@Nonnull CompletableFutur
         return result;
     }
 
-    public static CompletableFuture<Void> forLoop(int startI, @Nonnull final IntPredicate conditionPredicate,
+    public static CompletableFuture<Void> forLoop(final int startI, @Nonnull final IntPredicate conditionPredicate,
                                                   @Nonnull final IntUnaryOperator stepFunction,
                                                   @Nonnull final IntFunction<CompletableFuture<Void>> body,
                                                   @Nonnull final Executor executor) {
@@ -1073,6 +1076,51 @@ public static CompletableFuture<Void> forLoop(int startI, @Nonnull final IntPred
         }, executor);
     }
 
+    @SuppressWarnings("unchecked")
+    public static <T, U> CompletableFuture<List<U>> forEach(@Nonnull final Iterable<T> items,
+                                                            @Nonnull final Function<T, CompletableFuture<U>> body,
+                                                            final int parallelism,
+                                                            @Nonnull final Executor executor) {
+        // this deque is only modified by once upon creation
+        final ArrayDeque<T> toBeProcessed = new ArrayDeque<>();
+        for (final T item : items) {
+            toBeProcessed.addLast(item);
+        }
+
+        final List<CompletableFuture<Void>> working = Lists.newArrayList();
+        final AtomicInteger indexAtomic = new AtomicInteger(0);
+        final Object[] resultArray = new Object[toBeProcessed.size()];
+
+        return AsyncUtil.whileTrue(() -> {
+            working.removeIf(CompletableFuture::isDone);
+
+            while (working.size() <= parallelism) {
+                final T currentItem = toBeProcessed.pollFirst();
+                if (currentItem == null) {
+                    break;
+                }
+
+                final int index = indexAtomic.getAndIncrement();
+                working.add(body.apply(currentItem)
+                        .thenAccept(resultNode -> {
+                            Objects.requireNonNull(resultNode);
+                            resultArray[index] = resultNode;
+                        }));
+            }
+
+            if (working.isEmpty()) {
+                return AsyncUtil.READY_FALSE;
+            }
+            return AsyncUtil.whenAny(working).thenApply(ignored -> true);
+        }, executor).thenApply(ignored -> {
+            final ImmutableList.Builder<U> resultBuilder = ImmutableList.builder();
+            for (final Object o : resultArray) {
+                resultBuilder.add((U)o);
+            }
+            return resultBuilder.build();
+        });
+    }
+
     /**
      * 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/AbstractNode.java

@@ -28,17 +28,17 @@
 
 /**
  * TODO.
- * @param <R> node type class.
+ * @param <N> node type class.
  */
-abstract class AbstractNode<R extends NodeReference> implements Node<R> {
+abstract class AbstractNode<N extends NodeReference> implements Node<N> {
     @Nonnull
     private final Tuple primaryKey;
 
     @Nonnull
-    private final List<R> neighbors;
+    private final List<N> neighbors;
 
     protected AbstractNode(@Nonnull final Tuple primaryKey,
-                           @Nonnull final List<R> neighbors) {
+                           @Nonnull final List<N> neighbors) {
         this.primaryKey = primaryKey;
         this.neighbors = ImmutableList.copyOf(neighbors);
     }
@@ -51,13 +51,13 @@ public Tuple getPrimaryKey() {
 
     @Nonnull
     @Override
-    public List<R> getNeighbors() {
+    public List<N> getNeighbors() {
         return neighbors;
     }
 
     @Nonnull
     @Override
-    public R getNeighbor(final int index) {
+    public N getNeighbor(final int index) {
         return neighbors.get(index);
     }
 }

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

@@ -0,0 +1,53 @@
+/*
+ * InliningNode.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2023 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed 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 com.apple.foundationdb.async.hnsw;
+
+import com.apple.foundationdb.Transaction;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.List;
+
+/**
+ * TODO.
+ */
+class BaseNeighborsChangeSet<N extends NodeReference> implements NeighborsChangeSet<N> {
+    @Nonnull
+    private final NodeReferenceAndNode<N> baseNode;
+
+    public BaseNeighborsChangeSet(@Nonnull final NodeReferenceAndNode<N> baseNode) {
+        this.baseNode = baseNode;
+    }
+
+    @Nullable
+    public BaseNeighborsChangeSet<N> getParent() {
+        return null;
+    }
+
+    @Nonnull
+    public List<N> merge() {
+        return baseNode.getNode().getNeighbors();
+    }
+
+    @Override
+    public void writeDelta(@Nonnull final Transaction transaction) {
+    }
+}

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

@@ -22,7 +22,6 @@
 
 import com.apple.foundationdb.tuple.Tuple;
 import com.christianheina.langx.half4j.Half;
-import com.google.common.base.Verify;
 import com.google.common.collect.Lists;
 
 import javax.annotation.Nonnull;
@@ -39,8 +38,8 @@ public class CompactNode extends AbstractNode<NodeReference> {
         @SuppressWarnings("unchecked")
         @Nonnull
         @Override
-        public Node<NodeReference> create(@Nonnull final NodeKind nodeKind, @Nonnull final Tuple primaryKey, @Nullable final Vector<Half> vector, @Nonnull final List<? extends NodeReference> neighbors) {
-            Verify.verify(nodeKind == NodeKind.COMPACT);
+        public Node<NodeReference> create(@Nonnull final Tuple primaryKey, @Nullable final Vector<Half> vector,
+                                          @Nonnull final List<? extends NodeReference> neighbors) {
             return new CompactNode(primaryKey, Objects.requireNonNull(vector), (List<NodeReference>)neighbors);
         }
 
@@ -60,6 +59,12 @@ public CompactNode(@Nonnull final Tuple primaryKey, @Nonnull final Vector<Half>
         this.vector = vector;
     }
 
+    @Nonnull
+    @Override
+    public NodeReference getSelfReference(@Nullable final Vector<Half> vector) {
+        return new NodeReference(getPrimaryKey());
+    }
+
     @Nonnull
     @Override
     public NodeKind getKind() {

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

@@ -0,0 +1,63 @@
+/*
+ * InliningNode.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2023 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed 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 com.apple.foundationdb.async.hnsw;
+
+import com.apple.foundationdb.Transaction;
+import com.apple.foundationdb.tuple.Tuple;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
+
+import javax.annotation.Nonnull;
+import java.util.Collection;
+import java.util.Set;
+
+/**
+ * TODO.
+ */
+class DeleteNeighborsChangeSet<N extends NodeReference> implements NeighborsChangeSet<N> {
+    @Nonnull
+    private final NeighborsChangeSet<N> parent;
+
+    @Nonnull
+    private final Set<Tuple> deletedNeighborsTuples;
+
+    public DeleteNeighborsChangeSet(@Nonnull final NeighborsChangeSet<N> parent,
+                                    @Nonnull final Collection<Tuple> deletedNeighborsTuples) {
+        this.parent = parent;
+        this.deletedNeighborsTuples = ImmutableSet.copyOf(deletedNeighborsTuples);
+    }
+
+    @Nonnull
+    public NeighborsChangeSet<N> getParent() {
+        return parent;
+    }
+
+    @Nonnull
+    public Iterable<N> merge() {
+        return Iterables.filter(getParent().merge(),
+                current -> !deletedNeighborsTuples.contains(current.getPrimaryKey()));
+    }
+
+    @Override
+    public void writeDelta(@Nonnull final Transaction transaction) {
+        throw new UnsupportedOperationException("not implemented yet");
+    }
+}