Fix test.

RumbleDB · Mar 5, 2024 · 52adf34 · 52adf34
1 parent 7696551
commit 52adf34
Show file tree

Hide file tree

Showing 13 changed files with 552 additions and 482 deletions.
diff --git a/src/main/java/org/rumbledb/context/DynamicContext.java b/src/main/java/org/rumbledb/context/DynamicContext.java
@@ -121,13 +121,21 @@ public VariableValues getVariableValues() {
     @Override
     public void write(Kryo kryo, Output output) {
         kryo.writeObjectOrNull(output, this.parent, DynamicContext.class);
+        kryo.writeObject(output, this.conf);
         kryo.writeObject(output, this.variableValues);
+        // kryo.writeObject(output, this.namedFunctions);
+        kryo.writeObject(output, this.inScopeSchemaTypes);
+        kryo.writeObject(output, this.currentDateTime.getMillis());
     }
 
     @Override
     public void read(Kryo kryo, Input input) {
         this.parent = kryo.readObjectOrNull(input, DynamicContext.class);
+        this.conf = kryo.readObject(input, RumbleRuntimeConfiguration.class);
         this.variableValues = kryo.readObject(input, VariableValues.class);
+        this.namedFunctions = new NamedFunctions(this.conf);
+        this.inScopeSchemaTypes= kryo.readObject(input, InScopeSchemaTypes.class);
+        this.currentDateTime = new DateTime(kryo.readObject(input, Long.class));
     }
 
     public enum VariableDependency {

diff --git a/src/main/java/org/rumbledb/context/RuntimeStaticContext.java b/src/main/java/org/rumbledb/context/RuntimeStaticContext.java
@@ -8,14 +8,26 @@
 import org.rumbledb.expressions.ExecutionMode;
 import org.rumbledb.types.SequenceType;
 
-public class RuntimeStaticContext implements Serializable {
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.KryoSerializable;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+public class RuntimeStaticContext implements Serializable, KryoSerializable {
     private static final long serialVersionUID = 1L;
 
     private RumbleRuntimeConfiguration configuration;
     private SequenceType staticType;
     private ExecutionMode executionMode;
     private ExceptionMetadata metadata;
 
+    public RuntimeStaticContext() {
+        this.configuration = null;
+        this.staticType = null;
+        this.executionMode = null;
+        this.metadata = null;
+    }
+
     public RuntimeStaticContext(
             RumbleRuntimeConfiguration configuration,
             SequenceType staticType,
@@ -62,4 +74,20 @@ public ExceptionMetadata getMetadata() {
         return this.metadata;
     }
 
+	@Override
+	public void write(Kryo kryo, Output output) {
+		kryo.writeObject(output, this.configuration);
+		kryo.writeObject(output, this.staticType);
+		kryo.writeObject(output, this.executionMode);
+		kryo.writeObject(output, this.metadata);
+	}
+
+	@Override
+	public void read(Kryo kryo, Input input) {
+		this.configuration = kryo.readObject(input, RumbleRuntimeConfiguration.class);
+		this.staticType = kryo.readObject(input, SequenceType.class);
+		this.executionMode = kryo.readObject(input, ExecutionMode.class);
+		this.metadata = kryo.readObject(input, ExceptionMetadata.class);
+	}
+
 }
diff --git a/src/main/java/org/rumbledb/exceptions/ExceptionMetadata.java b/src/main/java/org/rumbledb/exceptions/ExceptionMetadata.java
@@ -22,20 +22,36 @@
 
 import java.io.Serializable;
 
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.KryoSerializable;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
 /**
  * Metadata for error reporting (line and column number)
  *
  * @author Stefan Irimescu, Ghislain Fourny
  */
-public class ExceptionMetadata implements Serializable {
+public class ExceptionMetadata implements Serializable, KryoSerializable {
+
+    private String location;
+    private int tokenLineNumber;
+    private int tokenColumnNumber;
+    private String code;
 
     private static final long serialVersionUID = 1L;
-    private final String location;
-    private final int tokenLineNumber;
-    private final int tokenColumnNumber;
-    private final String code;
     public static final ExceptionMetadata EMPTY_METADATA = new ExceptionMetadata("none", 1, 0, "");
 
+    /**
+     * Builds a new empty metadata object (for serialization and deserialization only)
+     */
+    public ExceptionMetadata() {
+        this.location = "";
+        this.tokenLineNumber = -1;
+        this.tokenColumnNumber = -1;
+        this.code = "";
+    }
+
     /**
      * Builds a new metadata object
      *
@@ -118,4 +134,20 @@ public String toString() {
             + getTokenColumnNumber()
             + ":";
     }
+
+    @Override
+	public void write(Kryo kryo, Output output) {
+    	kryo.writeObject(output, this.location);
+    	kryo.writeObject(output, this.tokenLineNumber);
+    	kryo.writeObject(output, this.tokenColumnNumber);
+    	kryo.writeObject(output, this.code);
+    }
+
+    @Override
+    public void read(Kryo kryo, Input input) {
+    	this.location = kryo.readObject(input, String.class);
+    	this.tokenLineNumber= kryo.readObject(input, Integer.class);
+    	this.tokenColumnNumber = kryo.readObject(input, Integer.class);
+    	this.code = kryo.readObject(input, String.class);
+	}
 }
diff --git a/src/main/java/org/rumbledb/items/FunctionItem.java b/src/main/java/org/rumbledb/items/FunctionItem.java
@@ -57,10 +57,9 @@
 public class FunctionItem implements Item {
 
     private static final long serialVersionUID = 1L;
+
     private FunctionIdentifier identifier;
     private List<Name> parameterNames;
-
-    // signature contains type information for all parameters and the return value
     private FunctionSignature signature;
     private RuntimeIterator bodyIterator;
     private DynamicContext dynamicModuleContext;
@@ -242,20 +241,8 @@ public void write(Kryo kryo, Output output) {
                     "We do not support serializing DataFrames in function closures."
             );
         }
-        // kryo.writeObject(output, this.dynamicModuleContext);
-
-        // convert RuntimeIterator to byte[] data
-        /*
-         * try {
-         * byte[] data = SerializationUtils.serialize(this.bodyIterator);
-         * output.writeInt(data.length);
-         * output.writeBytes(data);
-         * } catch (Exception e) {
-         * throw new OurBadException(
-         * "Error converting functionItem-bodyRuntimeIterator to byte[]:" + e.getMessage()
-         * );
-         * }
-         */
+        kryo.writeObject(output, this.dynamicModuleContext);
+        kryo.writeClassAndObject(output, this.bodyIterator);
     }
 
     @SuppressWarnings("unchecked")
@@ -267,20 +254,8 @@ public void read(Kryo kryo, Input input) {
         this.localVariablesInClosure = kryo.readObject(input, HashMap.class);
         this.RDDVariablesInClosure = new HashMap<>();
         this.dataFrameVariablesInClosure = new HashMap<>();
-        // this.dynamicModuleContext = kryo.readObject(input, DynamicContext.class);
-        // this.bodyIterator = kryo.readObject(input, RuntimeIterator.class);
-
-        /*
-         * try {
-         * int dataLength = input.readInt();
-         * byte[] data = input.readBytes(dataLength);
-         * this.bodyIterator = SerializationUtils.deserialize(data);
-         * } catch (Exception e) {
-         * throw new OurBadException(
-         * "Error converting functionItem-bodyRuntimeIterator to functionItem:" + e.getMessage()
-         * );
-         * }
-         */
+        this.dynamicModuleContext = kryo.readObject(input, DynamicContext.class);
+        this.bodyIterator = (RuntimeIterator) kryo.readClassAndObject(input);
     }
 
     @Override

diff --git a/src/main/java/org/rumbledb/runtime/AtMostOneItemLocalRuntimeIterator.java b/src/main/java/org/rumbledb/runtime/AtMostOneItemLocalRuntimeIterator.java
@@ -33,6 +33,10 @@
 import org.rumbledb.expressions.comparison.ComparisonExpression.ComparisonOperator;
 import org.rumbledb.types.BuiltinTypesCatalogue;
 
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
 import sparksoniq.spark.SparkSessionManager;
 
 import org.rumbledb.runtime.misc.ComparisonIterator;
@@ -47,6 +51,11 @@ public abstract class AtMostOneItemLocalRuntimeIterator extends RuntimeIterator
     private static final long serialVersionUID = 1L;
     private Item result;
 
+    protected AtMostOneItemLocalRuntimeIterator() {
+        super();
+        this.result = null;
+    }
+
     protected AtMostOneItemLocalRuntimeIterator(
             List<RuntimeIterator> children,
             RuntimeStaticContext staticContext
@@ -186,4 +195,14 @@ public boolean getEffectiveBooleanValueOrCheckPosition(DynamicContext dynamicCon
                 getMetadata()
         );
     }
+
+    @Override
+	public void write(Kryo kryo, Output output) {
+    	super.write(kryo, output);
+    }
+
+    @Override
+    public void read(Kryo kryo, Input input) {
+    	super.read(kryo, input);
+    }
 }
diff --git a/src/main/java/org/rumbledb/runtime/HybridRuntimeIterator.java b/src/main/java/org/rumbledb/runtime/HybridRuntimeIterator.java
@@ -43,6 +43,12 @@ public abstract class HybridRuntimeIterator extends RuntimeIterator {
     protected List<Item> result = null;
     private int currentResultIndex = 0;
 
+    protected HybridRuntimeIterator() {
+        super();
+        this.result = null;
+        this.currentResultIndex = 0;
+    }
+
     protected HybridRuntimeIterator(
             List<RuntimeIterator> children,
             RuntimeStaticContext staticContext

diff --git a/src/main/java/org/rumbledb/runtime/RuntimeIterator.java b/src/main/java/org/rumbledb/runtime/RuntimeIterator.java
@@ -72,6 +72,15 @@ public abstract class RuntimeIterator implements RuntimeIteratorInterface, KryoS
     protected URI staticURI;
     // private StaticContext staticContext;
 
+    public RuntimeIterator() {
+    	this.hasNext = false;
+    	this.isOpen = false;
+    	this.children = null;
+    	this.currentDynamicContextForLocalExecution = null;
+        this.staticContext = null;
+        this.staticURI = null;
+    }
+
     protected RuntimeIterator(List<RuntimeIterator> children, RuntimeStaticContext staticContext) {
         this.staticContext = staticContext;
         if (this.staticContext.getStaticType() == null) {
@@ -220,6 +229,7 @@ public void reset(DynamicContext context) {
 
     @Override
     public void write(Kryo kryo, Output output) {
+        kryo.writeObject(output, this.staticContext);
         kryo.writeObject(output, this.children);
         // TODO serializer other fields
     }
@@ -230,6 +240,7 @@ public void read(Kryo kryo, Input input) {
         this.hasNext = false;
         this.isOpen = false;
         this.currentDynamicContextForLocalExecution = null;
+        this.staticContext = kryo.readObject(input, RuntimeStaticContext.class);
         this.children = kryo.readObject(input, ArrayList.class);
         // TODO serializer other fields
     }