Native DF execution for more types

pom.xml

@@ -41,7 +41,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
-                <version>3.8.1</version>
+                <version>3.11.0</version>
                 <!-- Holy grail of eclipse configuration:  https://stackoverflow.com/a/49599478/10707488   -->
                 <configuration>
                     <compilerId>eclipse</compilerId>
@@ -88,7 +88,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-assembly-plugin</artifactId>
-                <version>3.1.1</version><!--$NO-MVN-MAN-VER$-->
+                <version>3.6.0</version><!--$NO-MVN-MAN-VER$-->
                 <configuration>
                     <archive>
                         <manifest>
@@ -178,7 +178,7 @@
             <plugin>
                 <groupId>com.diffplug.spotless</groupId>
                 <artifactId>spotless-maven-plugin</artifactId>
-                <version>1.26.0</version>
+                <version>2.36.0</version>
                 <executions>
                     <execution>
                         <phase>verify</phase>

src/main/java/org/rumbledb/api/SequenceOfItems.java

@@ -2,11 +2,16 @@
 
 import java.util.List;
 
+import org.apache.spark.SparkRuntimeException;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.rumbledb.config.RumbleRuntimeConfiguration;
 import org.rumbledb.context.DynamicContext;
+import org.rumbledb.exceptions.CastException;
+import org.rumbledb.exceptions.ExceptionMetadata;
+import org.rumbledb.exceptions.RumbleException;
+import org.rumbledb.exceptions.UnexpectedTypeException;
 import org.rumbledb.items.ItemFactory;
 import org.rumbledb.runtime.RuntimeIterator;
 
