[SYSTEMDS-3465] Typed return on CacheBlock Interface Slice

This task it to change the CacheBlock interface to return typed types on slice.
This makes the individual CacheBlock types independent and essentially make
FrameBlock not able to slice to MatrixBlock and vice versa.

Also changed in this commit is the memory estimates in the FrameBlock
that now use our memory estimates in utils/MemoryEstimates.java

Author: Baunsgaard

src/main/java/org/apache/sysds/hops/recompile/Recompiler.java

@@ -1610,8 +1610,8 @@ public static void executeInMemoryReblock(ExecutionContext ec, String varin, Str
 
 	@SuppressWarnings("unchecked")
 	public static void executeInMemoryReblock(ExecutionContext ec, String varin, String varout, LineageItem litem) {
-		CacheableData<CacheBlock> in = (CacheableData<CacheBlock>) ec.getCacheableData(varin);
-		CacheableData<CacheBlock> out = (CacheableData<CacheBlock>) ec.getCacheableData(varout);
+		CacheableData<CacheBlock<?>> in = (CacheableData<CacheBlock<?>>) ec.getCacheableData(varin);
+		CacheableData<CacheBlock<?>> out = (CacheableData<CacheBlock<?>>) ec.getCacheableData(varout);
 
 		if( in.isFederated() ) {
 			out.setMetaData(in.getMetaData());

src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java

@@ -59,7 +59,6 @@
 import org.apache.sysds.runtime.compress.lib.CLALibTSMM;
 import org.apache.sysds.runtime.compress.lib.CLALibUnary;
 import org.apache.sysds.runtime.compress.lib.CLALibUtils;
-import org.apache.sysds.runtime.controlprogram.caching.CacheBlock;
 import org.apache.sysds.runtime.controlprogram.caching.MatrixObject.UpdateType;
 import org.apache.sysds.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
 import org.apache.sysds.runtime.data.DenseBlock;
@@ -591,7 +590,7 @@ public void setOverlapping(boolean overlapping) {
 	}
 
 	@Override
-	public MatrixBlock slice(int rl, int ru, int cl, int cu, boolean deep, CacheBlock ret) {
+	public MatrixBlock slice(int rl, int ru, int cl, int cu, boolean deep, MatrixBlock ret) {
 		validateSliceArgument(rl, ru, cl, cu);
 		return CLALibSlice.slice(this, rl, ru, cl, cu, deep);
 	}

src/main/java/org/apache/sysds/runtime/controlprogram/caching/ByteBuffer.java

@@ -43,14 +43,14 @@ public class ByteBuffer
 	private final long _size;
 
 	protected byte[]     _bdata = null; //sparse matrix
-	protected CacheBlock _cdata = null; //dense matrix/frame
+	protected CacheBlock<?> _cdata = null; //dense matrix/frame
 
 	public ByteBuffer( long size ) {
 		_size = size;
 		_serialized = false;
 	}
 
-	public void serializeBlock( CacheBlock cb ) 
+	public void serializeBlock( CacheBlock<?> cb ) 
 		throws IOException
 	{	
 		_shallow = cb.isShallowSerialize(true);
@@ -85,10 +85,10 @@ public void serializeBlock( CacheBlock cb )
 		_serialized = true;
 	}
 
-	public CacheBlock deserializeBlock() 
+	public CacheBlock<?> deserializeBlock() 
 		throws IOException
 	{
-		CacheBlock ret = null;
+		CacheBlock<?> ret = null;
 
 		if( !_shallow ) { //sparse matrix / string frame
 			DataInput din = _matrix ? new CacheDataInput(_bdata) :
@@ -163,7 +163,7 @@ public void checkSerialized()
 	 * @param cb cache block
 	 * @return true if valid capacity
 	 */
-	public static boolean isValidCapacity( long size, CacheBlock cb )
+	public static boolean isValidCapacity( long size, CacheBlock<?> cb )
 	{
 		if( !cb.isShallowSerialize(true) ) { //SPARSE matrix blocks
 			// since cache blocks are serialized into a byte representation

src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheBlock.java

@@ -28,7 +28,7 @@
  * allows us to keep the buffer pool independent of matrix and frame blocks. 
  * 
  */
-public interface CacheBlock extends Writable 
+public interface CacheBlock <T> extends Writable 
 {
 
 	public int getNumRows();
@@ -96,7 +96,21 @@ public interface CacheBlock extends Writable
 	 * @param block cache block
 	 * @return sub-block of cache block
 	 */
-	public CacheBlock slice(int rl, int ru, int cl, int cu, CacheBlock block);
+	public CacheBlock<T> slice(int rl, int ru, int cl, int cu);
+	
+
+	/**
+	 * Slice a sub block out of the current block and write into the given output block.
+	 * This method returns the passed instance if not null.
+	 * 
+	 * @param rl row lower
+	 * @param ru row upper
+	 * @param cl column lower
+	 * @param cu column upper
+	 * @param block cache block
+	 * @return sub-block of cache block
+	 */
+	public CacheBlock<T> slice(int rl, int ru, int cl, int cu, T block);
 
 	/**
 	 * Slice a sub block out of the current block and write into the given output block.
@@ -110,7 +124,7 @@ public interface CacheBlock extends Writable
 	 * @param block cache block
 	 * @return sub-block of cache block
 	 */
-	public CacheBlock slice(int rl, int ru, int cl, int cu, boolean deep, CacheBlock block);
+	public CacheBlock<T> slice(int rl, int ru, int cl, int cu, boolean deep, T block);
 
 
 	/**
@@ -120,7 +134,7 @@ public interface CacheBlock extends Writable
 	 * @param that cache block
 	 * @param appendOnly ?
 	 */
-	public void merge(CacheBlock that, boolean appendOnly);
+	public void merge(T that, boolean appendOnly);
 
 	/**
 	 * Returns the double value at the passed row and column.
@@ -129,7 +143,7 @@ public interface CacheBlock extends Writable
 	 * @param c column of the value
 	 * @return double value at the passed row and column
 	 */
-	double getDouble(int r, int c);
+	public double getDouble(int r, int c);
 
 	/**
 	 * Returns the double value at the passed row and column.
@@ -138,7 +152,7 @@ public interface CacheBlock extends Writable
 	 * @param c column of the value
 	 * @return double value at the passed row and column
 	 */
-	double getDoubleNaN(int r, int c);
+	public double getDoubleNaN(int r, int c);
 
 	/**
 	 * Returns the string of the value at the passed row and column.
@@ -147,5 +161,5 @@ public interface CacheBlock extends Writable
 	 * @param c column of the value
 	 * @return string of the value at the passed row and column
 	 */
-	String getString(int r, int c);
+	public String getString(int r, int c);
 }

src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheBlockFactory.java

@@ -35,7 +35,7 @@
  */
 public class CacheBlockFactory 
 {
-	public static CacheBlock newInstance(int code) {
+	public static CacheBlock<?> newInstance(int code) {
 		switch( code ) {
 			case 0: return new MatrixBlock();
 			case 1: return new FrameBlock();
@@ -44,7 +44,17 @@ public static CacheBlock newInstance(int code) {
 		throw new RuntimeException("Unsupported cache block type: "+code);
 	}
 
-	public static int getCode(CacheBlock block) {
+	public static CacheBlock<?> newInstance(CacheBlock<?> block) {
+		if(block instanceof MatrixBlock)
+			return new MatrixBlock();
+		else if(block instanceof FrameBlock)
+			return new FrameBlock();
+		else if(block instanceof TensorBlock)
+			return new TensorBlock();
+		throw new RuntimeException("Unsupported cache block type: " + block.getClass().getName());
+	}
+
+	public static int getCode(CacheBlock<?> block) {
 		if (block instanceof MatrixBlock)
 			return 0;
 		else if (block instanceof FrameBlock)
@@ -54,7 +64,7 @@ else if (block instanceof TensorBlock)
 		throw new RuntimeException("Unsupported cache block type: " + block.getClass().getName());
 	}
 
-	public static ArrayList<?> getPairList(CacheBlock block) {
+	public static ArrayList<?> getPairList(CacheBlock<?> block) {
 		int code = getCode(block);
 		switch (code) {
 			case 0: return new ArrayList<Pair<MatrixIndexes, MatrixBlock>>();

src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheMaintenanceService.java

@@ -44,7 +44,7 @@ public void deleteFile(String fname) {
 			LocalFileUtils.deleteFileIfExists(fname, true);
 	}
 
-	public void serializeData(ByteBuffer bbuff, CacheBlock cb) {
+	public void serializeData(ByteBuffer bbuff, CacheBlock<?> cb) {
 		//sync or async file delete
 		if( CacheableData.CACHING_ASYNC_SERIALIZE )
 			_pool.submit(new CacheMaintenanceService.DataSerializerTask(bbuff, cb));
@@ -85,9 +85,9 @@ public void run() {
 
 	private static class DataSerializerTask implements Runnable {
 		private ByteBuffer _bbuff = null;
-		private CacheBlock _cb = null;
+		private CacheBlock<?> _cb = null;
 
-		public DataSerializerTask(ByteBuffer bbuff, CacheBlock cb) {
+		public DataSerializerTask(ByteBuffer bbuff, CacheBlock<?> cb) {
 			_bbuff = bbuff;
 			_cb = cb;
 		}

src/main/java/org/apache/sysds/runtime/controlprogram/caching/CacheableData.java

@@ -73,7 +73,7 @@
  * to allow Java garbage collection. If other parts of the system continue
  * keep references to the cache block, its eviction will not release any memory.
  */
-public abstract class CacheableData<T extends CacheBlock> extends Data
+public abstract class CacheableData<T extends CacheBlock<?>> extends Data
 {
 	private static final long serialVersionUID = -413810592207212835L;
 
@@ -1047,7 +1047,7 @@ protected boolean isBelowCachingThreshold() {
 		return (_data.getInMemorySize() <= CACHING_THRESHOLD);
 	}
 
-	public static boolean isBelowCachingThreshold(CacheBlock data) {
+	public static boolean isBelowCachingThreshold(CacheBlock<?> data) {
 		boolean ret;
 		if (OptimizerUtils.isUMMEnabled())
 			ret = UnifiedMemoryManager.getCacheBlockSize(data) <= CACHING_THRESHOLD;

src/main/java/org/apache/sysds/runtime/controlprogram/caching/LazyWriteBuffer.java

@@ -47,7 +47,7 @@ public enum RPolicy {
 	//maintenance service for synchronous or asynchronous delete of evicted files
 	private static CacheMaintenanceService _fClean;
 
-	public static int writeBlock(String fname, CacheBlock cb)
+	public static int writeBlock(String fname, CacheBlock<?> cb)
 		throws IOException
 	{
 		//obtain basic meta data of cache block
@@ -131,10 +131,10 @@ public static void deleteBlock(String fname)
 			_fClean.deleteFile(fname);
 	}
 
-	public static CacheBlock readBlock(String fname, boolean matrix)
+	public static CacheBlock<?> readBlock(String fname, boolean matrix)
 		throws IOException
 	{
-		CacheBlock cb = null;
+		CacheBlock<?> cb = null;
 		ByteBuffer ldata = null;
 
 		//probe write buffer
@@ -211,7 +211,7 @@ public static int getQueueSize() {
 		return _mQueue.size();
 	}
 
-	public static long getCacheBlockSize(CacheBlock cb) {
+	public static long getCacheBlockSize(CacheBlock<?> cb) {
 		return cb.isShallowSerialize() ?
 			cb.getInMemorySize() : cb.getExactSerializedSize();
 	}

src/main/java/org/apache/sysds/runtime/controlprogram/caching/UnifiedMemoryManager.java

@@ -253,10 +253,10 @@ public static long getUMMFree() {
 	}
 
 	// Reads a cached object. This is called from cacheabledata implementations
-	public static CacheBlock readBlock(String fname, boolean matrix)
+	public static CacheBlock<?> readBlock(String fname, boolean matrix)
 		throws IOException
 	{
-		CacheBlock cb = null;
+		CacheBlock<?> cb = null;
 		ByteBuffer ldata = null;
 
 		//probe write buffer
@@ -336,7 +336,7 @@ public static int makeSpace(long reqSpace) {
 	}
 
 	// Write an object to the cache
-	public static int writeBlock(String fname, CacheBlock cb)
+	public static int writeBlock(String fname, CacheBlock<?> cb)
 		throws IOException
 	{
 		//obtain basic metadata of the cache block
@@ -380,7 +380,7 @@ public static int writeBlock(String fname, CacheBlock cb)
 		return numEvicted;
 	}
 
-	public static long getCacheBlockSize(CacheBlock cb) {
+	public static long getCacheBlockSize(CacheBlock<?> cb) {
 		return cb.isShallowSerialize() ?
 			cb.getInMemorySize() : cb.getExactSerializedSize();
 	}

src/main/java/org/apache/sysds/runtime/controlprogram/context/ExecutionContext.java

@@ -628,7 +628,7 @@ public void setFrameOutput(String varName, FrameBlock outputData) {
 		setVariable(varName, fo);
 	}
 
-	public static CacheableData<?> createCacheableData(CacheBlock cb) {
+	public static CacheableData<?> createCacheableData(CacheBlock<?> cb) {
 		if( cb instanceof MatrixBlock )
 			return createMatrixObject((MatrixBlock) cb);
 		else if( cb instanceof FrameBlock )