Skip to content

Commit 7d4b396

Browse files
DarthMaxDarthMax
committed
Revert paging of ChunkedAdjacencyLists
1 parent fd9ed1e commit 7d4b396

File tree

11 files changed

+393
-553
lines changed

11 files changed

+393
-553
lines changed

‎core/src/main/java/org/neo4j/gds/api/compress/AdjacencyCompressor.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ public interface AdjacencyCompressor extends AutoCloseable {
4949
* @param degree The number of targets stored in `targets`.
5050
* @return the degree of the compressed adjacency list
5151
*/
52-
int compress(long nodeId, long[] targets, long[][][] properties, int degree);
52+
int compress(long nodeId, long[] targets, long[][] properties, int degree);
5353

5454
/**
5555
* Closing this compressor will release some internal data structures, making them eligible for garbage collection.

‎core/src/main/java/org/neo4j/gds/core/compression/common/AdjacencyCompression.java‎

Lines changed: 17 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020
package org.neo4j.gds.core.compression.common;
2121

2222
import com.carrotsearch.hppc.sorting.IndirectSort;
23-
import org.apache.commons.lang3.mutable.MutableInt;
2423
import org.neo4j.gds.api.compress.AdjacencyCompressor;
2524
import org.neo4j.gds.api.compress.LongArrayBuffer;
2625
import org.neo4j.gds.core.Aggregation;
@@ -39,12 +38,13 @@ public final class AdjacencyCompression {
3938

4039
/**
4140
* Decompress the given {@code array} into the given {@code into}.
42-
* After this, {@link org.neo4j.gds.api.compress.LongArrayBuffer#length} will reflect the number of decompressed values
41+
* After this, {@link org.neo4j.gds.api.compress.LongArrayBuffer#length} will reflect the number of decompressed
42+
* values
4343
* that are in the {@link org.neo4j.gds.api.compress.LongArrayBuffer#buffer}.
4444
*/
4545
public static void zigZagUncompressFrom(
4646
LongArrayBuffer into,
47-
byte[][] targets,
47+
byte[] targets,
4848
int compressedValues,
4949
int limit,
5050
AdjacencyCompressor.ValueMapper mapper
@@ -56,7 +56,7 @@ public static void zigZagUncompressFrom(
5656

5757
public static void zigZagUncompressFrom(
5858
long[] into,
59-
byte[][] targets,
59+
byte[] targets,
6060
int compressedValues,
6161
int limit,
6262
AdjacencyCompressor.ValueMapper mapper
@@ -77,7 +77,7 @@ public static int applyDeltaEncoding(long[] data, int length, Aggregation aggreg
7777
// TODO: requires lots of additional memory ... inline indirect sort to make reuse of - to be created - buffers
7878
public static int applyDeltaEncoding(
7979
LongArrayBuffer data,
80-
long[][][] weights,
80+
long[][] weights,
8181
long[][] sortedWeights,
8282
Aggregation[] aggregations,
8383
boolean noAggregation
@@ -96,7 +96,7 @@ public static int applyDeltaEncoding(
9696
public static int applyDeltaEncoding(
9797
long[] data,
9898
int length,
99-
long[][][] unsortedWeights,
99+
long[][] unsortedWeights,
100100
long[][] sortedWeights,
101101
Aggregation[] aggregations,
102102
boolean noAggregation
@@ -233,21 +233,23 @@ public static void prefixSumDeltaEncodedValues(long[] values, int length) {
233233
* Applies delta encoding to the given {@code values}.
234234
* Weights are not encoded.
235235
*
236-
* @param order Ordered indices into {@code values} and {@code weights} for consuming these in ascending value order.
236+
* @param order Ordered indices into {@code values} and {@code weights} for consuming these in ascending
237+
* value order.
237238
* @param values Relationships represented by target node ID.
238239
* @param outValues Sorted, delta-encoded and optionally aggregated relationships.
239240
* @param unsortedWeights Relationship properties by key, ordered by {@code order}.
240-
* @param outWeights Sorted and optionally aggregated relationship properties.
241+
* @param sortedWeights Sorted and optionally aggregated relationship properties.
241242
* @param length Number of relationships (degree of source node) to process.
242-
* @param aggregations Aggregations to apply to parallel edges. One per relationship property key in {@code weights}.
243+
* @param aggregations Aggregations to apply to parallel edges. One per relationship property key in
244+
* {@code weights}.
243245
* @param noAggregation Is true iff all aggregations are NONE.
244246
*/
245247
private static int applyDelta(
246248
int[] order,
247249
long[] values,
248250
long[] outValues,
249-
long[][][] unsortedWeights,
250-
long[][] outWeights,
251+
long[][] unsortedWeights,
252+
long[][] sortedWeights,
251253
int length,
252254
Aggregation[] aggregations,
253255
boolean noAggregation
@@ -256,41 +258,29 @@ private static int applyDelta(
256258
long value = values[firstSortIdx];
257259
long delta;
258260

259-
int[] chunkLengths = new int[unsortedWeights[0].length];
260-
int totalChunkLength = 0;
261-
for (int i = 0; i < unsortedWeights[0].length; i++) {
262-
totalChunkLength += unsortedWeights[0][i].length;
263-
chunkLengths[i] = totalChunkLength;
264-
}
265-
266-
var pageIndex = new MutableInt();
267-
var indexInPage = new MutableInt();
268-
findPosition(chunkLengths, firstSortIdx, pageIndex, indexInPage);
269-
270261
outValues[0] = values[firstSortIdx];
271262
for (int i = 0; i < unsortedWeights.length; i++) {
272-
outWeights[i][0] = unsortedWeights[i][pageIndex.intValue()][indexInPage.intValue()];
263+
sortedWeights[i][0] = unsortedWeights[i][firstSortIdx];
273264
}
274265

275266
int in = 1, out = 1;
276267
for (; in < length; ++in) {
277268
final int sortIdx = order[in];
278-
findPosition(chunkLengths, sortIdx, pageIndex, indexInPage);
279269
delta = values[sortIdx] - value;
280270
value = values[sortIdx];
281271

282272
if (delta > 0L || noAggregation) {
283273
for (int i = 0; i < unsortedWeights.length; i++) {
284-
outWeights[i][out] = unsortedWeights[i][pageIndex.intValue()][indexInPage.intValue()];
274+
sortedWeights[i][out] = unsortedWeights[i][sortIdx];
285275
}
286276
outValues[out++] = delta;
287277
} else {
288278
for (int i = 0; i < unsortedWeights.length; i++) {
289279
Aggregation aggregation = aggregations[i];
290280
int existingIdx = out - 1;
291-
long[] outWeight = outWeights[i];
281+
long[] outWeight = sortedWeights[i];
292282
double existingWeight = Double.longBitsToDouble(outWeight[existingIdx]);
293-
double newWeight = Double.longBitsToDouble(unsortedWeights[i][pageIndex.intValue()][indexInPage.intValue()]);
283+
double newWeight = Double.longBitsToDouble(unsortedWeights[i][sortIdx]);
294284
newWeight = aggregation.merge(existingWeight, newWeight);
295285
outWeight[existingIdx] = Double.doubleToLongBits(newWeight);
296286
}
@@ -299,18 +289,6 @@ private static int applyDelta(
299289
return out;
300290
}
301291

302-
public static void findPosition(int[] chunkLengths, int position, MutableInt pageIndex, MutableInt indexInPage) {
303-
int chunkPosition = Arrays.binarySearch(chunkLengths, position);
304-
if (chunkPosition >= 0) {
305-
pageIndex.setValue(chunkPosition + 1);
306-
indexInPage.setValue(0);
307-
} else {
308-
int index = -chunkPosition - 1;
309-
pageIndex.setValue(index);
310-
indexInPage.setValue(position - (index == 0 ? 0 : chunkLengths[index - 1]));
311-
}
312-
}
313-
314292
private AdjacencyCompression() {
315293
}
316294
}

‎core/src/main/java/org/neo4j/gds/core/compression/common/ZigZagLongDecoding.java‎

Lines changed: 13 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -32,16 +32,24 @@ public long map(long value) {
3232
}
3333
}
3434

35-
public static int zigZagUncompress(byte[] chunk, int numberOfBytes, long[] out) {
35+
public static int zigZagUncompress(byte[] compressedData, int length, long[] uncompressedData) {
36+
return zigZagUncompress(compressedData, length, uncompressedData, Identity.INSTANCE);
37+
}
38+
39+
static int zigZagUncompress(
40+
byte[] compressedData,
41+
int length,
42+
long[] uncompressedData,
43+
AdjacencyCompressor.ValueMapper mapper
44+
) {
3645
long input, startValue = 0L, value = 0L;
3746
int into = 0, shift = 0, offset = 0;
38-
39-
while (offset < numberOfBytes) {
40-
input = chunk[offset++];
47+
while (offset < length) {
48+
input = compressedData[offset++];
4149
value += (input & 127L) << shift;
4250
if ((input & 128L) == 128L) {
4351
startValue += ((value >>> 1L) ^ -(value & 1L));
44-
out[into++] = startValue;
52+
uncompressedData[into++] = mapper.map(startValue);
4553
value = 0L;
4654
shift = 0;
4755
} else {
@@ -51,40 +59,6 @@ public static int zigZagUncompress(byte[] chunk, int numberOfBytes, long[] out)
5159
return into;
5260
}
5361

54-
public static int zigZagUncompress(byte[][] chunks, int numberOfBytes, long[] out) {
55-
return zigZagUncompress(chunks, numberOfBytes, out, Identity.INSTANCE);
56-
}
57-
58-
public static int zigZagUncompress(
59-
byte[][] chunks,
60-
int numberOfBytes,
61-
long[] out,
62-
AdjacencyCompressor.ValueMapper mapper
63-
) {
64-
int currentChunk = 0;
65-
long input, startValue = 0L, value = 0L;
66-
int into = 0, shift = 0;
67-
68-
while (numberOfBytes > 0) {
69-
var chunk = chunks[currentChunk++];
70-
var bytesToConsumeForChunk = Math.min(numberOfBytes, chunk.length);
71-
for (int offset = 0; offset < bytesToConsumeForChunk; offset++) {
72-
input = chunk[offset];
73-
value += (input & 127L) << shift;
74-
if ((input & 128L) == 128L) {
75-
startValue += ((value >>> 1L) ^ -(value & 1L));
76-
out[into++] = mapper.map(startValue);
77-
value = 0L;
78-
shift = 0;
79-
} else {
80-
shift += 7;
81-
}
82-
}
83-
numberOfBytes -= bytesToConsumeForChunk;
84-
}
85-
return into;
86-
}
87-
8862
private ZigZagLongDecoding() {
8963
throw new UnsupportedOperationException("No instances");
9064
}

‎core/src/main/java/org/neo4j/gds/core/compression/packed/PackedCompressor.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,8 @@ public static AdjacencyCompressorFactory factory(
5050
boolean noAggregation,
5151
MemoryTracker memoryTracker
5252
) {
53-
AdjacencyListBuilder<long[], ? extends AdjacencyProperties>[] propertyBuilders = new AdjacencyListBuilder[propertyMappings.numberOfMappings()];
53+
AdjacencyListBuilder<long[], ? extends AdjacencyProperties>[] propertyBuilders = new AdjacencyListBuilder[propertyMappings
54+
.numberOfMappings()];
5455
Arrays.setAll(propertyBuilders, i -> adjacencyListBuilderFactory.newAdjacencyPropertiesBuilder(memoryTracker));
5556

5657
return new Factory(
@@ -171,7 +172,7 @@ private PackedCompressor(
171172
}
172173

173174
@Override
174-
public int compress(long nodeId, long[] targets, long[][][] properties, int degree) {
175+
public int compress(long nodeId, long[] targets, long[][] properties, int degree) {
175176
if (properties != null) {
176177
return packWithProperties(
177178
nodeId,
@@ -191,7 +192,7 @@ public int compress(long nodeId, long[] targets, long[][][] properties, int degr
191192
private int packWithProperties(
192193
long nodeId,
193194
long[] targets,
194-
long[][][] unsortedProperties,
195+
long[][] unsortedProperties,
195196
int degree
196197
) {
197198
long[][] sortedProperties = new long[unsortedProperties.length][degree];

‎core/src/main/java/org/neo4j/gds/core/compression/uncompressed/RawCompressor.java‎

Lines changed: 19 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020
package org.neo4j.gds.core.compression.uncompressed;
2121

2222
import com.carrotsearch.hppc.sorting.IndirectSort;
23-
import org.apache.commons.lang3.mutable.MutableInt;
2423
import org.jetbrains.annotations.Nullable;
2524
import org.neo4j.gds.PropertyMappings;
2625
import org.neo4j.gds.api.AdjacencyList;
@@ -40,8 +39,6 @@
4039
import java.util.Arrays;
4140
import java.util.function.LongSupplier;
4241

43-
import static org.neo4j.gds.core.compression.common.AdjacencyCompression.findPosition;
44-
4542
public final class RawCompressor implements AdjacencyCompressor {
4643

4744
public static AdjacencyCompressorFactory factory(
@@ -52,8 +49,10 @@ public static AdjacencyCompressorFactory factory(
5249
boolean noAggregation,
5350
MemoryTracker memoryTracker
5451
) {
55-
@SuppressWarnings("unchecked")
56-
AdjacencyListBuilder<long[], ? extends AdjacencyProperties>[] propertyBuilders = new AdjacencyListBuilder[propertyMappings.numberOfMappings()];
52+
@SuppressWarnings(
53+
"unchecked"
54+
) AdjacencyListBuilder<long[], ? extends AdjacencyProperties>[] propertyBuilders = new AdjacencyListBuilder[propertyMappings
55+
.numberOfMappings()];
5756
Arrays.setAll(propertyBuilders, i -> adjacencyListBuilderFactory.newAdjacencyPropertiesBuilder(memoryTracker));
5857

5958
return new Factory(
@@ -158,7 +157,7 @@ private RawCompressor(
158157
}
159158

160159
@Override
161-
public int compress(long nodeId, long[] targets, long[][][] properties, int degree) {
160+
public int compress(long nodeId, long[] targets, long[][] properties, int degree) {
162161
if (properties != null) {
163162
return withProperties(
164163
nodeId,
@@ -168,7 +167,9 @@ public int compress(long nodeId, long[] targets, long[][][] properties, int degr
168167
);
169168
} else {
170169
return withoutProperties(
171-
nodeId, targets, degree
170+
nodeId,
171+
targets,
172+
degree
172173
);
173174
}
174175
}
@@ -222,57 +223,44 @@ private int aggregate(long[] values, int length, Aggregation aggregation) {
222223
private int aggregateWithProperties(
223224
long[] values,
224225
int length,
225-
long[][][] unsortedProperties,
226+
long[][] unsortedProperties,
226227
long[][] sortedProperties,
227228
Aggregation[] aggregations
228229
) {
229230
int[] order = IndirectSort.mergesort(0, length, new AscendingLongComparator(values));
230231

232+
long[] outValues = new long[length];
233+
231234
int firstSortIdx = order[0];
232235
long value = values[firstSortIdx];
233236
long delta;
234237

235-
236-
int[] chunkLengths = new int[unsortedProperties[0].length];
237-
int totalChunkLength = 0;
238-
for (int i = 0; i < unsortedProperties[0].length; i++) {
239-
totalChunkLength += unsortedProperties[0][i].length;
240-
chunkLengths[i] = totalChunkLength;
241-
}
242-
243-
var pageIndex = new MutableInt();
244-
var indexInPage = new MutableInt();
245-
findPosition(chunkLengths, firstSortIdx, pageIndex, indexInPage);
246-
247-
long[] outValues = new long[length];
248238
outValues[0] = value;
249239
for (int i = 0; i < unsortedProperties.length; i++) {
250-
sortedProperties[i][0] = unsortedProperties[i][pageIndex.intValue()][indexInPage.intValue()];
240+
sortedProperties[i][0] = unsortedProperties[i][firstSortIdx];
251241
}
252242

253243
int in = 1, out = 1;
254244

255245
if (this.noAggregation) {
256246
for (; in < length; ++in) {
257247
int sortIdx = order[in];
258-
findPosition(chunkLengths, sortIdx, pageIndex, indexInPage);
259248

260249
for (int i = 0; i < unsortedProperties.length; i++) {
261-
sortedProperties[i][out] = unsortedProperties[i][pageIndex.intValue()][indexInPage.intValue()];
250+
sortedProperties[i][out] = unsortedProperties[i][sortIdx];
262251
}
263252

264253
outValues[out++] = values[sortIdx];
265254
}
266255
} else {
267256
for (; in < length; ++in) {
268257
int sortIdx = order[in];
269-
findPosition(chunkLengths, sortIdx, pageIndex, indexInPage);
270258
delta = values[sortIdx] - value;
271259
value = values[sortIdx];
272260

273261
if (delta > 0L) {
274262
for (int i = 0; i < unsortedProperties.length; i++) {
275-
sortedProperties[i][out] = unsortedProperties[i][pageIndex.intValue()][indexInPage.intValue()];
263+
sortedProperties[i][out] = unsortedProperties[i][sortIdx];
276264
}
277265
outValues[out++] = value;
278266
} else {
@@ -281,7 +269,7 @@ private int aggregateWithProperties(
281269
int existingIdx = out - 1;
282270
long[] outProperty = sortedProperties[i];
283271
double existingProperty = Double.longBitsToDouble(outProperty[existingIdx]);
284-
double newProperty = Double.longBitsToDouble(unsortedProperties[i][pageIndex.intValue()][indexInPage.intValue()]);
272+
double newProperty = Double.longBitsToDouble(unsortedProperties[i][sortIdx]);
285273
newProperty = aggregation.merge(existingProperty, newProperty);
286274
outProperty[existingIdx] = Double.doubleToLongBits(newProperty);
287275
}
@@ -290,6 +278,9 @@ private int aggregateWithProperties(
290278
}
291279

292280
System.arraycopy(outValues, 0, values, 0, out);
281+
for (int i = 0; i < sortedProperties.length; i++) {
282+
System.arraycopy(sortedProperties[i], 0, unsortedProperties[i], 0, out);
283+
}
293284

294285
return out;
295286

@@ -298,7 +289,7 @@ private int aggregateWithProperties(
298289
private int withProperties(
299290
long nodeId,
300291
long[] targets,
301-
long[][][] unsortedProperties,
292+
long[][] unsortedProperties,
302293
int degree
303294
) {
304295
long[][] sortedProperties = new long[unsortedProperties.length][degree];

0 commit comments

Comments
 (0)