Creation of RheemQuantum and the modification of Tuple and Record

rheem-basic/src/main/java/io/rheem/basic/data/Record.java

@@ -1,16 +1,18 @@
 package io.rheem.basic.data;
 
-import io.rheem.core.util.Copyable;
+import io.rheem.core.api.exception.RheemException;
 import io.rheem.core.util.ReflectionUtils;
 
-import java.io.Serializable;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutput;
+import java.io.ObjectOutputStream;
 import java.util.Arrays;
-import java.util.Objects;
 
 /**
  * A Type that represents a record with a schema, might be replaced with something standard like JPA entity.
  */
-public class Record implements Serializable, Copyable<Record> {
+public class Record implements RheemQuantum {
 
     private Object[] values;
 
@@ -28,19 +30,20 @@ public boolean equals(Object o) {
         if (this == o) return true;
         if (o == null || this.getClass() != o.getClass()) return false;
         Record record2 = (Record) o;
-        return Arrays.equals(this.values, record2.values);
+        return Arrays.deepEquals(this.values, record2.values);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(Arrays.hashCode(values));
+        return Arrays.deepHashCode(values);
     }
 
     @Override
     public String toString() {
-        return "Record" + Arrays.toString(this.values);
+        return "Record" + Arrays.deepToString(this.values);
     }
 
+    @Deprecated
     public Object getField(int index) {
         return this.values[index];
     }
@@ -51,6 +54,7 @@ public Object getField(int index) {
      * @param index the index of the field
      * @return the {@code double} representation of the field
      */
+    @Deprecated
     public double getDouble(int index) {
         Object field = this.values[index];
         return ReflectionUtils.toDouble(field);
@@ -62,6 +66,7 @@ public double getDouble(int index) {
      * @param index the index of the field
      * @return the {@code long} representation of the field
      */
+    @Deprecated
     public long getLong(int index) {
         Object field = this.values[index];
         if (field instanceof Integer) return (Integer) field;
@@ -77,6 +82,7 @@ public long getLong(int index) {
      * @param index the index of the field
      * @return the {@code int} representation of the field
      */
+    @Deprecated
     public int getInt(int index) {
         Object field = this.values[index];
         if (field instanceof Integer) return (Integer) field;
@@ -91,6 +97,7 @@ public int getInt(int index) {
      * @param index the index of the field
      * @return the field as a {@link String} (obtained via {@link Object#toString()}) or {@code null} if the field is {@code null}
      */
+    @Deprecated
     public String getString(int index) {
         Object field = this.values[index];
         return field == null ? null : field.toString();
@@ -101,8 +108,155 @@ public String getString(int index) {
      *
      * @return the number of fields in this instance
      */
+    @Deprecated
     public int size() {
+        return this.arity();
+    }
+
+
+    @Override
+    public int arity() {
         return this.values.length;
     }
 
+    @Override
+    public Object[] toArray() {
+        return this.values;
+    }
+
+    @Override
+    public RheemQuantum swap() {
+        Object[] tmp = new Object[this.arity()];
+
+        int end = this.arity() - 1;
+        for(int i = 0; i < this.arity(); i++){
+            tmp[end - i] = this.field(i);
+        }
+
+        return new Record(tmp);
+    }
+
+    @Override
+    public RheemQuantum project(int[] fields, boolean validate) {
+        //TODO add the validations
+        Object[] tmp = new Object[fields.length];
+
+        for(int i = 0; i < fields.length; i++){
+            tmp[i] = this.field(fields[i]);
+        }
+
+        return new Record(tmp);
+    }
+
+    @Override
+    public RheemQuantum join(RheemQuantum other) {
+        int final_size = this.arity() + other.arity();
+
+        Object[] elements = new Object[final_size];
+
+        System.arraycopy(this.toArray(), 0, elements, 0, this.arity());
+        System.arraycopy(other.toArray(), 0, elements, this.arity(), other.arity());
+
+        return new Record(elements);
+    }
+
+    @Override
+    public <T> T field(int index) {
+        return (T) this.values[index];
+    }
+
+    @Override
+    public String fieldAsString(int index) {
+        return this.field(index).toString();
+    }
+
+    @Override
+    public boolean fieldAsBoolean(int index) {
+        try {
+            return (boolean) this.field(index);
+        }catch (Exception e){
+            throw createExceptionCasting(e, index, "boolean");
+        }
+    }
+
+    @Override
+    public byte[] fieldAsBytes(int index) {
+        try{
+            return convertToBytes(this.fieldAsBytes(index));
+        }catch (Exception e){
+            throw createExceptionCasting(e, index, "byte[]");
+        }
+    }
+
+    @Override
+    public char fieldAsChar(int index) {
+        try {
+            return (char) this.field(index);
+        }catch (Exception e){
+            throw createExceptionCasting(e, index, "char");
+        }
+    }
+
+    @Override
+    public short fieldAsShort(int index) {
+        try {
+            return (short) this.field(index);
+        }catch (Exception e){
+            throw createExceptionCasting(e, index, "short");
+        }
+    }
+
+    @Override
+    public int fieldAsInt(int index) {
+        try {
+            return (int) this.field(index);
+        }catch (Exception e){
+            throw createExceptionCasting(e, index, "int");
+        }
+    }
+
+    @Override
+    public long fieldAsLong(int index) {
+        try {
+            return (long) this.field(index);
+        }catch (Exception e){
+            throw createExceptionCasting(e, index, "long");
+        }
+    }
+
+    @Override
+    public float fieldAsFloat(int index) {
+        try {
+            return (float) this.field(index);
+        }catch (Exception e){
+            throw createExceptionCasting(e, index, "float");
+        }
+    }
+
+    @Override
+    public double fieldAsDouble(int index) {
+        try {
+            return (double) this.field(index);
+        }catch (Exception e){
+            throw createExceptionCasting(e, index, "double");
+        }
+    }
+
+    protected RheemException createExceptionCasting(Exception cause, int index, String type){
+        return new RheemException(
+                String.format(
+                        "the field %d in the tuple is not castable to %s",
+                        index,
+                        type
+                ),
+                cause
+        );
+    }
+
+    protected byte[] convertToBytes(Object object) throws IOException {
+        ByteArrayOutputStream bos = new ByteArrayOutputStream();
+        ObjectOutput out = new ObjectOutputStream(bos);
+        out.writeObject(object);
+        return bos.toByteArray();
+    }
 }

rheem-basic/src/main/java/io/rheem/basic/data/RheemQuantum.java

@@ -0,0 +1,160 @@
+package io.rheem.basic.data;
+
+import io.rheem.core.util.Copyable;
+
+import java.io.Serializable;
+
+/**
+ * RheemQuantum is abstraction of the type use in data process and simplify
+ *  the manipulation
+ */
+public interface RheemQuantum extends Serializable, Copyable<RheemQuantum> {
+
+    /**
+     * Gets the number of field in the RheemQuantum.
+     *
+     * @return The number of fields in the RheemQuantum.
+     */
+    int arity();
+
+    /**
+     * create an object array with the content of the RheemQuantum
+     *
+     * @return object array
+     */
+    Object[] toArray();
+
+    /**
+     * @return a new instance with the fields of this instance swapped
+     */
+    RheemQuantum swap();
+
+    /**
+     * Creates a new RheemQuantum with projected fields and identical
+     *  from another RheemQuantum.
+     *
+     * <p>This method does not perform a deep copy.
+     *
+     * @param fields field indices to be projected
+     */
+    default RheemQuantum project(int[] fields){
+        return this.project(fields, false);
+    }
+
+    RheemQuantum project(int[] fields, boolean validate);
+
+    /**
+     * Creates a new RheemQuantum with fields that are copied from the other
+     *  RheemQuantum and appended to the resulting RheemQuantum in the given order.
+     *
+     * <p>This method does not perform a deep copy.
+     */
+    RheemQuantum join(RheemQuantum other);
+
+    /**
+     * Get type of the field in the position of the index
+     *
+     * @param index position of the required field
+     * @return Type of the field in the position of the index
+     */
+    default Class fieldType(int index){
+        return this.field(index).getClass();
+    }
+
+//    /**
+//     * Get field in the position of the index
+//     *
+//     * @param index position of the required field
+//     * @return field on the position of the index
+//     */
+//    Object field(int index);
+
+    /**
+     *
+     * Get field in the position of the index, adding the type
+     *
+     * @param index position of the required field
+     * @return field on the position of the index
+     */
+    <T> T field(int index);
+
+    /**
+     * Retrieve a field as a {@link String}.
+     *
+     * @param index the index of the field
+     * @return the field as a {@link String} (obtained via {@link Object#toString()})
+     *  or {@code null} if the field is {@code null}
+     */
+    String fieldAsString(int index);
+
+    /**
+     * Retrieve a field as a {@code boolean}. It must be castable as such.
+     *
+     * @param index the index of the field
+     * @return the {@code boolean} representation of the field
+     */
+    boolean fieldAsBoolean(int index);
+
+    /**
+     * Retrieve a field as a {@code byte[]}. It must be castable as such.
+     *
+     * @param index the index of the field
+     * @return the {@code byte[]} representation of the field
+     */
+    byte[] fieldAsBytes(int index);
+
+    /**
+     * Retrieve a field as a {@code char}. It must be castable as such.
+     *
+     * @param index the index of the field
+     * @return the {@code char} representation of the field
+     */
+    char fieldAsChar(int index);
+
+    /**
+     * Retrieve a field as a {@code short}. It must be castable as such.
+     *
+     * @param index the index of the field
+     * @return the {@code short} representation of the field
+     */
+    short fieldAsShort(int index);
+
+    /**
+     * Retrieve a field as a {@code int}. It must be castable as such.
+     *
+     * @param index the index of the field
+     * @return the {@code int} representation of the field
+     */
+    int fieldAsInt(int index);
+
+    /**
+     * Retrieve a field as a {@code long}. It must be castable as such.
+     *
+     * @param index the index of the field
+     * @return the {@code long} representation of the field
+     */
+    long fieldAsLong(int index);
+
+    /**
+     * Retrieve a field as a {@code float}. It must be castable as such.
+     *
+     * @param index the index of the field
+     * @return the {@code float} representation of the field
+     */
+    float fieldAsFloat(int index);
+
+    /**
+     * Retrieve a field as a {@code double}. It must be castable as such.
+     *
+     * @param index the index of the field
+     * @return the {@code double} representation of the field
+     */
+    double fieldAsDouble(int index);
+
+    //TODO validate if the date management is need it
+
+    default Object[] convertToArray(Object... objs){
+        return objs;
+    }
+
+}