Set splitter

Netflix · Aug 19, 2024 · a475e6e · a475e6e
1 parent aaed371
commit a475e6e
Show file tree

Hide file tree

Showing 7 changed files with 165 additions and 12 deletions.
diff --git a/...ain/java/com/netflix/hollow/core/read/engine/list/HollowListTypeDataElementsSplitter.java b/...ain/java/com/netflix/hollow/core/read/engine/list/HollowListTypeDataElementsSplitter.java
@@ -25,7 +25,6 @@ public void init() {
 
     @Override
     public void populateStats() {
-        int numSplits = to.length;
         long[] totalOfListSizes  = new long[numSplits];
 
         // count elements per split
@@ -57,13 +56,14 @@ public void populateStats() {
         }
 
         for(int toIndex=0;toIndex<numSplits;toIndex++) {
-            to[toIndex].bitsPerElement = from.bitsPerElement;   // retained: it's the max across all shards in type, so splitting has no effect
-            to[toIndex].bitsPerListPointer = maxShardTotalOfListSizes == 0 ? 1 : 64 - Long.numberOfLeadingZeros(maxShardTotalOfListSizes);
+            HollowListTypeDataElements target = to[toIndex];
+            target.bitsPerElement = from.bitsPerElement;   // retained: it's the max across all shards in type, so splitting has no effect
+            target.bitsPerListPointer = maxShardTotalOfListSizes == 0 ? 1 : 64 - Long.numberOfLeadingZeros(maxShardTotalOfListSizes);
 
-            to[toIndex].listPointerData = FixedLengthDataFactory.get((long)to[toIndex].bitsPerListPointer * (to[toIndex].maxOrdinal + 1), to[toIndex].memoryMode, to[toIndex].memoryRecycler);
-            to[toIndex].elementData = FixedLengthDataFactory.get((long)to[toIndex].bitsPerElement * totalOfListSizes[toIndex], to[toIndex].memoryMode, to[toIndex].memoryRecycler);
+            target.listPointerData = FixedLengthDataFactory.get((long)target.bitsPerListPointer * (target.maxOrdinal + 1), target.memoryMode, target.memoryRecycler);
+            target.elementData = FixedLengthDataFactory.get((long)target.bitsPerElement * totalOfListSizes[toIndex], target.memoryMode, target.memoryRecycler);
 
-            to[toIndex].totalNumberOfElements = totalOfListSizes[toIndex];  // useful for heap usage stats
+            target.totalNumberOfElements = totalOfListSizes[toIndex];  // useful for heap usage stats
         }
     }
 
@@ -79,6 +79,7 @@ public void copyRecords() {
 
             long startElement;
             long endElement;
+
             if (ordinal == 0) {
                 startElement = 0;
                 endElement = from.listPointerData.getElementValue(0, from.bitsPerListPointer);
@@ -89,12 +90,13 @@ public void copyRecords() {
                 endElement = from.listPointerData.getElementValue(endFixedLengthOffset, from.bitsPerListPointer);
             }
 
+            HollowListTypeDataElements target = to[toIndex];
             for (long element=startElement;element<endElement;element++) {
                 int elementOrdinal = (int)from.elementData.getElementValue(element * from.bitsPerElement, from.bitsPerElement);
-                to[toIndex].elementData.setElementValue(elementCounter[toIndex] * to[toIndex].bitsPerElement, to[toIndex].bitsPerElement, elementOrdinal);
+                target.elementData.setElementValue(elementCounter[toIndex] * target.bitsPerElement, target.bitsPerElement, elementOrdinal);
                 elementCounter[toIndex]++;
             }
-            to[toIndex].listPointerData.setElementValue(to[toIndex].bitsPerListPointer * toOrdinal, to[toIndex].bitsPerListPointer, elementCounter[toIndex]);
+            target.listPointerData.setElementValue(target.bitsPerListPointer * toOrdinal, target.bitsPerListPointer, elementCounter[toIndex]);
         }
     }
 }