@@ -87,7 +92,25 @@ public boolean hasNext() {
         if (!this.isMaterialisable()) {
             return false;
         }
-        return this.iterator.hasNext();
+        try {
+            return this.iterator.hasNext();
+        } catch (NumberFormatException e) {
+            RumbleException ex = new CastException(e.getMessage(), ExceptionMetadata.EMPTY_METADATA);
+            ex.initCause(e);
+            throw ex;
+        } catch (SparkRuntimeException e) {
+            if (e.getMessage().contains("CAST_INVALID_INPUT")) {
+                RumbleException ex = new CastException(e.getMessage(), ExceptionMetadata.EMPTY_METADATA);
+                ex.initCause(e);
+                throw ex;
+            } else {
+                throw e;
+            }
+        } catch (UnsupportedOperationException e) {
+            RumbleException ex = new CastException(e.getMessage(), ExceptionMetadata.EMPTY_METADATA);
+            ex.initCause(e);
+            throw ex;
+        }
     }
 
     /**
@@ -152,7 +175,25 @@ public JavaRDD<Item> getAsRDD() {
         if (this.isOpen) {
             throw new RuntimeException("Cannot obtain an RDD if the iterator is open.");
         }
-        return this.iterator.getRDD(this.dynamicContext);
+        try {
+            return this.iterator.getRDD(this.dynamicContext);
+        } catch (NumberFormatException e) {
+            RumbleException ex = new UnexpectedTypeException(e.getMessage(), ExceptionMetadata.EMPTY_METADATA);
+            ex.initCause(e);
+            throw ex;
+        } catch (SparkRuntimeException e) {
+            if (e.getMessage().contains("CAST_INVALID_INPUT")) {
+                RumbleException ex = new CastException(e.getMessage(), ExceptionMetadata.EMPTY_METADATA);
+                ex.initCause(e);
+                throw ex;
+            } else {
+                throw e;
+            }
+        } catch (UnsupportedOperationException e) {
+            RumbleException ex = new UnexpectedTypeException(e.getMessage(), ExceptionMetadata.EMPTY_METADATA);
+            ex.initCause(e);
+            throw ex;
+        }
     }
 
     /**
@@ -226,8 +267,26 @@ public long populateListWithWarningOnlyIfCapReached(List<Item> resultList) {
             if (!this.isMaterialisable()) {
                 return -1;
             }
-            JavaRDD<Item> rdd = this.iterator.getRDD(this.dynamicContext);
-            return SparkSessionManager.collectRDDwithLimitWarningOnly(rdd, resultList);
+            try {
+                JavaRDD<Item> rdd = this.iterator.getRDD(this.dynamicContext);
+                return SparkSessionManager.collectRDDwithLimitWarningOnly(rdd, resultList);
+            } catch (NumberFormatException e) {
+                RumbleException ex = new UnexpectedTypeException(e.getMessage(), ExceptionMetadata.EMPTY_METADATA);
+                ex.initCause(e);
+                throw ex;
+            } catch (SparkRuntimeException e) {
+                if (e.getMessage().contains("CAST_INVALID_INPUT")) {
+                    RumbleException ex = new CastException(e.getMessage(), ExceptionMetadata.EMPTY_METADATA);
+                    ex.initCause(e);
+                    throw ex;
+                } else {
+                    throw e;
+                }
+            } catch (UnsupportedOperationException e) {
+                RumbleException ex = new UnexpectedTypeException(e.getMessage(), ExceptionMetadata.EMPTY_METADATA);
+                ex.initCause(e);
+                throw ex;
+            }
         } else {
             return populateList(resultList);
         }

src/main/java/org/rumbledb/cli/Main.java

@@ -24,9 +24,13 @@
 
 import org.apache.commons.io.IOUtils;
 import org.apache.spark.SparkException;
+import org.apache.spark.SparkRuntimeException;
 import org.rumbledb.config.RumbleRuntimeConfiguration;
+import org.rumbledb.exceptions.CastException;
+import org.rumbledb.exceptions.ExceptionMetadata;
 import org.rumbledb.exceptions.OurBadException;
 import org.rumbledb.exceptions.RumbleException;
+import org.rumbledb.exceptions.UnexpectedTypeException;
 import org.rumbledb.server.RumbleServer;
 import org.rumbledb.shell.RumbleJLineShell;
 
@@ -81,29 +85,22 @@ public static void main(String[] args) throws IOException {
     }
 
     private static void handleException(Throwable ex, boolean showErrorInfo) {
+        ex = unboxException(ex);
         if (ex != null) {
-            if (ex instanceof SparkException) {
-                Throwable sparkExceptionCause = ex.getCause();
-                if (sparkExceptionCause != null) {
-                    handleException(sparkExceptionCause, showErrorInfo);
-                } else {
-                    if (showErrorInfo) {
-                        ex.printStackTrace();
-                    }
-                    handleException(
-                        new OurBadException(
-                                "There was a problem with Spark, but Spark did not provide any cause or stracktrace. The message from Spark is:  "
-                                    + ex.getMessage()
-                        ),
-                        showErrorInfo
-                    );
-                }
-            } else if (ex instanceof RumbleException && !(ex instanceof OurBadException)) {
+            if (ex instanceof RumbleException && !(ex instanceof OurBadException)) {
                 System.err.println("⚠️  ️" + ex.getMessage());
                 if (showErrorInfo) {
                     ex.printStackTrace();
                 }
                 System.exit(42);
+            } else if (ex instanceof NumberFormatException) {
+                handleException(
+                    new UnexpectedTypeException(
+                            "A cast failed. However, since this happened in a native Spark SQL execution, we cannot show you where. You can get more information on this type error by retrying with --native-execution no.",
+                            ExceptionMetadata.EMPTY_METADATA
+                    ),
+                    showErrorInfo
+                );
             } else if (ex instanceof OutOfMemoryError) {
                 System.err.println(
                     "⚠️  Java went out of memory."
@@ -154,24 +151,54 @@ private static void handleException(Throwable ex, boolean showErrorInfo) {
                     ex.printStackTrace();
                 }
                 System.exit(-42);
-            } else {
-                System.err.println(
-                    "We are very embarrassed, because an error has occured that we did not anticipate 🙈: "
-                        + ex.getMessage()
-                );
-                System.err.println(
-                    "We would like to investigate this and make sure to fix it. We would be very grateful if you could contact us or file an issue on GitHub with your query."
+            }
+            System.err.println(
+                "We are very embarrassed, because an error has occured that we did not anticipate 🙈: "
+                    + ex.getMessage()
+            );
+            System.err.println(
+                "We would like to investigate this and make sure to fix it. We would be very grateful if you could contact us or file an issue on GitHub with your query."
+            );
+            System.err.println("Link: https://github.com/RumbleDB/rumble/issues");
+            System.err.println(
+                "For more debug info (e.g., so you can communicate it to us), please try again using --show-error-info yes in your command line."
+            );
+            if (showErrorInfo) {
+                ex.printStackTrace();
+            }
+            System.exit(-42);
+        }
+    }
+
+    public static Throwable unboxException(Throwable ex) {
+        if (ex != null) {
+            if (ex instanceof SparkException) {
+                Throwable sparkExceptionCause = ex.getCause();
+                if (sparkExceptionCause != null) {
+                    return sparkExceptionCause;
+                }
+                return new OurBadException(
+                        "There was a problem with Spark, but Spark did not provide any cause or stracktrace. The message from Spark is:  "
+                            + ex.getMessage()
                 );
-                System.err.println("Link: https://github.com/RumbleDB/rumble/issues");
-                System.err.println(
-                    "For more debug info (e.g., so you can communicate it to us), please try again using --show-error-info yes in your command line."
+            }
+            if (ex instanceof NumberFormatException) {
+                return new UnexpectedTypeException(
+                        "A cast failed. However, since this happened in a native Spark SQL execution, we cannot show you where. You can get more information on this type error by retrying with --native-execution no.",
+                        ExceptionMetadata.EMPTY_METADATA
                 );
-                if (showErrorInfo) {
-                    ex.printStackTrace();
+            }
+            if (ex instanceof SparkRuntimeException) {
+                if (ex.getMessage().contains("CAST_INVALID_INPUT")) {
+                    RumbleException nex = new CastException(ex.getMessage(), ExceptionMetadata.EMPTY_METADATA);
+                    ex.initCause(ex);
+                    return nex;
                 }
-                System.exit(-42);
+                // general message.
             }
+            return ex;
         }
+        return new OurBadException("A null exception was returned.");
     }
 
     private static void runQueryExecutor(RumbleRuntimeConfiguration sparksoniqConf) throws IOException {

src/main/java/org/rumbledb/compiler/InferTypeVisitor.java

@@ -128,6 +128,8 @@
 import org.rumbledb.types.ItemType;
 import org.rumbledb.types.ItemTypeFactory;
 import org.rumbledb.types.SequenceType;
+import org.rumbledb.types.SequenceType.Arity;
+
 import sparksoniq.spark.SparkSessionManager;
 
 

src/main/java/org/rumbledb/context/DynamicContext.java

@@ -142,13 +142,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 int getCurrentMutabilityLevel() {

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);
+    }
+
 }

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);
+    }
 }