Merge pull request #1256 from RumbleDB/deltalake

Deltalake

.github/workflows/maven.yml

@@ -115,3 +115,26 @@ jobs:
     - name: MLTestsNativeDeactivated
       run: mvn -Dtest=MLTestsNativeDeactivated test
 
+  tests4:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Java 11
+      uses: actions/setup-java@v3
+      with:
+        java-version: 11
+        distribution: adopt
+    - name: Cache Maven packages
+      uses: actions/cache@v3
+      with:
+        path: ~/.m2
+        key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
+        restore-keys: ${{ runner.os }}-m2
+    - name: Install with Maven
+      run: mvn install -DskipTests -Dgpg.skip --quiet
+    - name: Compile with Maven
+      run: mvn clean compile assembly:single
+    - name: DeltaUpdateRuntimeTests
+      run: mvn -Dtest=DeltaUpdateRuntimeTests test

.gitlab-ci.yml

@@ -79,8 +79,13 @@ NativeFLWORRuntimeTestsParallelismDeactivated:
   script:
     - mvn -Dtest=NativeFLWORRuntimeTestsParallelismDeactivated test
 
-StaticTypingTest:
-  stage: tests3
+updatedeltaruntime-test:
+  stage: test
+  script:
+    - mvn -Dtest=DeltaUpdateRuntimeTests test
+
+statictyping-test:
+  stage: test
   script:
     - mvn -Dtest=StaticTypeTests test
 

pom.xml

@@ -197,6 +197,16 @@
     </build>
 
     <dependencies>
+        <dependency>
+            <groupId>org.openjdk.jmh</groupId>
+            <artifactId>jmh-core</artifactId>
+            <version>1.37</version>
+        </dependency>
+        <dependency>
+            <groupId>org.openjdk.jmh</groupId>
+            <artifactId>jmh-generator-annprocess</artifactId>
+            <version>1.37</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-core_2.12</artifactId>
@@ -257,36 +267,46 @@
             <artifactId>commons-lang3</artifactId>
             <version>3.12.0</version>
         </dependency>
-		<dependency>
+        <dependency>
+            <groupId>commons-net</groupId>
+            <artifactId>commons-net</artifactId>
+            <version>3.1</version>
+        </dependency>
+        <dependency>
             <groupId>commons-io</groupId>
             <artifactId>commons-io</artifactId>
             <version>2.11.0</version>
         </dependency>
-		<dependency>
-		  <groupId>org.apache.httpcomponents</groupId>
-		  <artifactId>httpclient</artifactId>
-		  <version>4.5.13</version>
-		</dependency>
-		<!--<dependency>
-		    <groupId>edu.vanderbilt.accre</groupId>
-		    <artifactId>laurelin</artifactId>
-		    <version>1.0.1</version>
-		</dependency>-->
-		<dependency>
-			<groupId>org.jgrapht</groupId>
-			<artifactId>jgrapht-core</artifactId>
-			<version>1.4.0</version>
-		</dependency>
-		<dependency>
-			<groupId>joda-time</groupId>
-			<artifactId>joda-time</artifactId>
-			<version>2.10.6</version>
-		</dependency>
-		<dependency>
-		  <groupId>com.fasterxml.jackson.dataformat</groupId>
-		  <artifactId>jackson-dataformat-yaml</artifactId>
-		  <version>2.13.4</version>
-		</dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>4.5.13</version>
+        </dependency>
+	<!--<dependency>
+	    <groupId>edu.vanderbilt.accre</groupId>
+	    <artifactId>laurelin</artifactId>
+	    <version>1.0.1</version>
+	</dependency>-->
+        <dependency>
+            <groupId>org.jgrapht</groupId>
+            <artifactId>jgrapht-core</artifactId>
+            <version>1.4.0</version>
+        </dependency>
+        <dependency>
+            <groupId>joda-time</groupId>
+            <artifactId>joda-time</artifactId>
+            <version>2.10.6</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+            <version>2.13.4</version>
+        </dependency>
+        <dependency>
+            <groupId>io.delta</groupId>
+            <artifactId>delta-core_2.12</artifactId>
+            <version>2.4.0</version>
+        </dependency>
     </dependencies>
 
     <distributionManagement>

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

