Implements Hive Input/Output format to use vineyard as the storage backend

java/hive/README.rst

@@ -215,24 +215,29 @@ Hive and Vineyard
         stored as
             INPUTFORMAT 'io.v6d.hive.ql.io.VineyardInputFormat'
             OUTPUTFORMAT 'io.v6d.hive.ql.io.VineyardOutputFormat';
-        insert into table hive_static_partition partition(value=666) values (1, 2);
+        insert into table hive_static_partition partition(value=666) values (999, 2), (999, 2), (999, 2);
         insert into table hive_static_partition partition(value=666) values (3, 4);
         insert into table hive_static_partition partition(value=114514) values (1, 2);
         select * from hive_static_partition;
         select * from hive_static_partition where value=666;
         select * from hive_static_partition where value=114514;
 
-- Test static partition(BUG):
+- Test dynamic partition:
+
+    We must set the batch size to 1. Because ArrowColumnarBatchSerDe do not process dynamic partitioning. If
+    the batch size is not set to 1, ArrowColumnarBatchSerDe will batch data from different partitions together,
+    and the data will be written to the same partition, which will cause the data to be written incorrectly.
 
     .. code:: sql
+
         set hive.fetch.task.conversion=none;
         set hive.vectorized.use.vectorized.input.format=true;
         set hive.vectorized.use.row.serde.deserialize=false;
         set hive.vectorized.use.vector.serde.deserialize=true;
         set hive.vectorized.execution.enabled=true;
         set hive.vectorized.execution.reduce.enabled=true;
         set hive.vectorized.row.serde.inputformat.excludes=io.v6d.hive.ql.io.VineyardInputFormat;
-        set hive.arrow.batch.size=500;
+        set hive.arrow.batch.size=1;
         set hive.exec.dynamic.partition=true;
         set hive.exec.dynamic.partition.mode=nonstrict;
         create table hive_dynamic_partition_data
@@ -245,9 +250,10 @@ Hive and Vineyard
         (
             src_id int,
             dst_id int
-        )partitioned by(year int)
+        )partitioned by(mounth int, year int)
         row format serde "org.apache.hadoop.hive.ql.io.arrow.ArrowColumnarBatchSerDe"
         stored as
             INPUTFORMAT 'io.v6d.hive.ql.io.VineyardInputFormat'
             OUTPUTFORMAT 'io.v6d.hive.ql.io.VineyardOutputFormat';
-        insert into table hive_dynamic_partition_test partition(year) select src_id,dst_id,year from hive_dynamic_partition_data;
+        insert into table hive_dynamic_partition_test partition(mounth=1, year) select src_id,dst_id,year from hive_dynamic_partition_data;
+        select * from hive_dynamic_partition_test;

java/hive/src/main/java/io/v6d/hive/ql/io/VineyardOutputFormat.java

@@ -25,17 +25,23 @@
 import io.v6d.modules.basic.arrow.Arrow;
 import io.v6d.modules.basic.arrow.RecordBatchBuilder;
 
+import java.io.File;
 import java.io.IOException;
 import java.util.Properties;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+
 import lombok.val;
 
 import org.apache.arrow.vector.VectorSchemaRoot;
 import org.apache.arrow.vector.types.pojo.*;
 import org.apache.arrow.vector.*;
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
 import org.apache.hadoop.hive.ql.io.HiveOutputFormat;
 import org.apache.hadoop.hive.ql.io.arrow.ArrowWrapperWritable;
@@ -45,6 +51,7 @@
 import org.apache.hadoop.mapred.RecordWriter;
 import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.util.Progressable;
+import org.apache.hadoop.hive.common.FileUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -81,6 +88,7 @@
     private Path finalOutPath;
     private Properties tableProperties;
     private Progressable progress;
+    private FileSystem fs;
 
     // vineyard
     private IPCClient client;
@@ -92,10 +100,21 @@
     private List<String> partitions;
     private boolean hasData = false;
 
