[voltdb]: update subtree v2 using STARTS WITH (#337)

* [voltdb]:  update subtree v2 using `STARTS WITH`

* Fix: VoltDB Stored Procedure

filescale_init/build_with_volt.sh

@@ -1,9 +1,11 @@
 # copy the following command lines into test.sh
 set -xe
 
+cd $HADOOP_HOME
 ./sbin/stop-dfs.sh
 
 export DATABASE="VOLT"
+export VOLT_SERVER="localhost"
 
 if [ -z "$1" ]
 then
@@ -14,7 +16,7 @@ fi
 
 # compile stored procedures
 cd ~/hadoop/filescale_init/voltdb && bash clean_procedures.sh $IP
-cd .. && javac HdfsMetaInfoSchema.java && java HdfsMetaInfoSchema
+cd ../src/main/java && javac HdfsMetaInfoSchema.java && java HdfsMetaInfoSchema
 cd ~/hadoop/filescale_init/voltdb && bash create_procedures.sh $IP
 
 # restart hadoop hdfs

filescale_init/pom.xml

@@ -30,6 +30,13 @@
             <artifactId>ignite-spring</artifactId>
             <version>${ignite.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.voltdb</groupId>
+            <artifactId>voltdbclient</artifactId>
+            <version>9.0</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
 </project>

filescale_init/src/main/java/HdfsMetaInfoSchema.java

@@ -264,50 +264,6 @@ private HdfsMetaInfoSchema() throws SQLException {
     } catch (SQLException ex) {
       System.err.println(ex.getMessage());
     }
-
-    // key-value API test
-		IgniteCluster cluster = ignite_client.cluster();
-    cluster.active(true);
-    cluster.enableWal("inodes");
-		cluster.baselineAutoAdjustEnabled(false);
-
-    Collection<String> collection = ignite_client.cacheNames();
-    System.out.println("cache names = " + collection);
-
-    IgniteCache<BinaryObject, BinaryObject> inodesBinary = ignite_client.cache("inodes").withKeepBinary();
-    System.out.println(">> Updating inode record:");
-
-    BinaryObjectBuilder inodeKeyBuilder = ignite_client.binary().builder("InodeKey");
-    BinaryObject inodeKey = inodeKeyBuilder.setField("parentName", "/").setField("name", "hello").build();
-    BinaryObjectBuilder inodeBuilder = ignite_client.binary().builder("INode");
-    BinaryObject inode = inodeBuilder
-      .setField("id", 11111L, Long.class)
-      .setField("parent", 0L, Long.class)
-      .setField("parentName", "/")
-      .setField("name", "hello")
-      .setField("accessTime", 22222L, Long.class)
-      .setField("modificationTime", 33333L, Long.class)
-      .setField("header", 0L, Long.class)
-      .setField("permission", 777L, Long.class)
-      .build();
-    System.out.printf("The dir: %s, id: %s \n", inode.field("parentName"), inode.field("name"), inode.field("id"));
-    inodesBinary.put(inodeKey, inode);
-
-    IgniteCompute compute = ignite_client.compute();
-    // Execute closure on all cluster nodes.
-    IgniteCallable<String> call = new WalPointerTask(); 
-    String res = compute.call(call);
-    System.out.printf("Last Wal pointer: " + res);
-    ignite_client.close();
-
-    // SQL test
-    try {
-      Statement st = connection.createStatement();
-      st.execute("delete from inodes where id = 11111;");
-      st.close();
-    } catch (SQLException ex) {
-      System.err.println(ex.getMessage());
-    } 
   }
 
   public Connection getConnection() {

filescale_init/voltdb/SetPermission.java

@@ -1,7 +1,7 @@
 import org.voltdb.*;
 
 // https://docs.voltdb.com/tutorial/Part5.php
-public class SetPermissions extends VoltProcedure {
+public class SetPermission extends VoltProcedure {
 
   public final SQLStmt sql = new SQLStmt("UPDATE inodes SET permission = ? WHERE id = ?;");
 

filescale_init/voltdb/UpdateSubtreeV2.java

@@ -0,0 +1,71 @@
+import org.voltdb.*;
+import java.util.*;
+
+// https://docs.voltdb.com/tutorial/Part5.php
+public class UpdateSubtreeV2 extends VoltProcedure {
+  public final SQLStmt sql1 = new SQLStmt(
+    "SELECT id, name, accessTime, modificationTime, permission,"
+    + "header, parent, parentName from inodes WHERE parentName STARTS WITH ?;");
+
+  public final SQLStmt sql2 = new SQLStmt("INSERT INTO inodes("
+    + "id, name, accessTime, modificationTime, permission, header, parent, parentName"
+    + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?);");
+
+  public final SQLStmt sql3 = new SQLStmt("DELETE FROM inodes where parentName STARTS WITH ?;");
+
+  public long run(final long dir_id, final long dest_id, final String old_parent_name,
+    final String new_parent_name, final long new_parent) throws VoltAbortException {
+    // 1. find subtree records
+    voltQueueSQL(sql1, old_parent_name);
+    VoltTable[] res = voltExecuteSQL();
+
+    // 2. update subtree records
+    Long id = null;
+    String name = null;
+    Long accessTime = null;
+    Long modificationTime = null;
+    Long permission = null;
+    Long header = null;
+    Long parent = null;
+    String parentName = null;
+    for (int j = 0; j < res.length; ++j) {
+      for (int i = 0; i < res[j].getRowCount(); ++i) {
+        VoltTableRow row = res[j].fetchRow(i);
+        id = row.getLong(0);
+        name = row.getString(1);
+        accessTime = row.getLong(2);
+        modificationTime = row.getLong(3);
+        permission = row.getLong(4);
+        header = row.getLong(5);
+        parent = row.getLong(6);
+        parentName = row.getString(7);
+
+        if (id == dir_id) {
+          id += dest_id;
+          parent = new_parent;
+          parentName = new_parent_name;
+        } else {
+          id += dest_id;
+          parent += dest_id;
+          parentName = new_parent_name + parentName.substring(old_parent_name.length());
+        }
+        voltQueueSQL(sql2,
+          id,
+          name,
+          accessTime,
+          modificationTime,
+          permission,
+          header,
+          parent,
+          parentName);
+      }
+    }
+    voltExecuteSQL();
+
+    // 3. delete old subtree records
+    voltQueueSQL(sql3, old_parent_name);
+    voltExecuteSQL();
+
+    return getUniqueId();
+  }
+}

filescale_init/voltdb/docker/deploy.py

@@ -42,3 +42,6 @@
           "create",
           "--deployment=/root/voltdb-ent/deployment.xml",
           "--host=" + sys.argv[3]])
+
+# run voltdb
+# ./deploy.py 1 0 0.0.0.0 &

hadoop-hdfs-project/hadoop-hdfs-db/src/main/java/org/apache/hadoop/hdfs/db/DatabaseINode.java

@@ -2119,7 +2119,7 @@ public static long updateSubtree(final long dir_id, final long dest_id, final St
       if (env.equals("VOLT")) {
         try {
           VoltTable[] results = obj.getVoltClient()
-              .callProcedure("UpdateSubtree", dir_id, dest_id, old_parent_name,
+              .callProcedure("UpdateSubtreeV2", dir_id, dest_id, old_parent_name,
               new_parent_name, new_parent).getResults();
               VoltTable result = results[0];
               result.resetRowPosition();
@@ -2137,7 +2137,7 @@ public static long updateSubtree(final long dir_id, final long dest_id, final St
       System.err.println(ex.getMessage());
     }
     if (LOG.isInfoEnabled()) {
-      LOG.info("txnId: " + res + " updateSubtree [UPDATE]: " + dir_id);
+      LOG.info("txnId: " + res + " updateSubtree v2 [UPDATE]: " + dir_id);
     }
     return res;
   }