@@ -692,17 +692,113 @@ default boolean isNaN() {
      * @return an int representing nestedness of the item inside transform expressions.
      */
     default int getMutabilityLevel() {
-        return 0;
+        return -1;
     }
 
     /**
-     * Sets the mutability level of the item.
+     * Sets the mutability level of the item to a supplied value.
      *
-     * @param mutabilityLevel the new mutability level.
+     * @param mutabilityLevel new mutability level.
      */
     default void setMutabilityLevel(int mutabilityLevel) {
     }
 
+    /**
+     * Returns the top level ID of the item.
+     *
+     * @return int representing the rowID of the item within a DeltaFile.
+     */
+    default long getTopLevelID() {
+        return -1;
+    }
+
+    /**
+     * Sets the top level ID of the item to a supplied value.
+     *
+     * @param topLevelID new top level ID.
+     */
+    default void setTopLevelID(long topLevelID) {
+    }
+
+    /**
+     * Returns the path from the top level object of a DeltaFile for the item.
+     *
+     * @return String representing the path of the item from the top level within a DeltaFile.
+     */
+    default String getPathIn() {
+        return "null";
+    }
+
+    /**
+     * Sets the path from the top level object of a DeltaFile for the item to a supplied value.
+     *
+     * @param pathIn new path from top level.
+     */
+    default void setPathIn(String pathIn) {
+    }
+
+    /**
+     * Returns the location of the DeltaFile for the item.
+     *
+     * @return String representing the location of the DeltaFile for the item.
+     */
+    default String getTableLocation() {
+        return null;
+    }
+
+
+    /**
+     * Sets the location of the DeltaFile for the item to a supplied value.
+     *
+     * @param location new location of the DeltaFile for the item.
+     */
+    default void setTableLocation(String location) {
+    }
+
+    /**
+     * Returns the SparkSQL value of the item for use in a query.
+     *
+     * @return String representing the SparkSQL value of the item.
+     */
+    default String getSparkSQLValue() {
+        throw new UnsupportedOperationException("Operation not defined for type " + this.getDynamicType());
+    }
+
+    /**
+     * Returns the SparkSQL value of the item for use in a query.
+     *
+     * @return String representing the SparkSQL value of the item.
+     */
+    default String getSparkSQLValue(ItemType itemType) {
+        throw new UnsupportedOperationException("Operation not defined for type " + this.getDynamicType());
+    }
+
+    /**
+     * Returns the SparkSQL type of the item for use in a query.
+     *
+     * @return String representing the SparkSQL type of the item.
+     */
+    default String getSparkSQLType() {
+        throw new UnsupportedOperationException("Operation not defined for type " + this.getDynamicType());
+    }
+
+    /**
+     * Tests for physical equality. The semantics are that of the eq operator.
+     *
+     * @param other another item.
+     * @return true it is equal to other, false otherwise.
+     */
+    default boolean physicalEquals(Object other) {
+        if (!(other instanceof Item)) {
+            return false;
+        }
+        Item otherItem = (Item) other;
+        if (this.getTopLevelID() == -1 || otherItem.getTopLevelID() == -1) {
+            return System.identityHashCode(this) == System.identityHashCode(otherItem);
+        }
+        return this.getTopLevelID() == otherItem.getTopLevelID() && this.getPathIn().equals(otherItem.getPathIn());
+    }
+
     /**
      * Tests for logical equality. The semantics are that of the eq operator.
      *

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

@@ -5,7 +5,6 @@
 import org.rumbledb.context.DynamicContext;
 import org.rumbledb.expressions.module.MainModule;
 import org.rumbledb.runtime.RuntimeIterator;
-import org.rumbledb.runtime.update.PendingUpdateList;
 import sparksoniq.spark.SparkSessionManager;
 
 import java.io.IOException;
@@ -52,11 +51,6 @@ public SequenceOfItems runQuery(String query) {
             this.configuration
         );
 
-        if (iterator.isUpdating()) {
-            PendingUpdateList pul = iterator.getPendingUpdateList(dynamicContext);
-            pul.applyUpdates(iterator.getMetadata());
-        }
-
         return new SequenceOfItems(iterator, dynamicContext, this.configuration);
     }
 
@@ -78,12 +72,6 @@ public SequenceOfItems runQuery(URI location) throws IOException {
             this.configuration
         );
 
-        if (iterator.isUpdating()) {
-            PendingUpdateList pul = iterator.getPendingUpdateList(dynamicContext);
-            pul.applyUpdates(iterator.getMetadata());
-        }
-
-        System.err.println("final iterator is: " + iterator.isUpdating());
         return new SequenceOfItems(iterator, dynamicContext, this.configuration);
     }