+    public static final PathFilter VINEYARD_FILES_PATH_FILTER = new PathFilter() {
+        @Override
+        public boolean accept(Path p) {
+            String name = p.getName();
+            return name.startsWith("_task_tmp.") && (!name.substring(10, name.length()).startsWith("-"));
+        }
+    };
+
     private void getTableName(Properties tableProperties) {
         String location = tableProperties.getProperty("location");
         // int index = location.lastIndexOf("/");
         // tableName = location.substring(index + 1);
+        System.out.println("finalOutPath : "+ finalOutPath.toString());
+
+        // Get partition count
         String partition = tableProperties.getProperty("partition_columns.types");
         int index = -1;
         int partitionCount= 0;
@@ -106,14 +125,57 @@
             } while(index != -1);
         }
         System.out.println("Partition count:" + partitionCount);
-        index = location.length() + 1;
-        for (int i = 0; i < partitionCount; i++) {
-            index = finalOutPath.toString().indexOf("/", index + 1);
+
+        // Construct table name
+        String path = finalOutPath.toString();
+        path = path.substring(location.length(), path.length());
+        String pathSplits[] = path.split("/");
+        PathFilter vineyardPathFilter = VINEYARD_FILES_PATH_FILTER;
+        PathFilter hiddernPathFilter = FileUtils.HIDDEN_FILES_PATH_FILTER;
+        if (pathSplits.length == 0) {
+            return;
+        }
+        tableName = location;
+        for (int i = 1; i < pathSplits.length; i++) {
+            if (pathSplits[i].length() > 0 && hiddernPathFilter.accept(new Path(pathSplits[i]))) {
+                System.out.println("path split:" + pathSplits[i]);
+                tableName += "/" + pathSplits[i];
+            } else if (pathSplits[i].length() > 0 && vineyardPathFilter.accept(new Path(pathSplits[i]))) {
+                System.out.println("path split:" + pathSplits[i].substring(10, pathSplits[i].length()));
+                tableName += "/" + pathSplits[i].substring(10, pathSplits[i].length());
+            }
         }
-        tableName = finalOutPath.toString().substring(0, index);
-        tableName = tableName.replace('/', '#');
+        tableName = tableName.replaceAll("/", "#");
         System.out.println("Table name:" + tableName);
-        System.out.println("fina path:" + finalOutPath.toString());
+
+        // Create temp file
+        String tmpFilePath = finalOutPath.toString();
+        tmpFilePath = tmpFilePath.substring(0, tmpFilePath.lastIndexOf("/"));
+        System.out.println("out path:" + tmpFilePath);
+        // File file = FileUtils.createTempFile(outPath, "vineyard", ".tmp");
+        tmpFilePath = tmpFilePath.replaceAll("_task", "");
+        Path tmpPath = new Path(tmpFilePath, "vineyard.tmp");
+        try {
+            fs = finalOutPath.getFileSystem(jc);
+            System.out.println("tmp path:" + tmpPath.toString());
+            FSDataOutputStream output = FileSystem.create(fs, tmpPath, new FsPermission("777"));
+            if (output != null) {
+                System.out.println("Create succeed!");
+                output.write("test".getBytes(), 0, 4);
+                output.close();
+            }
+            // System.in.read();
+        } catch (Exception e) {
+            System.out.println("Create failed!");
+        }
+        // index = location.length() + 1;
+        // for (int i = 0; i < partitionCount; i++) {
+        //     index = finalOutPath.toString().indexOf("/", index + 1);
+        // }
+        // tableName = finalOutPath.toString().substring(0, index);
+        // tableName = tableName.replace('/', '#');
+        // System.out.println("Table name:" + tableName);
+        // System.out.println("fina path:" + finalOutPath.toString());
 
     }
 