diff --git a/...rc/main/java/com/netflix/hollow/core/read/engine/map/HollowMapTypeDataElementsJoiner.java b/...rc/main/java/com/netflix/hollow/core/read/engine/map/HollowMapTypeDataElementsJoiner.java
@@ -0,0 +1,30 @@
+package com.netflix.hollow.core.read.engine.map;
+
+import com.netflix.hollow.core.read.engine.AbstractHollowTypeDataElementsJoiner;
+
+
+/**
+ * Join multiple {@code HollowMapTypeDataElements}s into 1 {@code HollowMapTypeDataElements}.
+ * Ordinals are remapped and corresponding data is copied over.
+ * The original data elements are not destroyed.
+ * The no. of passed data elements must be a power of 2.
+ */
+class HollowMapTypeDataElementsJoiner extends AbstractHollowTypeDataElementsJoiner<HollowMapTypeDataElements> {
+
+    public HollowMapTypeDataElementsJoiner(HollowMapTypeDataElements[] from) {
+        super(from);
+    }
+
+    @Override
+    public void init() {
+        this.to = new HollowMapTypeDataElements(from[0].memoryMode, from[0].memoryRecycler);
+    }
+
+    @Override
+    public void populateStats() {
+    }
+
+    @Override
+    public void copyRecords() {
+    }
+}
diff --git a/.../main/java/com/netflix/hollow/core/read/engine/map/HollowMapTypeDataElementsSplitter.java b/.../main/java/com/netflix/hollow/core/read/engine/map/HollowMapTypeDataElementsSplitter.java
@@ -0,0 +1,34 @@
+package com.netflix.hollow.core.read.engine.map;
+
+import com.netflix.hollow.core.read.engine.AbstractHollowTypeDataElementsSplitter;
+
+/**
+ * Split a {@code HollowMapTypeDataElements} into multiple {@code HollowMapTypeDataElements}s.
+ * Ordinals are remapped and corresponding data is copied over.
+ * The original data elements are not destroyed.
+ * {@code numSplits} must be a power of 2.
+ */
+public class HollowMapTypeDataElementsSplitter extends AbstractHollowTypeDataElementsSplitter<HollowMapTypeDataElements> {
+
+    public HollowMapTypeDataElementsSplitter(HollowMapTypeDataElements from, int numSplits) {
+        super(from, numSplits);
+    }
+
+    @Override
+    public void init() {
+        this.to = new HollowMapTypeDataElements[numSplits];
+        for(int i=0;i<to.length;i++) {
+            to[i] = new HollowMapTypeDataElements(from.memoryMode, from.memoryRecycler);
+        }
+    }
+
+    @Override
+    public void populateStats() {
+
+    }
+
+    @Override
+    public void copyRecords() {
+
+    }
+}
diff --git a/hollow/src/main/java/com/netflix/hollow/core/read/engine/set/HollowSetTypeDataElements.java b/hollow/src/main/java/com/netflix/hollow/core/read/engine/set/HollowSetTypeDataElements.java
@@ -35,8 +35,8 @@
 
 public class HollowSetTypeDataElements extends AbstractHollowTypeDataElements {
 
-    FixedLengthData setPointerAndSizeData;
-    FixedLengthData elementData;
+    public FixedLengthData setPointerAndSizeData;
+    public FixedLengthData elementData;
 
     int bitsPerSetPointer;
     int bitsPerSetSizeValue;

diff --git a/...rc/main/java/com/netflix/hollow/core/read/engine/set/HollowSetTypeDataElementsJoiner.java b/...rc/main/java/com/netflix/hollow/core/read/engine/set/HollowSetTypeDataElementsJoiner.java
@@ -0,0 +1,30 @@
+package com.netflix.hollow.core.read.engine.set;
+
+import com.netflix.hollow.core.read.engine.AbstractHollowTypeDataElementsJoiner;
+
+
+/**
+ * Join multiple {@code HollowSetTypeDataElements}s into 1 {@code HollowSetTypeDataElements}.
+ * Ordinals are remapped and corresponding data is copied over.
+ * The original data elements are not destroyed.
+ * The no. of passed data elements must be a power of 2.
+ */
+class HollowSetTypeDataElementsJoiner extends AbstractHollowTypeDataElementsJoiner<HollowSetTypeDataElements> {
+
+    public HollowSetTypeDataElementsJoiner(HollowSetTypeDataElements[] from) {
+        super(from);
+    }
+
+    @Override
+    public void init() {
+        this.to = new HollowSetTypeDataElements(from[0].memoryMode, from[0].memoryRecycler);
+    }
+
+    @Override
+    public void populateStats() {
+    }
+
+    @Override
+    public void copyRecords() {
+    }
+}
diff --git a/.../main/java/com/netflix/hollow/core/read/engine/set/HollowSetTypeDataElementsSplitter.java b/.../main/java/com/netflix/hollow/core/read/engine/set/HollowSetTypeDataElementsSplitter.java
@@ -1,5 +1,8 @@
 package com.netflix.hollow.core.read.engine.set;
 
+import static com.netflix.hollow.core.read.engine.set.HollowSetTypeReadStateShard.getAbsoluteBucketStart;
+
+import com.netflix.hollow.core.memory.FixedLengthDataFactory;
 import com.netflix.hollow.core.read.engine.AbstractHollowTypeDataElementsSplitter;
 
 /**
@@ -24,11 +27,65 @@ public void init() {
 
     @Override
     public void populateStats() {
+        long[] shardTotalOfSetBuckets  = new long[numSplits];
+        long maxShardTotalOfSetBuckets = 0;
+
+        // count buckets per split
+        for(int ordinal=0;ordinal<=from.maxOrdinal;ordinal++) {
+            int toIndex = ordinal & toMask;
+            int toOrdinal = ordinal >> toOrdinalShift;
+            to[toIndex].maxOrdinal = toOrdinal;
+
+            long startBucket = getAbsoluteBucketStart(from, ordinal);
+            long endBucket = from.setPointerAndSizeData.getElementValue((long)ordinal * from.bitsPerFixedLengthSetPortion, from.bitsPerSetPointer);
+            long numBuckets = endBucket - startBucket;
+
+            shardTotalOfSetBuckets[toIndex] += numBuckets;
+            if(shardTotalOfSetBuckets[toIndex] > maxShardTotalOfSetBuckets) {
+                maxShardTotalOfSetBuckets = shardTotalOfSetBuckets[toIndex];
+            }
+        }
+
+        for(int toIndex=0;toIndex<numSplits;toIndex++) {
+            HollowSetTypeDataElements target = to[toIndex];
+            // retained because these are computed based on max across all shards, so splitting has no effect
+            target.bitsPerElement = from.bitsPerElement;
+            target.bitsPerSetSizeValue = from.bitsPerSetSizeValue;
+            target.emptyBucketValue = from.emptyBucketValue;
 
+            // recomputed based on max per split shards
+            target.bitsPerSetPointer = 64 - Long.numberOfLeadingZeros(maxShardTotalOfSetBuckets);
+            target.totalNumberOfBuckets = shardTotalOfSetBuckets[toIndex];
+            target.bitsPerFixedLengthSetPortion = target.bitsPerSetPointer + target.bitsPerSetSizeValue;
+
+            target.setPointerAndSizeData = FixedLengthDataFactory.get(((long)target.maxOrdinal + 1) * target.bitsPerFixedLengthSetPortion, target.memoryMode, target.memoryRecycler);
+            target.elementData = FixedLengthDataFactory.get(target.totalNumberOfBuckets * target.bitsPerElement, target.memoryMode, target.memoryRecycler);
+        }
     }
 
     @Override
     public void copyRecords() {
+        int numSplits = to.length;
+        long bucketCounter[] = new long[numSplits];
+
+        // count elements per split
+        for(int ordinal=0;ordinal<=from.maxOrdinal;ordinal++) {
+            int toIndex = ordinal & toMask;
+            int toOrdinal = ordinal >> toOrdinalShift;
 
+            long startBucket = getAbsoluteBucketStart(from, ordinal);
+            long endBucket = from.setPointerAndSizeData.getElementValue((long)ordinal * from.bitsPerFixedLengthSetPortion, from.bitsPerSetPointer);
+
+            HollowSetTypeDataElements target = to[toIndex];
+            for (long bucket=startBucket;bucket<endBucket;bucket++) {
+                long bucketVal = from.elementData.getElementValue(bucket * from.bitsPerElement, from.bitsPerElement);
+                if(bucketVal == from.emptyBucketValue)
+                    bucketVal = target.emptyBucketValue;
+                target.elementData.setElementValue(bucketCounter[toIndex] * target.bitsPerElement, target.bitsPerElement, bucketVal);
+                bucketCounter[toIndex]++;
+            }
+            long setSize = from.setPointerAndSizeData.getElementValue((ordinal * from.bitsPerFixedLengthSetPortion) + from.bitsPerSetPointer, from.bitsPerSetSizeValue);
+            target.setPointerAndSizeData.setElementValue((toOrdinal * target.bitsPerFixedLengthSetPortion) + target.bitsPerSetPointer, target.bitsPerSetSizeValue, setSize);
+        }
     }
 }
diff --git a/...ow/src/main/java/com/netflix/hollow/core/read/engine/set/HollowSetTypeReadStateShard.java b/...ow/src/main/java/com/netflix/hollow/core/read/engine/set/HollowSetTypeReadStateShard.java
@@ -139,11 +139,11 @@ public int relativeBucketValue(int setOrdinal, int bucketIndex) {
         return value;
     }
 
-    private long getAbsoluteBucketStart(HollowSetTypeDataElements currentData, int ordinal) {
+    public static long getAbsoluteBucketStart(HollowSetTypeDataElements currentData, int ordinal) {
         return ordinal == 0 ? 0 : currentData.setPointerAndSizeData.getElementValue((long)(ordinal - 1) * currentData.bitsPerFixedLengthSetPortion, currentData.bitsPerSetPointer);
     }
 
-    private int absoluteBucketValue(HollowSetTypeDataElements currentData, long absoluteBucketIndex) {
+    public static int absoluteBucketValue(HollowSetTypeDataElements currentData, long absoluteBucketIndex) {
         return (int)currentData.elementData.getElementValue(absoluteBucketIndex * currentData.bitsPerElement, currentData.bitsPerElement);
     }