[SYSTEMDS-3484] Frame Append Optimizations

src/main/java/org/apache/sysds/common/Types.java

@@ -117,9 +117,81 @@ public static ValueType fromExternalString(String value) {
 					throw new DMLRuntimeException("Unknown value type: "+value);
 			}
 		}
+
 		public static boolean isSameTypeString(ValueType vt1, ValueType vt2) {
 			return vt1.toExternalString().equals(vt2.toExternalString());
 		}
+
+		/**
+		 * Get the highest common type that both ValueTypes can be contained in
+		 * 
+		 * @param a First ValueType
+		 * @param b Second ValueType
+		 * @return The common higest type to represent both
+		 */
+		public static ValueType getHighestCommonType(ValueType a, ValueType b){
+			if(a == b)
+				return a;
+			if(b == UNKNOWN)
+				throw new DMLRuntimeException(
+					"Invalid or not implemented support for comparing valueType of: " + a + " and " + b);
+
+			switch(a){
+				case STRING:
+					return a;
+				case FP64:
+					switch(b){
+						case STRING:
+							return b;
+						default:
+							return a;
+					}
+				case FP32:
+				switch(b){
+					case STRING:
+					case FP64:
+						return b;
+					default:
+						return a;
+				}
+				case INT64:
+				switch(b){
+					case STRING:
+					case FP64:
+					case FP32:
+						return b;
+					default:
+						return a;
+				}
+				case INT32:
+				switch(b){
+					case STRING:
+					case FP64:
+					case FP32:
+					case INT64:
+						return b;
+					default:
+						return a;
+				}
+				case UINT8:
+				switch(b){
+					case STRING:
+					case FP64:
+					case FP32:
+					case INT64:
+					case INT32:
+						return b;
+					default:
+						return a;
+				}
+				case BOOLEAN:
+					return b; // always higher type in b;
+				case UNKNOWN:
+				default:
+					throw new DMLRuntimeException(
+						"Invalid or not implemented support for comparing valueType of: " + a + " and " + b);
+			}
+		}
 	}
 
 	/**

src/main/java/org/apache/sysds/runtime/frame/data/FrameBlock.java

@@ -31,8 +31,6 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;
@@ -1034,56 +1032,6 @@ public void slice(ArrayList<Pair<Long, FrameBlock>> outList, IndexRange range, i
 	 */
 	public FrameBlock append(FrameBlock that, boolean cbind) {
 		return FrameLibAppend.append(this, that, cbind);
-
-		// FrameBlock ret = new FrameBlock();
-		// if(cbind) // COLUMN APPEND
-		// {
-		// 	// sanity check row dimension mismatch
-		// 	if(getNumRows() != that.getNumRows()) {
-		// 		throw new DMLRuntimeException(
-		// 			"Incompatible number of rows for cbind: " + that.getNumRows() + " (expected: " + getNumRows() + ")");
-		// 	}
-
-		// 	// allocate output frame
-		// 	ret._numRows = _numRows;
-
-		// 	// concatenate schemas (w/ deep copy to prevent side effects)
-		// 	ret._schema = (ValueType[]) ArrayUtils.addAll(_schema, that._schema);
-		// 	ret._colnames = (String[]) ArrayUtils.addAll(getColumnNames(), that.getColumnNames());
-		// 	ret._colmeta = (ColumnMetadata[]) ArrayUtils.addAll(_colmeta, that._colmeta);
-
-		// 	// check and enforce unique columns names
-		// 	if(!Arrays.stream(ret._colnames).allMatch(new HashSet<>()::add))
-		// 		ret._colnames = createColNames(ret.getNumColumns());
-
-		// 	// concatenate column data (w/ shallow copy which is safe due to copy on write semantics)
-		// 	ret._coldata = (Array[]) ArrayUtils.addAll(_coldata, that._coldata);
-		// }
-		// else // ROW APPEND
-		// {
-		// 	// sanity check column dimension mismatch
-		// 	if(getNumColumns() != that.getNumColumns()) {
-		// 		throw new DMLRuntimeException("Incompatible number of columns for rbind: " + that.getNumColumns()
-		// 			+ " (expected: " + getNumColumns() + ")");
-		// 	}
-		// 	ret._numRows = _numRows; // note set to previous since each row is appended on.
-		// 	ret._schema = _schema.clone();
-		// 	ret._colnames = (_colnames != null) ? _colnames.clone() : null;
-		// 	ret._colmeta = new ColumnMetadata[getNumColumns()];
-		// 	for(int j = 0; j < _schema.length; j++)
-		// 		ret._colmeta[j] = new ColumnMetadata();
-
-		// 	// concatenate data (deep copy first, append second)
-		// 	ret._coldata = new Array[getNumColumns()];
-		// 	for(int j = 0; j < getNumColumns(); j++)
-		// 		ret._coldata[j] = _coldata[j].clone();
-		// 	Iterator<Object[]> iter = IteratorFactory.getObjectRowIterator(that, _schema);
-		// 	while(iter.hasNext())
-		// 		ret.appendRow(iter.next());
-		// }
-
-		// ret._msize = -1;
-		// return ret;
 	}
 
 	public FrameBlock copy() {

src/main/java/org/apache/sysds/runtime/frame/data/columns/Array.java

@@ -173,7 +173,7 @@ public final void setNz(Array<T> value) {
 	 * @param value array of other type
 	 */
 	public final void setFromOtherTypeNz(Array<?> value) {
-		setFromOtherTypeNz(0, value.size(), value);
+		setFromOtherTypeNz(0, value.size()-1, value);
 	}
 
 	/**
@@ -201,6 +201,17 @@ public final void setFromOtherTypeNz(Array<?> value) {
 	 */
 	public abstract void append(T value);
 
+	/**
+	 * append other array, if the other array is fitting in current allocated size use that allocated size, otherwise
+	 * allocate new array to combine the other with this.
+	 * 
+	 * This method should use the set range function, and should be preferred over the append single values.
+	 * 
+	 * @param other The other array of same type to append to this.
+	 * @return The combined arrays.
+	 */
+	public abstract Array<T> append(Array<T> other);
+
 	/**
 	 * Slice out the sub range and return new array with the specified type.
 	 * 

src/main/java/org/apache/sysds/runtime/frame/data/columns/ArrayFactory.java

@@ -88,10 +88,10 @@ public static long getInMemorySize(ValueType type, int _numRows) {
 		}
 	}
 
-	public static Array<?> allocate(ValueType v, int nRow, String val){
+	public static Array<?> allocate(ValueType v, int nRow, String val) {
 		Array<?> a = allocate(v, nRow);
 		a.fill(val);
-		return a; 
+		return a;
 	}
 
 	public static Array<?> allocate(ValueType v, int nRow) {
@@ -146,4 +146,27 @@ public static Array<?> read(DataInput in, int nRow) throws IOException {
 		arr.readFields(in);
 		return arr;
 	}
+
+	/**
+	 * append arrays to each other, and cast to highest common type if different types.
+	 * 
+	 * @param <C> The type to return, java automatically make this Object, and this is fine.
+	 * @param a   The first array to append to (potentially modifying this a if applicable)
+	 * @param b   THe array to append to a, (not getting modified).
+	 * @return A array containing the concatenation of the two.
+	 */
+	@SuppressWarnings("unchecked")
+	public static <C> Array<C> append(Array<?> a, Array<?> b) {
+
+		// get common highest datatype.
+		final ValueType ta = a.getValueType();
+		final ValueType tb = b.getValueType();
+		final ValueType tc = ValueType.getHighestCommonType(ta, tb);
+
+		Array<C> ac = (Array<C>) (ta != tc ? a.changeType(tc) : a);
+		Array<C> bc = (Array<C>) (tb != tc ? b.changeType(tc) : b);
+
+		return ac.append(bc);
+	}
+
 }

src/main/java/org/apache/sysds/runtime/frame/data/columns/BitSetArray.java

@@ -236,6 +236,15 @@ public void append(String value) {
 		append(BooleanArray.parseBoolean(value));
 	}
 
+	@Override
+	public BitSetArray append(Array<Boolean> other) {
+		final int endSize = this._size + other.size();
+		final BitSetArray retBS = new BitSetArray(endSize);
+		retBS.set(0, this._size - 1, this, 0);
+		retBS.set(this._size, endSize - 1, other,0);
+		return retBS;
+	}
+
 	@Override
 	public void append(Boolean value) {
 		if(_data.length * 64 < _size + 1)
@@ -447,8 +456,8 @@ public String toString() {
 	}
 
 	@Override
-	public double getAsDouble(int i){
-		return get(i) ? 1.0: 0.0;
+	public double getAsDouble(int i) {
+		return get(i) ? 1.0 : 0.0;
 	}
 
 	public static String longToBits(long l) {

src/main/java/org/apache/sysds/runtime/frame/data/columns/BooleanArray.java

@@ -34,6 +34,11 @@
 public class BooleanArray extends Array<Boolean> {
 	protected boolean[] _data;
 
+	public BooleanArray(int size) {
+		super(size);
+		_data = new boolean[size];
+	}
+
 	public BooleanArray(boolean[] data) {
 		super(data.length);
 		_data = data;
@@ -118,6 +123,23 @@ public void append(Boolean value) {
 		_data[_size++] = (value != null) ? value : false;
 	}
 
+	@Override
+	public Array<Boolean> append(Array<Boolean> other) {
+		final int endSize = this._size + other.size();
+		if(other instanceof BooleanArray && endSize < ArrayFactory.bitSetSwitchPoint) {
+			final boolean[] ret = new boolean[endSize];
+			System.arraycopy(_data, 0, ret, 0, this._size);
+			System.arraycopy((boolean[]) other.get(), 0, ret, this._size, other.size());
+			return new BooleanArray(ret);
+		}
+		else {
+			final BooleanArray retBS = new BooleanArray(endSize);
+			retBS.set(0, this._size - 1, this, 0);
+			retBS.set(this._size, endSize - 1, other, 0);
+			return retBS;
+		}
+	}
+
 	@Override
 	public void write(DataOutput out) throws IOException {
 		out.writeByte(FrameArrayType.BOOLEAN.ordinal());
@@ -144,8 +166,11 @@ public Array<Boolean> slice(int rl, int ru) {
 
 	@Override
 	public void reset(int size) {
-		if(_data.length < size)
+		if(_data.length < size || _data.length > 2 * size)
 			_data = new boolean[size];
+		else
+			for(int i = 0; i < size; i++)
+				_data[i] = false;
 		_size = size;
 	}
 
@@ -249,7 +274,7 @@ public void fill(String value) {
 	public void fill(Boolean value) {
 		for(int i = 0; i < _size; i++)
 			_data[i] = value;
-		
+
 	}
 
 	@Override
@@ -264,8 +289,8 @@ public String toString() {
 	}
 
 	@Override
-	public double getAsDouble(int i){
-		return _data[i] ? 1.0: 0.0;
+	public double getAsDouble(int i) {
+		return _data[i] ? 1.0 : 0.0;
 	}
 
 	protected static boolean parseBoolean(String value) {

src/main/java/org/apache/sysds/runtime/frame/data/columns/DoubleArray.java

@@ -108,6 +108,15 @@ public void append(Double value) {
 		_data[_size++] = (value != null) ? value : 0d;
 	}
 
+	@Override
+	public DoubleArray append(Array<Double> other) {
+		final int endSize = this._size + other.size();
+		final double[] ret = new double[endSize];
+		System.arraycopy(_data, 0, ret, 0, this._size);
+		System.arraycopy((double[])other.get(), 0, ret, this._size, other.size());
+		return new DoubleArray(ret);
+	}
+
 	@Override
 	public void write(DataOutput out) throws IOException {
 		out.writeByte(FrameArrayType.FP64.ordinal());
@@ -134,8 +143,11 @@ public Array<Double> slice(int rl, int ru) {
 
 	@Override
 	public void reset(int size) {
-		if(_data.length < size)
+		if(_data.length < size || _data.length > 2 * size)
 			_data = new double[size];
+		else
+			for(int i = 0; i < size; i++)
+				_data[i] = 0;
 		_size = size;
 	}
 

src/main/java/org/apache/sysds/runtime/frame/data/columns/FloatArray.java

@@ -107,6 +107,16 @@ public void append(Float value) {
 		_data[_size++] = (value != null) ? value : 0f;
 	}
 
+	@Override
+	public FloatArray append(Array<Float> other) {
+		final int endSize = this._size + other.size();
+		final float[] ret = new float[endSize];
+		System.arraycopy(_data, 0, ret, 0, this._size);
+
+		System.arraycopy((float[])other.get(), 0, ret, this._size, other.size());
+		return new FloatArray(ret);
+	}
+
 	@Override
 	public void write(DataOutput out) throws IOException {
 		out.writeByte(FrameArrayType.FP32.ordinal());
@@ -133,8 +143,11 @@ public Array<Float> slice(int rl, int ru) {
 
 	@Override
 	public void reset(int size) {
-		if(_data.length < size)
+		if(_data.length < size || _data.length > 2 * size)
 			_data = new float[size];
+			else
+			for(int i = 0; i < size; i++)
+				_data[i] = 0;
 		_size = size;
 	}
 

src/main/java/org/apache/sysds/runtime/frame/data/columns/IntegerArray.java

@@ -108,6 +108,15 @@ public void append(Integer value) {
 		_data[_size++] = (value != null) ? value : 0;
 	}
 
+	@Override
+	public IntegerArray append(Array<Integer> other) {
+		final int endSize = this._size + other.size();
+		final int[] ret = new int[endSize];
+		System.arraycopy(_data, 0, ret, 0, this._size);
+		System.arraycopy((int[])other.get(), 0, ret, this._size, other.size());
+		return new IntegerArray(ret);
+	}
+
 	@Override
 	public void write(DataOutput out) throws IOException {
 		out.writeByte(FrameArrayType.INT32.ordinal());
@@ -133,8 +142,11 @@ public Array<Integer> slice(int rl, int ru) {
 	}
 
 	public void reset(int size) {
-		if(_data.length < size)
+		if(_data.length < size || _data.length > 2 * size)
 			_data = new int[size];
+			else
+			for(int i = 0; i < size; i++)
+				_data[i] = 0;
 		_size = size;
 	}