perf: cut what edge ingest allocates by about a third

benchmarks/src/main/java/io/hstore/bench/HStoreStore.java

@@ -16,7 +16,6 @@
 import io.hstore.engine.txn.Durability;
 
 import java.nio.file.Path;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -120,10 +119,10 @@ public void ingestEdges(Dataset dataset, int batch) {
         }
     }
 
-    private List<Long> members(int[] indexes) {
-        List<Long> members = new ArrayList<>(indexes.length);
-        for (int index : indexes) {
-            members.add(nodes[index]);
+    private long[] members(int[] indexes) {
+        long[] members = new long[indexes.length];
+        for (int i = 0; i < indexes.length; i++) {
+            members[i] = nodes[indexes[i]];
         }
         return members;
     }
@@ -191,11 +190,7 @@ public void update(int[] probes, int from, int to, int round) {
     public void largeEdge(int count) {
         large = database.write(writer -> {
             long edge = writer.edge(EDGE);
-            List<Long> members = new ArrayList<>(count);
-            for (int i = 0; i < count; i++) {
-                members.add(nodes[i]);
-            }
-            writer.load(edge, members, MemberSpec.PLAIN);
+            writer.load(edge, Arrays.copyOf(nodes, count), MemberSpec.PLAIN);
             return edge;
         });
     }

database/src/main/java/io/hstore/db/Writer.java

@@ -21,6 +21,7 @@
 import io.hstore.db.value.TypeTag;
 import io.hstore.db.value.Value;
 import io.hstore.db.value.Values;
+import io.hstore.engine.Dictionary;
 import io.hstore.engine.HStoreException;
 import io.hstore.engine.catalog.AtomRecord;
 import io.hstore.engine.catalog.EngineSlots;
@@ -31,6 +32,7 @@
 import io.hstore.engine.txn.Transaction;
 
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -138,13 +140,11 @@ private TypeDef kindOf(String type, AtomKind kind) {
     }
 
     private void assign(long atom, TypeDef definition, Map<String, ?> properties) {
-        definition.properties().stream()
-                .filter(PropertyDef::required)
-                .filter(property -> !properties.containsKey(property.name()))
-                .findFirst()
-                .ifPresent(missing -> {
-                    throw HStoreException.invalid(definition.name() + " requires property '" + missing.name() + "'");
-                });
+        for (PropertyDef property : definition.properties()) {
+            if (property.required() && !properties.containsKey(property.name())) {
+                throw HStoreException.invalid(definition.name() + " requires property '" + property.name() + "'");
+            }
+        }
         properties.forEach((name, value) -> set(atom, name, Value.of(value)));
     }
 
@@ -178,7 +178,19 @@ public void removeAt(long edge, long index) {
     }
 
     public void load(long edge, List<Long> members, MemberSpec spec) {
-        txn.load(edge, members.stream().map(member -> incidence(edge, member, spec)).toList());
+        load(edge, members.stream().mapToLong(Long::longValue).toArray(), spec);
+    }
+
+    public void load(long edge, long[] members, MemberSpec spec) {
+        Incidence[] incidences = new Incidence[members.length];
+        if (members.length > 0) {
+            Incidence first = incidences[0] = incidence(edge, members[0], spec);
+            for (int i = 1; i < members.length; i++) {
+                requireVisible(members[i]);
+                incidences[i] = new Incidence(members[i], first.roleSet(), first.weight(), first.validFrom(), first.validTo(), first.dataRef(), first.qualifier());
+            }
+        }
+        txn.load(edge, Arrays.asList(incidences));
     }
 
     public void qualify(long edge, long member, long qualifier) {
@@ -188,20 +200,30 @@ public void qualify(long edge, long member, long qualifier) {
     }
 
     private Incidence incidence(long edge, long member, MemberSpec spec) {
+        requireMembership(edge, spec);
+        requireVisible(member);
+        int roleSet = spec.roles().isEmpty() ? Dictionary.NO_ROLES : database.engine().dictionary().roleSet(spec.roles());
+        long weight = spec.weight().isPresent() ? Weight.of(spec.weight().getAsDouble()) : Weight.ONE;
+        return new Incidence(member, roleSet, weight, spec.validity().from(), spec.validity().to(), spec.dataRef(), spec.qualifier());
+    }
+
+    private void requireMembership(long edge, MemberSpec spec) {
         principal().requireWrite();
         Atom container = require(edge);
         if (!container.isEdge()) {
             throw HStoreException.invalid("atom " + edge + " is not a hyperedge");
         }
-        spec.roles().stream().filter(role -> !container.type().allowsRole(role)).findFirst().ifPresent(role -> {
-            throw HStoreException.invalid(container.type().name() + " does not declare role '" + role + "'");
-        });
+        for (String role : spec.roles()) {
+            if (!container.type().allowsRole(role)) {
+                throw HStoreException.invalid(container.type().name() + " does not declare role '" + role + "'");
+            }
+        }
+    }
+
+    private void requireVisible(long member) {
         if (!visible(member)) {
             throw HStoreException.invalid("member atom " + member + " does not exist");
         }
-        int roleSet = database.engine().dictionary().roleSet(spec.roles());
-        return new Incidence(member, roleSet, spec.weight().stream().mapToLong(Weight::of).findFirst().orElse(Weight.ONE),
-                spec.validity().from(), spec.validity().to(), spec.dataRef(), spec.qualifier());
     }
 
     public void set(long atom, String name, Value value) {

database/src/test/java/io/hstore/db/DatabaseTest.java

@@ -136,6 +136,49 @@ void membersCarryRolesWeightsAndValidity() {
         });
     }
 
+    @Test
+    void bulkLoadedMembersMatchMembersAddedOneAtATime() {
+        MemberSpec spec = MemberSpec.roles("author").withWeight(0.5).withValidity(new Validity(10, 20));
+        long[] edges = database.write(writer -> {
+            long[] people = new long[5];
+            for (int i = 0; i < people.length; i++) {
+                people[i] = writer.node("Person", "p" + i, Map.of("name", "Person " + i));
+            }
+            long added = writer.edge("Coauthorship", Map.of("year", 2024));
+            for (long person : people) {
+                writer.add(added, person, spec);
+            }
+            long fromArray = writer.edge("Coauthorship", Map.of("year", 2024));
+            writer.load(fromArray, new long[]{people[3], people[0], people[4], people[1], people[2]}, spec);
+            long fromList = writer.edge("Coauthorship", Map.of("year", 2024));
+            writer.load(fromList, List.of(people[4], people[2], people[0], people[3], people[1]), spec);
+            long empty = writer.edge("Coauthorship", Map.of("year", 2024));
+            writer.load(empty, new long[0], spec);
+            return new long[]{added, fromArray, fromList, empty};
+        });
+        database.read(reader -> {
+            List<Member> expected = reader.members(edges[0]).toList();
+            assertEquals(5, expected.size());
+            assertEquals(expected, reader.members(edges[1]).toList());
+            assertEquals(expected, reader.members(edges[2]).toList());
+            assertEquals(0, reader.cardinality(edges[3]));
+            return null;
+        });
+        long node = database.read(reader -> reader.resolve("Person", "p0"));
+        assertThrows(HStoreException.InvalidSchema.class, () -> database.write(writer -> {
+            writer.load(edges[3], new long[]{node}, MemberSpec.roles("reviewer"));
+            return null;
+        }));
+        assertThrows(HStoreException.class, () -> database.write(writer -> {
+            writer.load(edges[3], new long[]{node, 999_999_999L}, spec);
+            return null;
+        }));
+        assertThrows(HStoreException.class, () -> database.write(writer -> {
+            writer.load(node, new long[]{node}, spec);
+            return null;
+        }));
+    }
+
     @Test
     void incidentEdgeVisitorsSeeTheSameEdgesAsTheIncidenceStream() {
         long[] people = database.write(writer -> {

docs/database/data-model.md

@@ -450,7 +450,7 @@ Optional.empty
 | `add(edge, member, spec)` | WRITER | upsert an incidence (roles validated against the type) |
 | `insertAt(edge, index, member, spec)` / `append(...)` | WRITER | ordered edges |
 | `remove(edge, member)` / `removeAt(edge, index)` | WRITER | membership removal |
-| `load(edge, members, spec)` | WRITER | replaces the whole membership with a tree built bottom-up from `members` (`Hyperedge.build`) |
+| `load(edge, members, spec)` | WRITER | replaces the whole membership with a tree built bottom-up from `members` (`Hyperedge.build`); `members` is a `List<Long>` or a `long[]`, and the edge and the roles are checked once per call |
 | `qualify(edge, member, qualifier)` | WRITER | attach an assertion id to an incidence |
 | `set(atom, name, value[, validity])` / `unset(atom, name)` | WRITER | per-key OCC on the property bag |
 | `document(atom, json)` | WRITER | stores `$document` |

docs/storage/change-feed.md

@@ -77,7 +77,7 @@ offset  size  field
 
 **Open.** `restore()` opens every `*.feed`, scans all frames from the first segment's start to the end of the last segment, and stops at the first frame that is short, has a negative or overlong length, fails its CRC, or is not newer than the frame before it. The CRC covers only the payload, so a frame also counts as damaged unless its payload is non-empty and starts with the same generation as its header (`FeedCodec` always writes the generation first); this keeps a damaged header, or a run of zeros left where an unsynced frame never reached the disk, from passing as a frame. `cutAt(position)` then deletes every segment that starts after the failure point and truncates the segment containing it, so the feed always ends at a clean frame.
 
-**Index.** Each `Segment` remembers `firstGeneration`, the generation of its first frame. An in-memory sparse index `generation -> global position` holds the first frame of **every segment** and every 64th frame thereafter (`INDEX_STRIDE = 64`). `startFor(g)` seeks to the floor entry (never before the first retained segment) and scans forward, so locating a generation costs at most 63 frame reads plus segment boundaries.
+**Index.** Each `Segment` remembers `firstGeneration`, the generation of its first frame. An in-memory sparse index `generation -> global position` holds the first frame of **every segment** and every 64th frame thereafter (`INDEX_STRIDE = 64`). `startFor(g)` seeks to the floor entry (never before the first retained segment) and scans forward, so locating a generation costs at most 63 frame reads plus segment boundaries. `replay(after)` reads only the 16-byte header of a frame whose generation is at most `after` and steps over its payload, so a subscriber that wakes after every commit doesn't read the payloads of the frames it has already consumed again.
 
 ```text
 feed/

engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java

@@ -10,8 +10,8 @@
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
-import java.lang.foreign.MemorySegment;
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.nio.channels.FileChannel;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -307,7 +307,7 @@ public void truncateAfter(long generation) {
     }
 
     public Stream<CommitEvent> replay(long afterGeneration) {
-        Iterator<Frame> frames = new FrameReader(startFor(afterGeneration), end);
+        Iterator<Frame> frames = new FrameReader(startFor(afterGeneration), end, afterGeneration);
         return StreamSupport.stream(Spliterators.spliteratorUnknownSize(frames, Spliterator.ORDERED | Spliterator.NONNULL), false)
                 .filter(frame -> frame.generation() > afterGeneration)
                 .map(frame -> codec.decode(frame.payload()));
@@ -351,13 +351,20 @@ private record Frame(long start, long next, long generation, byte[] payload) {
     }
 
     private final class FrameReader implements Iterator<Frame> {
+        private final ByteBuffer header = ByteBuffer.allocate(FRAME_HEADER).order(ByteOrder.LITTLE_ENDIAN);
         private final long limit;
+        private final long skipThrough;
         private long position;
         private Frame next;
 
         FrameReader(long from, long limit) {
+            this(from, limit, Long.MIN_VALUE);
+        }
+
+        FrameReader(long from, long limit, long skipThrough) {
             this.position = from;
             this.limit = limit;
+            this.skipThrough = skipThrough;
         }
 
         @Override
@@ -389,32 +396,37 @@ private Frame read() {
         }
 
         private Frame readLocked() {
-            Map.Entry<Long, Segment> entry = segments.floorEntry(position);
-            if (entry == null || !entry.getValue().file.isOpen()) {
-                return null;
-            }
-            FileHandle file = entry.getValue().file;
-            long local = position - entry.getKey();
             try {
-                ByteBuffer header = ByteBuffer.allocate(FRAME_HEADER);
-                if (!readFully(file, header, local)) {
-                    return null;
-                }
-                ByteCursor in = ByteCursor.over(MemorySegment.ofBuffer(header.flip()));
-                int length = in.getInt();
-                int crc = in.getInt();
-                long generation = in.getLong();
-                if (length < 0 || position + FRAME_HEADER + length > limit) {
-                    return null;
-                }
-                ByteBuffer body = ByteBuffer.allocate(length);
-                if (!readFully(file, body, local + FRAME_HEADER)) {
-                    return null;
+                while (position + FRAME_HEADER <= limit) {
+                    Map.Entry<Long, Segment> entry = segments.floorEntry(position);
+                    if (entry == null || !entry.getValue().file.isOpen()) {
+                        return null;
+                    }
+                    FileHandle file = entry.getValue().file;
+                    long local = position - entry.getKey();
+                    if (!readFully(file, header.clear(), local)) {
+                        return null;
+                    }
+                    int length = header.getInt(0);
+                    int crc = header.getInt(Integer.BYTES);
+                    long generation = header.getLong(2 * Integer.BYTES);
+                    if (length < 0 || position + FRAME_HEADER + length > limit) {
+                        return null;
+                    }
+                    if (generation <= skipThrough) {
+                        position += FRAME_HEADER + length;
+                        continue;
+                    }
+                    ByteBuffer body = ByteBuffer.allocate(length);
+                    if (!readFully(file, body, local + FRAME_HEADER)) {
+                        return null;
+                    }
+                    byte[] payload = body.array();
+                    return Checksums.crc32c(payload) == crc && carries(payload, generation)
+                            ? new Frame(position, position + FRAME_HEADER + length, generation, payload)
+                            : null;
                 }
-                byte[] payload = body.array();
-                return Checksums.crc32c(payload) == crc && carries(payload, generation)
-                        ? new Frame(position, position + FRAME_HEADER + length, generation, payload)
-                        : null;
+                return null;
             } catch (IOException e) {
                 throw new UncheckedIOException(e);
             }

engine/src/main/java/io/hstore/engine/topology/IncidenceCodec.java

@@ -79,8 +79,11 @@ public int maxSize(Incidence incidence) {
     public void encode(ByteCursor out, long[] keys, List<Incidence> values) {
         int flags = 0;
         for (Column column : COLUMNS) {
-            if (values.stream().anyMatch(column.present)) {
-                flags |= 1 << column.ordinal();
+            for (int i = 0; i < values.size(); i++) {
+                if (column.present.test(values.get(i))) {
+                    flags |= 1 << column.ordinal();
+                    break;
+                }
             }
         }
         out.putByte(flags);

engine/src/main/java/io/hstore/engine/tree/BulkBuilder.java

@@ -14,15 +14,16 @@
 public final class BulkBuilder<V> {
 
     private static final int SPILL_BATCH = 32;
+    private static final int INITIAL_CAPACITY = 8;
 
     private final TreeSchema<V> schema;
     private final NodeSource source;
     private final WriteScope scope;
     private final Layout layout;
     private final List<Ref> leaves = new ArrayList<>();
     private BinaryOperator<V> merge = (_, latest) -> latest;
-    private long[] keys = new long[64];
-    private Object[] values = new Object[64];
+    private long[] keys = new long[0];
+    private Object[] values = new Object[0];
     private int size;
     private int bytes;
     private int unspilled;
@@ -58,8 +59,9 @@ public BulkBuilder<V> add(long key, V value) {
         }
         entryBytes += size == 0 ? ByteCursor.signedVarLongSize(key) : ByteCursor.varLongSize(key - keys[size - 1]);
         if (size == keys.length) {
-            keys = Arrays.copyOf(keys, size * 2);
-            values = Arrays.copyOf(values, size * 2);
+            int capacity = Math.max(INITIAL_CAPACITY, size * 2);
+            keys = Arrays.copyOf(keys, capacity);
+            values = Arrays.copyOf(values, capacity);
         }
         keys[size] = key;
         values[size] = value;