Decode and determine equality for strings directly from immutable byte segments

hollow-perf/src/jmh/java/com/netflix/hollow/core/read/engine/object/HollowObjectTypeReadStateShardBenchmark.java

@@ -1,6 +1,10 @@
 package com.netflix.hollow.core.read.engine.object;
 
+import com.netflix.hollow.core.memory.MemoryMode;
+import com.netflix.hollow.core.memory.pool.WastefulRecycler;
+import com.netflix.hollow.core.read.engine.HollowReadConfiguration;
 import com.netflix.hollow.core.read.dataaccess.HollowObjectTypeDataAccess;
+import com.netflix.hollow.core.read.engine.ExperimentalFeature;
 import com.netflix.hollow.core.read.engine.HollowReadStateEngine;
 import com.netflix.hollow.core.util.StateEngineRoundTripper;
 import com.netflix.hollow.core.write.HollowWriteStateEngine;
@@ -41,20 +45,25 @@ public class HollowObjectTypeReadStateShardBenchmark {
     int countStringsDb;
 
     ArrayList<Integer> readOrder;
+    ArrayList<String> strings;
 
     @Param({ "5", "25", "50", "150", "1000" })
     int maxStringLength;
 
     @Param({ "10" })
     int probabilityUnicode;
 
+    @Param({ "false", "true" })
+    boolean directSegmentStrings;
+
     @Setup
     public void setUp() throws IOException {
         writeStateEngine = new HollowWriteStateEngine();
         objectMapper = new HollowObjectMapper(writeStateEngine);
         objectMapper.initializeTypeState(String.class);
 
         Random r = new Random();
+        strings = new ArrayList<>(countStringsDb);
         for (int i = 0; i < countStringsDb; i++) {
             StringBuilder sb = new StringBuilder();
             sb.append("string_");
@@ -68,15 +77,20 @@ public void setUp() throws IOException {
                     sb.append((char) (r.nextInt(26) + 'a'));
                 }
             }
-            objectMapper.add(sb.toString());
+            String value = sb.toString();
+            objectMapper.add(value);
+            strings.add(value);
         }
 
         readOrder = new ArrayList<>(countStrings);
         for (int i = 0; i < countStrings; i++) {
             readOrder.add(r.nextInt(countStringsDb));
         }
 
-        readStateEngine = new HollowReadStateEngine();
+        readStateEngine = new HollowReadStateEngine(new HollowReadConfiguration(MemoryMode.ON_HEAP,
+                WastefulRecycler.DEFAULT_INSTANCE, directSegmentStrings
+                ? new ExperimentalFeature[] { ExperimentalFeature.SHARD_READ_FAST_PATHS, ExperimentalFeature.DIRECT_SEGMENT_STRING_READS }
+                : new ExperimentalFeature[] { ExperimentalFeature.SHARD_READ_FAST_PATHS }));
 
         StateEngineRoundTripper.roundTripSnapshot(writeStateEngine, readStateEngine, null);
         dataAccess = (HollowObjectTypeDataAccess) readStateEngine.getTypeDataAccess("String", 0);
@@ -90,4 +104,11 @@ public void testReadString(Blackhole bh) {
             bh.consume(result);
         }
     }
+
+    @Benchmark
+    public void testStringEquality(Blackhole bh) {
+        for (int j : readOrder) {
+            bh.consume(dataAccess.isStringFieldEqual(j, 0, strings.get(j)));
+        }
+    }
 }

hollow/src/main/java/com/netflix/hollow/core/memory/SegmentedByteArray.java

@@ -20,8 +20,12 @@
 import com.netflix.hollow.core.read.HollowBlobInput;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.nio.CharBuffer;
