Set joiner

Netflix · Aug 20, 2024 · 3767beb · 3767beb
1 parent a475e6e
commit 3767beb
Show file tree

Hide file tree

Showing 2 changed files with 64 additions and 2 deletions.
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
@@ -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.AbstractHollowTypeDataElementsJoiner;
 
 
@@ -22,9 +25,68 @@ public void init() {
 
     @Override
     public void populateStats() {
+        for(int fromIndex=0;fromIndex<from.length;fromIndex++) {
+            int mappedMaxOrdinal = from[fromIndex].maxOrdinal == -1 ? -1 : (from[fromIndex].maxOrdinal * from.length) + fromIndex;
+            to.maxOrdinal = Math.max(to.maxOrdinal, mappedMaxOrdinal);
+
+            // uneven bitsPerElement could be the case for consumers that skip type shards with no additions, so pick max across all shards
+            HollowSetTypeDataElements source = from[fromIndex];
+            if (source.bitsPerElement > to.bitsPerElement) {
+                to.bitsPerElement = source.bitsPerElement;
+            }
+            if (source.bitsPerSetSizeValue > to.bitsPerSetSizeValue) {
+                to.bitsPerSetSizeValue = source.bitsPerSetSizeValue;
+            }
+        }
+        to.emptyBucketValue = (1 << to.bitsPerElement) - 1;
+
+        long totalOfSetBuckets = 0;
+        for(int ordinal=0;ordinal<=to.maxOrdinal;ordinal++) {
+            int fromIndex = ordinal & fromMask;
+            int fromOrdinal = ordinal >> fromOrdinalShift;
+
+            HollowSetTypeDataElements source = from[fromIndex];
+
+            long startBucket = getAbsoluteBucketStart(source, ordinal);
+            long endBucket = source.setPointerAndSizeData.getElementValue((long)fromOrdinal * source.bitsPerFixedLengthSetPortion, source.bitsPerSetPointer);
+            long numBuckets = endBucket - startBucket;
+
+            totalOfSetBuckets += numBuckets;
+        }
+
+        // retained because these are computed based on max across all shards, splitting has no effect
+        to.totalNumberOfBuckets = totalOfSetBuckets;
+        to.bitsPerSetPointer = 64 - Long.numberOfLeadingZeros(to.totalNumberOfBuckets);
+        to.bitsPerFixedLengthSetPortion = to.bitsPerSetPointer + to.bitsPerSetSizeValue;
+
+        to.setPointerAndSizeData = FixedLengthDataFactory.get(((long)to.maxOrdinal + 1) * to.bitsPerFixedLengthSetPortion, to.memoryMode, to.memoryRecycler);
+        to.elementData = FixedLengthDataFactory.get(to.totalNumberOfBuckets * to.bitsPerElement, to.memoryMode, to.memoryRecycler);
     }
 
     @Override
     public void copyRecords() {
+        long bucketCounter = 0;
+        for(int ordinal=0;ordinal<=to.maxOrdinal;ordinal++) {
+            int fromIndex = ordinal & fromMask;
+            int fromOrdinal = ordinal >> fromOrdinalShift;
+
+            HollowSetTypeDataElements source = from[fromIndex];
+
+            if (fromOrdinal <= from[fromIndex].maxOrdinal) {
+                long startBucket = getAbsoluteBucketStart(source, fromOrdinal);
+                long endBucket = source.setPointerAndSizeData.getElementValue((long)fromOrdinal * source.bitsPerFixedLengthSetPortion, source.bitsPerSetPointer);
+
+                for (long bucket=startBucket;bucket<endBucket;bucket++) {
+                    int bucketOrdinal = (int)source.elementData.getElementValue(bucket * source.bitsPerElement, source.bitsPerElement);
+                    to.elementData.setElementValue(bucketCounter * to.bitsPerElement, to.bitsPerElement, bucketOrdinal);
+                    bucketCounter++;
+                }
+            } // else: lopsided shards could result for consumers that skip type shards with no additions, that gets handled
+            // by not writing anything to elementData, and writing the cached value of bucketCounter to listPointerData
+            // SNAP: TODO: write a test for lopsided list shards (similar to for list types)
+
+            long setSize = source.setPointerAndSizeData.getElementValue((fromOrdinal * source.bitsPerFixedLengthSetPortion) + source.bitsPerSetPointer, source.bitsPerSetSizeValue);
+            to.setPointerAndSizeData.setElementValue((ordinal * to.bitsPerFixedLengthSetPortion) + to.bitsPerSetPointer, to.bitsPerSetSizeValue, setSize);
+        }
     }
 }
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
@@ -48,12 +48,12 @@ public void populateStats() {
 
         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
+            // retained because these are computed based on max across all shards, splitting has no effect
             target.bitsPerElement = from.bitsPerElement;
             target.bitsPerSetSizeValue = from.bitsPerSetSizeValue;
             target.emptyBucketValue = from.emptyBucketValue;
 
-            // recomputed based on max per split shards
+            // recomputed based on split shards
             target.bitsPerSetPointer = 64 - Long.numberOfLeadingZeros(maxShardTotalOfSetBuckets);
             target.totalNumberOfBuckets = shardTotalOfSetBuckets[toIndex];
             target.bitsPerFixedLengthSetPortion = target.bitsPerSetPointer + target.bitsPerSetSizeValue;