Add RuntimeDataFrame<T> and RuntimeDataFrameFactory<T> interface

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

@@ -21,13 +21,11 @@
 package org.rumbledb.context;
 
 import org.apache.spark.api.java.JavaRDD;
-import org.apache.spark.sql.Row;
 import org.rumbledb.api.Item;
 import org.rumbledb.config.RumbleRuntimeConfiguration;
 import org.rumbledb.errorcodes.ErrorCode;
 import org.rumbledb.exceptions.*;
 import org.rumbledb.items.ItemFactory;
-import org.rumbledb.items.parsing.RowToItemMapper;
 import org.rumbledb.items.structured.JSoundDataFrame;
 import org.rumbledb.runtime.HybridRuntimeIterator;
 
@@ -223,7 +221,7 @@ public List<Item> getLocalVariableValue(Name varName, ExceptionMetadata metadata
             }
             JSoundDataFrame df = this.getDataFrameVariableValue(varName, metadata);
             return HybridRuntimeIterator.collectRDDwithLimit(
-                HybridRuntimeIterator.dataFrameToRDDOfItems(df, metadata),
+                df.toRDD(metadata),
                 this.configuration,
                 metadata
             );
@@ -271,8 +269,7 @@ public JavaRDD<Item> getRDDVariableValue(Name varName, ExceptionMetadata metadat
                 throw new JobWithinAJobException(metadata);
             }
             JSoundDataFrame df = this.dataFrameVariableValues.get(varName);
-            JavaRDD<Row> rowRDD = df.javaRDD();
-            return rowRDD.map(new RowToItemMapper(metadata, df.getItemType()));
+            return df.toRDD(metadata);
         }
 
         if (this.parent != null) {
@@ -466,4 +463,3 @@ public void changeVariableValue(Name varName, JSoundDataFrame value) {
         nodeWithVariableDecl.dataFrameVariableValues.put(varName, value);
     }
 }
-

src/main/java/org/rumbledb/exceptions/ExitStatementException.java

@@ -8,8 +8,6 @@
 import java.io.Serial;
 import java.util.List;
 
-import static org.rumbledb.runtime.HybridRuntimeIterator.dataFrameToRDDOfItems;
-
 public class ExitStatementException extends RuntimeException {
     @Serial
     private static final long serialVersionUID = 1L;
@@ -43,7 +41,7 @@ public List<Item> getLocalResult() {
         } else if (hasRDDResult()) {
             return this.rddResult.collect();
         } else if (hasDataFrameResult()) {
-            return dataFrameToRDDOfItems(this.dataFrameResult, this.exceptionMetadata).collect();
+            return this.dataFrameResult.toRDD(this.exceptionMetadata).collect();
         }
         throw new OurBadException("Expected local result but there was nothing to return from the exit statement!");
     }

src/main/java/org/rumbledb/items/structured/JSoundDataFrame.java

@@ -18,14 +18,16 @@
 import org.rumbledb.exceptions.ExceptionMetadata;
 import org.rumbledb.exceptions.OurBadException;
 import org.rumbledb.items.parsing.ItemParser;
+import org.rumbledb.items.parsing.RowToItemMapper;
+import org.rumbledb.runtime.dataframe.RuntimeDataFrame;
 import org.rumbledb.runtime.flwor.FlworDataFrameColumn;
 import org.rumbledb.types.BuiltinTypesCatalogue;
 import org.rumbledb.types.ItemType;
 import org.rumbledb.types.ItemTypeFactory;
 
 import sparksoniq.spark.SparkSessionManager;
 
