[Feature][regression-test]CSV import and export support header (#8487)

regression-test/conf/regression-conf.groovy

@@ -41,4 +41,10 @@ testSuites = ""
 // empty directories will test all directories
 testDirectories = ""
 
-customConf1 = "test_custom_conf_value"
+customConf1 = "test_custom_conf_value"
+
+// for test csv with header
+hdfsFs = "hdfs://127.0.0.1:9000"
+hdfsUser = "palo-qa"
+hdfsPasswd = ""
+brokerName = "broker_name"

regression-test/data/demo/testheader/csv.txt

@@ -0,0 +1,10 @@
+2017-07-03,78,5,OlpJfkVz,176
+2017-07-03,73,18,iVM0NyAH,49
+2017-07-03,5,13,wvim4aqW,95
+2017-07-03,20,12,ZksFGmLv,172
+2017-07-03,39,15,Mf0RinDC,48
+2017-07-03,55,9,VGq4T2kt,10
+2017-07-03,50,12,p2TolhzU,149
+2017-07-03,22,17,Q8kWnJyU,122
+2017-07-03,54,16,upn9ZRPC,151
+2017-07-03,22,10,Nzp8B0L2,94

regression-test/data/demo/testheader/csv_with_names.txt

@@ -0,0 +1,11 @@
+event_day,siteid,citycode,username,pv
+2017-07-03,77,18,rFphH1sk,165
+2017-07-03,51,8,R47GKb1c,157
+2017-07-03,82,3,uD7bYnZK,62
+2017-07-03,78,8,Bf8K9rFd,122
+2017-07-03,38,20,A8JHqEfY,194
+2017-07-03,18,12,Emu786j5,84
+2017-07-03,43,10,yklMRVYJ,132
+2017-07-03,82,3,RvBK0g4o,118
+2017-07-03,0,18,EMuWnD0y,9
+2017-07-03,99,18,IjO9Hiof,132

regression-test/data/demo/testheader/csv_with_names_and_types.txt

@@ -0,0 +1,12 @@
+event_day,siteid,citycode,username,pv
+date,int,samllint,varchar,int
+2017-07-03,77,18,rFphH1sk,165
+2017-07-03,51,8,R47GKb1c,157
+2017-07-03,82,3,uD7bYnZK,62
+2017-07-03,78,8,Bf8K9rFd,122
+2017-07-03,38,20,A8JHqEfY,194
+2017-07-03,18,12,Emu786j5,84
+2017-07-03,43,10,yklMRVYJ,132
+2017-07-03,82,3,RvBK0g4o,118
+2017-07-03,0,18,EMuWnD0y,9
+2017-07-03,99,18,IjO9Hiof,1323

regression-test/framework/pom.xml

@@ -75,6 +75,7 @@ under the License.
         <groovy-eclipse-batch.version>3.0.7-01</groovy-eclipse-batch.version>
         <groovy-eclipse-compiler.version>3.7.0</groovy-eclipse-compiler.version>
         <antlr.version>4.9.3</antlr.version>
+        <hadoop.version>2.8.0</hadoop.version>
     </properties>
     <build>
         <plugins>
@@ -259,5 +260,20 @@ under the License.
             <artifactId>antlr4-runtime</artifactId>
             <version>${antlr.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-hdfs</artifactId>
+            <version>${hadoop.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>${hadoop.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-mapreduce-client-core</artifactId>
+            <version>${hadoop.version}</version>
+        </dependency>
     </dependencies>
 </project>

regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy

@@ -29,6 +29,7 @@ import org.apache.doris.regression.action.StreamLoadAction
 import org.apache.doris.regression.action.SuiteAction
 import org.apache.doris.regression.action.TestAction
 import org.apache.doris.regression.util.JdbcUtils
+import org.apache.doris.regression.util.Hdfs
 import org.junit.jupiter.api.Assertions
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -41,6 +42,8 @@ import java.util.stream.LongStream
 
 import static org.apache.doris.regression.util.DataUtils.sortByToString
 
+import java.io.File
+
 class Suite implements GroovyInterceptable {
     final SuiteContext context
     final String name
@@ -250,6 +253,66 @@ class Suite implements GroovyInterceptable {
         runAction(new TestAction(context), actionSupplier)
     }
 
+    String getBrokerName() {
+        String brokerName = context.config.otherConfigs.get("brokerName")
+        return brokerName
+    }
+
+    String getHdfsFs() {
+        String hdfsFs = context.config.otherConfigs.get("hdfsFs")
+        return hdfsFs
+    }
+
+    String getHdfsUser() {
+        String hdfsUser = context.config.otherConfigs.get("hdfsUser")
+        return hdfsUser
+    }
+
+    String getHdfsPasswd() {
+        String hdfsPasswd = context.config.otherConfigs.get("hdfsPasswd")
+        return hdfsPasswd
+    }
+
+    String getHdfsDataDir() {
+        String dataDir = context.config.dataPath + "/" + group + "/"
+        String hdfsFs = context.config.otherConfigs.get("hdfsFs")
+        String hdfsUser = context.config.otherConfigs.get("hdfsUser")
+        Hdfs hdfs = new Hdfs(hdfsFs, hdfsUser, dataDir)
+        return hdfs.genRemoteDataDir()
+    }
+
+    String uploadToHdfs(String localFile) {
+        String dataDir = context.config.dataPath + "/" + group + "/"
+        localFile = dataDir + localFile
+        String hdfsFs = context.config.otherConfigs.get("hdfsFs")
+        String hdfsUser = context.config.otherConfigs.get("hdfsUser")
+        Hdfs hdfs = new Hdfs(hdfsFs, hdfsUser, dataDir)
+        String remotePath = hdfs.upload(localFile)
+        return remotePath;
+    }
+
+    int getTotalLine(String filePath) {
+        def file = new File(filePath)
+        int lines = 0;
+        file.eachLine {
+            lines++;
+        }
+        return lines;
+    }
+
+    boolean deleteFile(String filePath) {
+        def file = new File(filePath)
+        file.delete()
+    }
+
+    List<String> downloadExportFromHdfs(String label) {
+        String dataDir = context.config.dataPath + "/" + group + "/"
+        String hdfsFs = context.config.otherConfigs.get("hdfsFs")
+        String hdfsUser = context.config.otherConfigs.get("hdfsUser")
+        Hdfs hdfs = new Hdfs(hdfsFs, hdfsUser, dataDir)
+        return hdfs.downLoad(label)
+    }
+
     void streamLoad(Closure actionSupplier) {
         runAction(new StreamLoadAction(context), actionSupplier)
     }

regression-test/framework/src/main/groovy/org/apache/doris/regression/util/Hdfs.groovy

@@ -0,0 +1,109 @@
+// 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 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.regression.util
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.junit.Assert;
+
+import java.io.FileOutputStream;
+import java.io.IOException;
+
+// upload file from  "regression-test/data/hdfs"  to  "hdfs://xx/user/${userName}/groovy/"
+// download file from "hdfs://xx/user/${userName}/groovy/" to  "regression-test/data/hdfs"
+class Hdfs {
+    FileSystem fs = null;
+    String uri;
+    String userName;
+    String testRemoteDir;
+    String localDataDir;
+
+    Hdfs(String uri, String username, String localDataDir) {
+        Configuration conf = new Configuration();
+        conf.setStrings("fs.default.name", uri);
+        this.uri = uri;
+        this.userName = username;
+        System.setProperty("HADOOP_USER_NAME", username);
+        conf.setBoolean("fs.hdfs.impl.disable.cache", true);
+        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
+        conf.set("fs.file.impl", "org.apache.hadoop.fs.LocalFileSystem");
+        this.testRemoteDir = genRemoteDataDir();
+        this.localDataDir = localDataDir;
+        try {
+            fs = FileSystem.get(conf);
+            Path remoteDirPath = new Path(testRemoteDir);
+            if (!fs.exists(remoteDirPath)) {
+                fs.mkdirs(remoteDirPath);
+            }
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    List<String> downLoad(String prefix) {
+        List<String> files = new ArrayList<>();
+        try {
+            String filepath = this.testRemoteDir + prefix + "*";
+            FileStatus[] fileStatusArray = fs.globStatus(new Path(filepath + "*"));
+            for (FileStatus fileStatus : fileStatusArray) {
+                Path path = fileStatus.getPath();
+                FSDataInputStream fsDataInputStream = fs.open(path);
+                String localFilePath = getAbsoluteLocalPath(prefix.split('/')[0] + path.getName())
+                FileOutputStream fileOutputStream = new FileOutputStream(localFilePath);
+                IOUtils.copy(fsDataInputStream, fileOutputStream);
+                files.add(localFilePath);
+            }
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return files;
+    }
+
+    String getAbsoluteLocalPath(String file_name) {
+        String localAbsolutePath = this.localDataDir + "/" + file_name;
+        return localAbsolutePath;
+    }
+
+    String genRemoteDataDir() {
+        return this.uri + "/user/" + this.userName + "/groovy/";
+    }
+
+    String getAbsoluteRemotePath(String file_name) {
+        String remoteAbsolutePath = genRemoteDataDir() + file_name;
+        return remoteAbsolutePath;
+    }
+
+    String upload(String local_file) {
+        try {
+            Path src = new Path(local_file);
+            String remote_file = getAbsoluteRemotePath(src.getName());
+            Path dst = new Path(remote_file);
+            fs.copyFromLocalFile(src, dst);
+            return remote_file;
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return "";
+    }
+}