@@ -161,6 +223,7 @@
     public void write(Writable w) throws IOException {
         System.out.println("write");
         if (w == null) {
+            System.out.println("w is null");
             return;
         }
         ArrowWrapperWritable arrowWrapperWritable = (ArrowWrapperWritable) w;
@@ -181,7 +244,9 @@
         System.out.println("close");
         System.out.println("table name:" + tableName);
         Table oldTable = null;
-        if (!hasData) {
+        System.out.println("has data:" + hasData);
+        if (schemaBuilder == null) {
+            System.out.println("No data to write.");
             client.disconnect();
             System.out.println("Bye, vineyard!");
             return;
@@ -217,6 +282,7 @@
             throw new IOException("Seal TableBuilder failed");
         }
         client.disconnect();
+
         System.out.println("Bye, vineyard!");
     }
 
@@ -234,62 +300,62 @@
         recordBatchBuilders.add(recordBatchBuilder);
     }
 
     private void fillColumns(VectorSchemaRoot root, Schema schema)
         throws VineyardException {
         for (int i = 0; i < schema.getFields().size(); i++) {
             val column = recordBatchBuilder.getColumnBuilder(i);
             Field field = schema.getFields().get(i);
             if (field.getType().equals(Arrow.Type.Boolean)) {
                 BitVector vector = (BitVector) root.getFieldVectors().get(i);
                 for (int j = 0; j < root.getRowCount(); j++) {
                     if (vector.get(j) != 0) {
                         column.setBoolean(j, true);
                     } else {
                         column.setBoolean(j, false);
                     }
                 }
             } else if (field.getType().equals(Arrow.Type.Int)) {
                 IntVector vector= (IntVector) root.getFieldVectors().get(i);
                 // System.out.println("row count:" + root.getRowCount());
                 for (int j = 0; j < root.getRowCount(); j++) {
                     column.setInt(j, vector.get(j));
                     // System.out.println("int value:" + vector.get(j));
                 }
             } else if (field.getType().equals(Arrow.Type.Int64)) {
                 BigIntVector vector = (BigIntVector) root.getFieldVectors().get(i);
                 for (int j = 0; j < root.getRowCount(); j++) {
                     column.setLong(j, vector.get(j));
                 }
             } else if (field.getType().equals(Arrow.Type.Float)) {
                 Float4Vector vector = (Float4Vector) root.getFieldVectors().get(i);
                 for (int j = 0; j < root.getRowCount(); j++) {
                     column.setFloat(j, vector.get(j));
                 }
             } else if (field.getType().equals(Arrow.Type.Double)) {
                 Float8Vector vector = (Float8Vector) root.getFieldVectors().get(i);
                 for (int j = 0; j < root.getRowCount(); j++) {
                     column.setDouble(j, vector.get(j));
                 } 
             } else if (field.getType().equals(Arrow.Type.LargeVarChar)) {
                 LargeVarCharVector vector = (LargeVarCharVector) root.getFieldVectors().get(i);
                 for (int j = 0; j < root.getRowCount(); j++) {
                     column.setUTF8String(j, vector.getObject(j));
                 } 
             } else if (field.getType().equals(Arrow.Type.VarChar)) {
                 VarCharVector vector = (VarCharVector) root.getFieldVectors().get(i);
                 for (int j = 0; j < root.getRowCount(); j++) {
                     column.setUTF8String(j, vector.getObject(j));
                 } 
             } else {
                 throw new VineyardException.NotImplemented(
                         "array builder for type " + field.getType() + " is not supported");
             }
         }
     }
 }
 
 class MapredRecordWriter<K extends NullWritable, V extends ArrowWrapperWritable>
         implements RecordWriter<K, V> {
     MapredRecordWriter() throws IOException {
         System.out.printf("creating vineyard record writer\n");
         throw new RuntimeException("mapred record writter: unimplemented");
@@ -305,3 +371,12 @@
         System.out.printf("closing\n");
     }
 }
+
+// file:/opt/hive/data/warehouse/hive_dynamic_partition_test6/.hive-staging_hive_2023-07-19_10-49-12_537_8312530081206971822-1/_task_tmp.-ext-10000/year=2017/_tmp.000000_0
+// file:/opt/hive/data/warehouse/hive_dynamic_partition_test6/.hive-staging_hive_2023-07-19_10-49-12_537_8312530081206971822-1/-ext-10000
+
+// file:/opt/hive/data/warehouse/hive_dynamic_partition_test6/year=2018/.hive-staging_hive_2023-07-19_10-52-50_144_3570501016920325767-1/_task_tmp.-ext-10000/_tmp.000000_0
+// file:/opt/hive/data/warehouse/hive_dynamic_partition_test6/year=2018/.hive-staging_hive_2023-07-19_10-52-50_144_3570501016920325767-1/-ext-10000
+
+//file:/opt/hive/data/warehouse/hive_dynamic_partition_test6/.hive-staging_hive_2023-07-19_11-15-58_433_128099267613906011-1/-ext-10000
+//file:/opt/hive/data/warehouse/hive_dynamic_partition_test7/.hive-staging_hive_2023-07-19_11-14-48_835_1857436151368976840-1/-ext-10000