WIP: delta transitions in shared-memory mode

hollow-diff-ui/src/tools/java/com/netflix/hollow/diff/ui/HistoryUITest.java

@@ -229,7 +229,7 @@ private void exploreOrdinals(HollowReadStateEngine readStateEngine) {
         System.out.println("CUSTOM_VERSION_TAG= " + readStateEngine.getHeaderTags().get(CUSTOM_VERSION_TAG));
         for (HollowTypeReadState typeReadState : readStateEngine.getTypeStates()) {
             BitSet populatedOrdinals = typeReadState.getPopulatedOrdinals();
-            System.out.println("SNAP: PopulatedOrdinals= " + populatedOrdinals);
+            // System.out.println("SNAP: PopulatedOrdinals= " + populatedOrdinals);
             int ordinal = populatedOrdinals.nextSetBit(0);
             while (ordinal != -1) {
                 HollowObjectTypeReadState o = (HollowObjectTypeReadState) typeReadState;

hollow/src/main/java/com/netflix/hollow/api/client/HollowDataHolder.java

@@ -206,16 +206,17 @@ private void applyDeltaOnlyPlan(HollowUpdatePlan updatePlan, HollowConsumer.Refr
     }
 
     private void applyDeltaTransition(HollowConsumer.Blob blob, boolean isSnapshotPlan, HollowConsumer.RefreshListener[] refreshListeners) throws Throwable {
-        if (!memoryMode.equals(MemoryMode.ON_HEAP)) {
-            LOG.warning("Skipping delta transition in shared-memory mode");
-            return;
-        }
+        LOG.info(String.format("Attempting delta transition from v %s to v %s in %s mode",
+                blob.getFromVersion(), blob.getToVersion(), memoryMode));
 
         try (HollowBlobInput in = HollowBlobInput.modeBasedSelector(memoryMode, blob);
              OptionalBlobPartInput optionalPartIn = blob.getOptionalBlobPartInputs()) {
             applyStateEngineTransition(in, optionalPartIn, blob, refreshListeners);
 
             if(objLongevityConfig.enableLongLivedObjectSupport()) {
+                if (!memoryMode.equals(MemoryMode.ON_HEAP)) {
+                    throw new UnsupportedOperationException("Shared memory mode doesn't support object longevity... yet");
+                }
                 HollowDataAccess previousDataAccess = currentAPI.getDataAccess();
                 HollowHistoricalStateDataAccess priorState = new HollowHistoricalStateCreator(null).createBasedOnNewDelta(currentVersion, stateEngine);
                 HollowProxyDataAccess newDataAccess = new HollowProxyDataAccess();
@@ -245,7 +246,11 @@ private void applyDeltaTransition(HollowConsumer.Blob blob, boolean isSnapshotPl
 
         } catch(Throwable t) {
             failedTransitionTracker.markFailedTransition(blob);
+            LOG.warning("SNAP: Delta transition encountered exception: " + t);
             throw t;
+        } finally {
+            LOG.info(String.format("Delta transition completed from v %s to v %s in %s mode",
+                    blob.getFromVersion(), blob.getToVersion(), memoryMode));
         }
     }
 

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

@@ -18,30 +18,64 @@
 
 import com.netflix.hollow.core.memory.encoding.BlobByteBuffer;
 import com.netflix.hollow.core.read.HollowBlobInput;
+import java.io.File;
 import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.util.logging.Logger;
 
 /**
- * {@code BlobByteBuffer} based implementation of variable length byte data that only supports read.
+ * {@code BlobByteBuffer} based implementation of variable length byte data that only supports read.    // TODO: update when supports write
  */
 public class EncodedByteBuffer implements VariableLengthData {
+    private static final Logger LOG = Logger.getLogger(EncodedByteBuffer.class.getName());
 
     private BlobByteBuffer bufferView;
     private long size;
 
-    public EncodedByteBuffer() {
+    private final File managedFile;
+    private boolean destroyActionHasBeenTakenBeforeDiag = false;
+
+    public EncodedByteBuffer(File managedFile) {
+        this.managedFile = managedFile;
         this.size = 0;
     }
 
+    public BlobByteBuffer getBufferView() {
+        return bufferView;
+    }
+
+    public void destroy() throws IOException {
+        if (bufferView != null) {
+            bufferView.unmapBlob();
+            destroyActionHasBeenTakenBeforeDiag = true;
+        } else {
+            if (destroyActionHasBeenTakenBeforeDiag) {
+                LOG.warning("SNAP: destroy() called on EncodedByteBuffer thats already been destroyed previously");
+            }
+        }
+        bufferView = null;
+        if (managedFile != null) {
+            // LOG.info("SNAP: EncodedByteBuffer destroy() is also deleting staged file " + managedFile.getAbsolutePath());
+            Files.delete(managedFile.toPath());
+        }
+    }
+
     @Override
     public byte get(long index) {
-        if (index >= this.size) {
-            throw new IllegalStateException();
+        if (index >= this.size) {   // SNAP: TODO: realized in transformer
+            if (index >= this.size + Long.BYTES) {
+                LOG.warning(String.format("SNAP: unexpected get from EncodedByteBuffer: index=%s, size=%s", index, size));
+            }
         }
-
         byte retVal = this.bufferView.getByte(this.bufferView.position() + index);
         return retVal;
     }
 
+    public int getBytes(long index, long len, byte[] bytes) {
+        return this.bufferView.getBytes(this.bufferView.position() + index, len, bytes, true);
+    }
+
     /**
      * {@inheritDoc}
      * This is achieved by initializing a {@code BlobByteBuffer} that is a view on the underlying {@code BlobByteBuffer}
@@ -59,16 +93,34 @@ public void loadFrom(HollowBlobInput in, long length) throws IOException {
 
     @Override
     public void copy(ByteData src, long srcPos, long destPos, long length) {
-        throw new UnsupportedOperationException("Operation not supported in shared-memory mode");
+        throw new UnsupportedOperationException("Operation not supported in shared-memory mode - EncodedByteBuffers are read-only");
     }
 
     @Override
-    public void orderedCopy(VariableLengthData src, long srcPos, long destPos, long length) {
-        throw new UnsupportedOperationException("Operation not supported in shared-memory mode");
+    public void orderedCopy(VariableLengthData src, long srcPos, long destPos, long length) throws IOException {
+        throw new UnsupportedOperationException("Underlying data can only be mutated using " + VariableLengthDataFactory.StagedVariableLengthData.class.getName());
     }
 
     @Override
     public long size() {
         return size;
     }
+
+    @Override
+    public void set(long index, byte value) {
+        throw new UnsupportedOperationException("Operation not supported in shared-memory mode");
+    }
+
+    /**
+     * Write a portion of this data to an OutputStream.
+     *
+     * @param os the output stream to write to
+     * @param startPosition the position to begin copying from this array
+     * @param len the length of the data to copy
+     * @throws IOException if the write to the output stream could not be performed
+     */
+    @Override
+    public void writeTo(OutputStream os, long startPosition, long len) throws IOException {
+        throw new UnsupportedOperationException("Not supported for shared memory mode, supports the type filter feature");
+    }
 }

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

@@ -1,10 +1,15 @@
 package com.netflix.hollow.core.memory;
 
+import static com.netflix.hollow.core.memory.encoding.BlobByteBuffer.MAX_SINGLE_BUFFER_CAPACITY;
+
 import com.netflix.hollow.core.memory.encoding.EncodedLongBuffer;
 import com.netflix.hollow.core.memory.encoding.FixedLengthElementArray;
+import com.netflix.hollow.core.memory.encoding.VarInt;
 import com.netflix.hollow.core.memory.pool.ArraySegmentRecycler;
 import com.netflix.hollow.core.read.HollowBlobInput;
+import java.io.File;
 import java.io.IOException;
+import java.io.RandomAccessFile;
 import java.util.logging.Logger;
 
 public class FixedLengthDataFactory {
@@ -22,11 +27,46 @@ public static FixedLengthData get(HollowBlobInput in, MemoryMode memoryMode, Arr
         }
     }
 
-    public static void destroy(FixedLengthData fld, ArraySegmentRecycler memoryRecycler) {
+    // allocate (for write) // unused
+    public static FixedLengthData allocate(HollowBlobInput in,
+                                           MemoryMode memoryMode, ArraySegmentRecycler memoryRecycler,
+                                           String fileName) throws IOException {
+
+        long numLongs = VarInt.readVLong(in);
+        long numBits = numLongs << 6;
+        return allocate(numBits, memoryMode, memoryRecycler, fileName);
+    }
+
+    public static FixedLengthData allocate(long numBits, MemoryMode memoryMode, ArraySegmentRecycler memoryRecycler,
+                                         String fileName) throws IOException {
+        if (memoryMode.equals(MemoryMode.ON_HEAP)) {
+            return new FixedLengthElementArray(memoryRecycler, numBits);
+        } else {
+            long numLongs = ((numBits - 1) >>> 6) + 1;
+            long numBytes = numLongs << 3;
+            // add Long.BYTES to provisioned file size to accommodate unaligned read starting offset in last long
+            File targetFile = provisionTargetFile(numBytes + Long.BYTES, fileName);
+            try (HollowBlobInput targetBlob = HollowBlobInput.randomAccess(targetFile, MAX_SINGLE_BUFFER_CAPACITY)) {
+                return EncodedLongBuffer.newFrom(targetBlob, numLongs, targetFile);   // TODO: test with different single buffer capacities
+            }
+        }
+    }
+
+    static File provisionTargetFile(long numBytes, String fileName) throws IOException {
+        File targetFile = new File(fileName);
+        RandomAccessFile raf = new RandomAccessFile(targetFile, "rw");
+        raf.setLength(numBytes);
+        raf.close();
+        // System.out.println("SNAP: Provisioned targetFile (one per shard per type) of size " + numBytes + " bytes: " + targetFile.getPath());
+        return targetFile;
+    }
+
+    public static void destroy(FixedLengthData fld, ArraySegmentRecycler memoryRecycler) throws IOException {
         if (fld instanceof FixedLengthElementArray) {
             ((FixedLengthElementArray) fld).destroy(memoryRecycler);
         } else if (fld instanceof EncodedLongBuffer) {
-            LOG.warning("Destroy operation is a no-op in shared memory mode");
+            // LOG.info("SNAP: Destroy operation invoked on EncodedLongBuffer (FixedLengthData)");
+            ((EncodedLongBuffer) fld).destroy();
         } else {
             throw new UnsupportedOperationException("Unknown type");
         }

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

@@ -0,0 +1,31 @@
+package com.netflix.hollow.core.memory;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Random;
+
+public class MemoryFileUtil {
+
+    public static String filepath() {
+        return System.getProperty("java.io.tmpdir");
+    }
+
+    // whichData is null for object types, and listPointerData, listElementData, mapEntryData etc. for collection types
+    public static String fixedLengthDataFilename(String type, String whichData, int shardNo) {
+        return "hollow-fixedLengthData-"
+                + (whichData != null ? whichData + "_" : "")
+                + type + "_"
+                + shardNo + "_"
+                + LocalDateTime.now().format(DateTimeFormatter.ofPattern("MM-dd HH:mm")) + "_"
+                + new Random().nextInt();
+    }
+
+    public static String varLengthDataFilename(String type, String field, int shardNo) {
+        return "hollow-varLengthData-"
+                + field + "_"
+                + type + "_"
+                + shardNo + "_"
+                + LocalDateTime.now().format(DateTimeFormatter.ofPattern("MM-dd HH:mm")) + "_"
+                + new Random().nextInt();
+    }
+}

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

@@ -61,6 +61,7 @@ public SegmentedByteArray(ArraySegmentRecycler memoryRecycler) {
      * @param index the index
      * @param value the byte value
      */
+    @Override
     public void set(long index, byte value) {
         int segmentIndex = (int)(index >> log2OfSegmentSize);
         ensureCapacity(segmentIndex);
@@ -236,6 +237,7 @@ public void loadFrom(HollowBlobInput is, long length) throws IOException {
      * @param len the length of the data to copy
      * @throws IOException if the write to the output stream could not be performed
      */
+    @Override
     public void writeTo(OutputStream os, long startPosition, long len) throws IOException {
         int segmentSize = 1 << log2OfSegmentSize;
         int remainingBytesInSegment = segmentSize - (int)(startPosition & bitmask);

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

@@ -2,6 +2,7 @@
 
 import com.netflix.hollow.core.read.HollowBlobInput;
 import java.io.IOException;
+import java.io.OutputStream;
 
 /**
  * Conceptually this can be thought of as a single byte array or buffer of undefined length.  It will grow automatically
@@ -38,11 +39,15 @@ public interface VariableLengthData extends ByteData {
      * @param destPos position in destination to begin copying to
      * @param length length of data to copy in bytes
      */
-    void orderedCopy(VariableLengthData src, long srcPos, long destPos, long length);
+    void orderedCopy(VariableLengthData src, long srcPos, long destPos, long length) throws IOException;
 
     /**
      * Data size in bytes
      * @return size in bytes
      */
     long size();
+
+    void set(long index, byte value);
+
+    void writeTo(OutputStream os, long startPosition, long len) throws IOException;
 }