+import java.nio.charset.CharsetDecoder;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import sun.misc.Unsafe;
 
 /**
  * A segmented byte array backs the {@link ByteData} interface with array segments, which potentially come from a pool of reusable memory.<p>
@@ -42,7 +46,14 @@
 @SuppressWarnings("restriction")
 public class SegmentedByteArray implements VariableLengthData {
 
     private static final Unsafe unsafe = HollowUnsafeHandle.getUnsafe();
+    // JDK 19 changed String.decodeASCII to use the prefix-returning countPositives intrinsic.
+    private static final boolean asciiDecoderFastPath = javaMajorVersion() >= 19;
+
+    private static class AsciiDecoderHolder {
+        private static final ThreadLocal<CharsetDecoder> decoder =
+                ThreadLocal.withInitial(() -> StandardCharsets.US_ASCII.newDecoder());
+    }
 
     private byte[][] segments;
     private final int log2OfSegmentSize;
@@ -77,6 +88,166 @@
         return segments[(int)(index >>> log2OfSegmentSize)][(int)(index & bitmask)];
     }
 
+    /**
+     * Decode a String whose UTF-16 code units are encoded as variable-length integers in an
+     * immutable range of this array. Segment references are loaded once per contiguous range
+     * rather than once per encoded byte.
+     *
+     * <p>On JDK 19 and later, a contiguous range is first passed through the intrinsic-backed
+     * ASCII decoder to identify and inflate its ASCII prefix in bulk. A fully ASCII range is passed
+     * directly to String's ISO-8859-1 constructor. Other content is decoded to UTF-16 code units
+     * and passed to String, which performs any Latin-1 compression using its JIT intrinsic.
+     *
+     * @param position the position of the first encoded byte
+     * @param length the number of encoded bytes
+     * @param output scratch space with capacity for at least {@code length} characters
+     * @return the decoded String
+     */
+    public String readVIntString(long position, int length, char[] output) {
+        if(length == 0)
+            return "";
+
+        byte[][] currentSegments = segments;
+        int segmentSize = 1 << log2OfSegmentSize;
+        int firstSegmentIndex = (int)(position >>> log2OfSegmentSize);
+        int firstSegmentOffset = (int)(position & bitmask);
+        if(asciiDecoderFastPath && firstSegmentOffset + length <= segmentSize)
+            return readContiguousVIntString(currentSegments[firstSegmentIndex], firstSegmentOffset, length, output);
+
+        int i = 0;
+
+        ascii:
+        while(i < length) {
+            long currentPosition = position + i;
+            int segmentOffset = (int)(currentPosition & bitmask);
+            int end = Math.min(length, i + segmentSize - segmentOffset);
+            byte[] segment = currentSegments[(int)(currentPosition >>> log2OfSegmentSize)];
+
+            while(i < end) {
+                int b = segment[segmentOffset++];
+                if(b < 0)
+                    break ascii;
+                output[i++] = (char)b;
+            }
+        }
+
+        if(i == length) {
+            if(firstSegmentOffset + length <= segmentSize)
+                return new String(currentSegments[firstSegmentIndex], firstSegmentOffset, length,
+                        StandardCharsets.ISO_8859_1);
+            return new String(output, 0, length);
+        }
+
+        int count = i;
+        int value = 0;
+        while(i < length) {
+            long currentPosition = position + i;
+            int segmentOffset = (int)(currentPosition & bitmask);
+            int end = Math.min(length, i + segmentSize - segmentOffset);
+            byte[] segment = currentSegments[(int)(currentPosition >>> log2OfSegmentSize)];
+
+            while(i < end) {
+                int b = segment[segmentOffset++];
+                value = (value << 7) | (b & 0x7f);
+                i++;
+                if(b >= 0) {
+                    output[count++] = (char)value;
+                    value = 0;
+                }
+            }
+        }
+
+        return new String(output, 0, count);
+    }
+
+    private static String readContiguousVIntString(byte[] data, int offset, int length, char[] output) {
+        ByteBuffer source = ByteBuffer.wrap(data, offset, length);
+        CharBuffer target = CharBuffer.wrap(output, 0, length);
+        CharsetDecoder decoder = AsciiDecoderHolder.decoder.get();
+        decoder.reset();
+        decoder.decode(source, target, true);
+
+        if(!source.hasRemaining())
+            return new String(data, offset, length, StandardCharsets.ISO_8859_1);
+
+        int value = 0;
+        byte b;
+        do {
+            b = source.get();
+            value = (value << 7) | (b & 0x7f);
+        } while(b < 0);
+        target.put((char)value);
+
+        int input = source.position();
+        int end = offset + length;
+        int count = target.position();
+        value = 0;
+        while(input < end) {
+            b = data[input++];
+            value = (value << 7) | (b & 0x7f);
+            if(b >= 0) {
+                output[count++] = (char)value;
+                value = 0;
+            }
+        }
+        return new String(output, 0, count);
+    }
+
+    private static int javaMajorVersion() {
+        String version = System.getProperty("java.specification.version", "1.8");
+        int start = version.startsWith("1.") ? 2 : 0;
+        int end = version.indexOf('.', start);
+        try {
+            return Integer.parseInt(end == -1 ? version.substring(start) : version.substring(start, end));
+        } catch(NumberFormatException ignored) {
+            return 8;
+        }
+    }
+
+    /**
+     * Compare a String with UTF-16 code units encoded as variable-length integers in an immutable
+     * range of this array. Segment references are loaded once per contiguous range rather than
+     * once per encoded character.
+     *
+     * @param position the position of the first encoded byte
+     * @param length the number of encoded bytes
+     * @param testValue the String to compare
+     * @return true if the encoded range is equal to the String
+     */
+    public boolean isVIntStringEqual(long position, int length, String testValue) {
+        int testLength = testValue.length();
+        if(length < testLength)
+            return false;
+
+        byte[][] currentSegments = segments;
+        int segmentSize = 1 << log2OfSegmentSize;
+        int i = 0;
+        int count = 0;
+        int value = 0;
+        boolean complete = true;
+
+        while(i < length) {
+            long currentPosition = position + i;
+            int segmentOffset = (int)(currentPosition & bitmask);
+            int end = Math.min(length, i + segmentSize - segmentOffset);
+            byte[] segment = currentSegments[(int)(currentPosition >>> log2OfSegmentSize)];
+
+            while(i < end) {
+                int b = segment[segmentOffset++];
+                value = (value << 7) | (b & 0x7f);
+                i++;
+                complete = b >= 0;
+                if(complete) {
+                    if(count == testLength || testValue.charAt(count++) != (char)value)
+                        return false;
+                    value = 0;
+                }
+            }
+        }
+
+        return complete && count == testLength;
+    }
+
     @Override
     public void copy(ByteData src, long srcPos, long destPos, long length) {
         for(long i=0;i<length;i++) {
@@ -254,7 +425,7 @@
 
     private void orderedCopy(byte[] src, int srcPos, byte[] dest, int destPos, int length) {
         int endSrcPos = srcPos + length;
         destPos += Unsafe.ARRAY_BYTE_BASE_OFFSET;
 
         while(srcPos < endSrcPos) {
             unsafe.putByteVolatile(dest, destPos++, src[srcPos++]);

hollow/src/main/java/com/netflix/hollow/core/read/engine/ExperimentalFeature.java

@@ -14,5 +14,8 @@ public enum ExperimentalFeature {
      * with contiguous fixed-length storage, bulk byte-field reads, and scalar fast paths.
      * Does not change recycler selection or permit validation elision for recycling arrays.
      */
-    SHARD_READ_FAST_PATHS
+    SHARD_READ_FAST_PATHS,
+
+    /** Direct decoding and equality checks against immutable on-heap byte segments. */
+    DIRECT_SEGMENT_STRING_READS
 }

hollow/src/main/java/com/netflix/hollow/core/read/engine/object/HollowObjectTypeReadState.java

@@ -30,6 +30,7 @@
 import com.netflix.hollow.core.memory.pool.ArraySegmentRecycler;
 import com.netflix.hollow.core.read.HollowBlobInput;
 import com.netflix.hollow.core.read.dataaccess.HollowObjectTypeDataAccess;
+import com.netflix.hollow.core.read.engine.ExperimentalFeature;
 import com.netflix.hollow.core.read.engine.HollowReadStateEngine;
 import com.netflix.hollow.core.read.engine.HollowTypeDataElements;
 import com.netflix.hollow.core.read.engine.HollowTypeReadState;
@@ -49,6 +50,7 @@
 public class HollowObjectTypeReadState extends HollowTypeReadState implements HollowObjectTypeDataAccess {
     private final HollowObjectSchema unfilteredSchema;
     private final HollowObjectSampler sampler;
+    private final boolean useDirectSegmentStringReads;
 
     private int maxOrdinal;
 
@@ -77,13 +79,16 @@ public HollowTypeReadStateShard createTypeReadStateShard(HollowSchema schema, Ho
     public HollowObjectTypeReadState(HollowReadStateEngine fileEngine, MemoryMode memoryMode, HollowObjectSchema schema, HollowObjectSchema unfilteredSchema) {
         super(fileEngine, memoryMode, schema);
         this.sampler = new HollowObjectSampler(schema, DisabledSamplingDirector.INSTANCE);
+        this.useDirectSegmentStringReads = fileEngine != null
+                && fileEngine.isExperimentalFeatureEnabled(ExperimentalFeature.DIRECT_SEGMENT_STRING_READS);
         this.unfilteredSchema = unfilteredSchema;
         this.shardsVolatile = null;
     }
 
     public HollowObjectTypeReadState(HollowObjectSchema schema, HollowObjectTypeDataElements dataElements) {
         super(null, MemoryMode.ON_HEAP, schema);
         this.sampler = new HollowObjectSampler(schema, DisabledSamplingDirector.INSTANCE);
+        this.useDirectSegmentStringReads = false;
         this.unfilteredSchema = schema;
 
         HollowObjectTypeReadStateShard newShard = new HollowObjectTypeReadStateShard(schema, dataElements, 0);
@@ -382,7 +387,8 @@ public String readString(int ordinal, int fieldIndex) {
                 startByte = shardOrdinal != 0 ? shard.dataElements.fixedLengthData.getElementValue(currentBitOffset - shard.dataElements.bitsPerRecord, numBitsForField) : 0;
             } while(readWasUnsafe(shardsHolder, ordinal, shard));
 
-            result = shard.readString(startByte, endByte, numBitsForField, fieldIndex);
+            result = shard.readString(startByte, endByte, numBitsForField, fieldIndex,
+                    useDirectSegmentStringReads && shardsAreImmutable && memoryMode == MemoryMode.ON_HEAP);
         } while(readWasUnsafe(shardsHolder, ordinal, shard));
 
         return result;
@@ -413,7 +419,8 @@ public boolean isStringFieldEqual(int ordinal, int fieldIndex, String testValue)
                 startByte = shardOrdinal != 0 ? shard.dataElements.fixedLengthData.getElementValue(currentBitOffset - shard.dataElements.bitsPerRecord, numBitsForField) : 0;
             } while(readWasUnsafe(shardsHolder, ordinal, shard));
 
-            result = shard.isStringFieldEqual(startByte, endByte, numBitsForField, fieldIndex, testValue);
+            result = shard.isStringFieldEqual(startByte, endByte, numBitsForField, fieldIndex,
+                    testValue, useDirectSegmentStringReads && shardsAreImmutable && memoryMode == MemoryMode.ON_HEAP);
         } while(readWasUnsafe(shardsHolder, ordinal, shard));
 
         return result;