-public class JSoundDataFrame implements Serializable {
+public class JSoundDataFrame implements RuntimeDataFrame<Item>, Serializable {
     @Serial
     private static final long serialVersionUID = 1L;
 
@@ -126,6 +128,7 @@ public static JSoundDataFrame emptyDataFrame() {
         );
     }
 
+    @Override
     public Dataset<Row> getDataFrame() {
         return this.dataFrame;
     }
@@ -135,8 +138,15 @@ public void show() {
         this.dataFrame.show();
     }
 
-    public JavaRDD<Row> javaRDD() {
-        return this.dataFrame.javaRDD();
+    /**
+     * Converts this JSONiq DataFrame to its logical item representation.
+     *
+     * @param metadata query metadata used if a row cannot be decoded
+     * @return an RDD containing the represented items
+     */
+    @Override
+    public JavaRDD<Item> toRDD(ExceptionMetadata metadata) {
+        return this.dataFrame.javaRDD().map(new RowToItemMapper(metadata, this.itemType));
     }
 
     public long count() {

src/main/java/org/rumbledb/runtime/HybridRuntimeIterator.java

@@ -21,7 +21,6 @@
 package org.rumbledb.runtime;
 
 import org.apache.spark.api.java.JavaRDD;
-import org.apache.spark.sql.Row;
 import org.rumbledb.api.Item;
 import org.rumbledb.config.RumbleRuntimeConfiguration;
 import org.rumbledb.context.DynamicContext;
@@ -32,9 +31,7 @@
 import org.rumbledb.exceptions.MoreThanOneItemException;
 import org.rumbledb.exceptions.NoItemException;
 import org.rumbledb.expressions.ExecutionMode;
-import org.rumbledb.items.parsing.RowToItemMapper;
-import org.rumbledb.items.structured.JSoundDataFrame;
-
+import org.rumbledb.runtime.dataframe.RuntimeDataFrame;
 import sparksoniq.spark.SparkSessionManager;
 
 import java.io.Serial;
@@ -100,10 +97,7 @@ public boolean hasNext() {
             this.currentResultIndex = 0;
             JavaRDD<Item> rdd = null;
             if (!isRDD() && implementsDataFrames()) {
-                rdd = dataFrameToRDDOfItems(
-                    this.getDataFrame(this.currentDynamicContextForLocalExecution),
-                    this.getMetadata()
-                );
+                rdd = this.getDataFrame(this.currentDynamicContextForLocalExecution).toRDD(this.getMetadata());
             } else {
                 rdd = this.getRDDAux(this.currentDynamicContextForLocalExecution);
             }
@@ -141,8 +135,8 @@ public Item next() {
     @Override
     public JavaRDD<Item> getRDD(DynamicContext context) {
         if ((isDataFrame() && implementsDataFrames()) || (isRDD() && implementsDataFrames() && !implementsRDD())) {
-            JSoundDataFrame df = this.getDataFrame(context);
-            return dataFrameToRDDOfItems(df, getMetadata());
+            RuntimeDataFrame<Item> df = this.getDataFrame(context);
+            return df.toRDD(getMetadata());
         }
         if (isRDDOrDataFrame()) {
             return getRDDAux(context);
@@ -151,11 +145,6 @@ public JavaRDD<Item> getRDD(DynamicContext context) {
         return SparkSessionManager.getInstance().getJavaSparkContext().parallelize(contents);
     }
 
-    public static JavaRDD<Item> dataFrameToRDDOfItems(JSoundDataFrame df, ExceptionMetadata metadata) {
-        JavaRDD<Row> rowRDD = df.javaRDD();
-        return rowRDD.map(new RowToItemMapper(metadata, df.getItemType()));
-    }
-
     public static List<Item> collectRDDwithLimit(
             JavaRDD<Item> rdd,
             RumbleRuntimeConfiguration configuration,

src/main/java/org/rumbledb/runtime/RuntimeIterator.java

@@ -42,16 +42,13 @@
 import org.rumbledb.expressions.ExecutionMode;
 import org.rumbledb.expressions.comparison.ComparisonExpression.ComparisonOperator;
 import org.rumbledb.items.structured.JSoundDataFrame;
+import org.rumbledb.runtime.dataframe.ItemRuntimeDataFrameFactory;
 import org.rumbledb.runtime.flwor.NativeClauseContext;
 import org.rumbledb.runtime.misc.ComparisonIterator;
-import org.rumbledb.runtime.typing.TypeInferrenceUtils;
-import org.rumbledb.runtime.typing.ValidateTypeIterator;
 import org.rumbledb.runtime.update.PendingUpdateList;
 import org.rumbledb.types.BuiltinTypesCatalogue;
-import org.rumbledb.types.ItemType;
 import org.rumbledb.types.SequenceType;
 
-
 public abstract class RuntimeIterator implements RuntimeIteratorInterface<Item> {
 
     protected static final String FLOW_EXCEPTION_MESSAGE = "Invalid next() call; ";
@@ -311,57 +308,9 @@ public final JSoundDataFrame getOrCreateDataFrame(DynamicContext context) {
             return this.getDataFrame(context);
         }
         if (isRDD()) {
-            if (this.getStaticType().getItemType().isCompatibleWithDataFrames(this.getConfiguration())) {
-                return ValidateTypeIterator.convertRDDToValidDataFrame(
-                    this.getRDD(context),
-                    this.getStaticType().getItemType(),
-                    context,
-                    true,
-                    this.staticContext
-                );
-            } else {
-                JavaRDD<Item> rdd = this.getRDD(context);
-                ItemType type = TypeInferrenceUtils.inferItemTypeOfRDDItems(
-                    rdd,
-                    getMetadata(),
-                    TypeInferrenceUtils.TypeMergeMode.LAX
-                );
-                return ValidateTypeIterator.convertRDDToValidDataFrame(
-                    rdd,
-                    type,
-                    context,
-                    true,
-                    this.staticContext
-                );
-            }
-        }
-        List<Item> items = new ArrayList<>();
-        materialize(context, items);
-        if (this.getStaticType().getItemType().isCompatibleWithDataFrames(this.getConfiguration())) {
-            return ValidateTypeIterator.convertLocalItemsToDataFrame(
-                items,
-                this.getStaticType().getItemType(),
-                context,
-                true,
-                this.staticContext
-            );
-        } else {
-            ItemType type = TypeInferrenceUtils.inferItemTypeOfLocalItems(
-                items,
-                getMetadata(),
-                TypeInferrenceUtils.TypeMergeMode.LAX
-            );
-            if (this.getConfiguration().printInferredTypes()) {
-                System.err.println("Inferred DataFrame type:\n" + this.getStaticType().getItemType());
-            }
-            return ValidateTypeIterator.convertLocalItemsToDataFrame(
-                items,
-                type,
-                context,
-                true,
-                this.staticContext
-            );
+            return ItemRuntimeDataFrameFactory.INSTANCE.fromRDD(this.getRDD(context), context, this.staticContext);
         }
+        return ItemRuntimeDataFrameFactory.INSTANCE.fromLocal(this.materialize(context), context, this.staticContext);
     }
 
     public boolean isUpdating() {

src/main/java/org/rumbledb/runtime/dataframe/ItemRuntimeDataFrameFactory.java

@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.rumbledb.runtime.dataframe;
+
+import java.io.Serial;
+import java.util.List;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.rumbledb.api.Item;
+import org.rumbledb.context.DynamicContext;
+import org.rumbledb.context.RuntimeStaticContext;
+import org.rumbledb.items.structured.JSoundDataFrame;
+import org.rumbledb.runtime.typing.TypeInferrenceUtils;
+import org.rumbledb.runtime.typing.ValidateTypeIterator;
+import org.rumbledb.types.ItemType;
+
+/**
+ * Encodes item RDDs as {@link JSoundDataFrame}s.
+ */
+public final class ItemRuntimeDataFrameFactory implements RuntimeDataFrameFactory<Item> {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    public static final ItemRuntimeDataFrameFactory INSTANCE = new ItemRuntimeDataFrameFactory();
+
+    private ItemRuntimeDataFrameFactory() {
+    }
+
+    @Override
+    public JSoundDataFrame fromLocal(
+            List<Item> items,
+            DynamicContext context,
+            RuntimeStaticContext staticContext
+    ) {
+        ItemType itemType = staticContext.getStaticType().getItemType();
+        if (!itemType.isCompatibleWithDataFrames(staticContext.getConfiguration())) {
+            itemType = TypeInferrenceUtils.inferItemTypeOfLocalItems(
+                items,
+                staticContext.getMetadata(),
+                TypeInferrenceUtils.TypeMergeMode.LAX
+            );
+            if (staticContext.getConfiguration().printInferredTypes()) {
+                System.err.println("Inferred DataFrame type:\n" + itemType);
+            }
+        }
+        return ValidateTypeIterator.convertLocalItemsToDataFrame(items, itemType, context, true, staticContext);
+    }
+
+    @Override
+    public JSoundDataFrame fromRDD(
+            JavaRDD<Item> rdd,
+            DynamicContext context,
+            RuntimeStaticContext staticContext
+    ) {
+        ItemType itemType = staticContext.getStaticType().getItemType();
+        if (!itemType.isCompatibleWithDataFrames(staticContext.getConfiguration())) {
+            itemType = TypeInferrenceUtils.inferItemTypeOfRDDItems(
+                rdd,
+                staticContext.getMetadata(),
+                TypeInferrenceUtils.TypeMergeMode.LAX
+            );
+        }
+        return ValidateTypeIterator.convertRDDToValidDataFrame(rdd, itemType, context, true, staticContext);
+    }
+}

src/main/java/org/rumbledb/runtime/dataframe/RuntimeDataFrame.java

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.rumbledb.runtime.dataframe;
+
+import java.io.Serializable;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.rumbledb.exceptions.ExceptionMetadata;
+
+/**
+ * A Spark DataFrame whose rows represent runtime values of type {@code T}.
+ *
+ * <p>
+ * Spark stores the physical representation as {@link Row}; implementations define how rows are mapped back to the
+ * logical runtime type.
+ * </p>
+ *
+ * @param <T> the logical runtime value represented by each row
+ */
+public interface RuntimeDataFrame<T> extends Serializable {
+
+    /**
+     * Returns the underlying physical Spark DataFrame.
+     */
+    Dataset<Row> getDataFrame();
+
+    /**
+     * Converts this DataFrame to its logical runtime representation.
+     *
+     * @param metadata query metadata used if a row cannot be decoded
+     * @return an RDD of logical runtime values
+     */
+    JavaRDD<T> toRDD(ExceptionMetadata metadata);
+}

src/main/java/org/rumbledb/runtime/dataframe/RuntimeDataFrameFactory.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.rumbledb.runtime.dataframe;
+
+import java.io.Serializable;
+import java.util.List;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.rumbledb.context.DynamicContext;
+import org.rumbledb.context.RuntimeStaticContext;
+
+/**
+ * Creates a typed runtime DataFrame from the logical values stored in an RDD.
+ *
+ * @param <T> the logical runtime value represented by each DataFrame row
+ */
+public interface RuntimeDataFrameFactory<T> extends Serializable {
+
+    RuntimeDataFrame<T> fromLocal(
+            List<T> values,
+            DynamicContext context,
+            RuntimeStaticContext staticContext
+    );
+
+    RuntimeDataFrame<T> fromRDD(
+            JavaRDD<T> rdd,
+            DynamicContext context,
+            RuntimeStaticContext staticContext
+    );
+}