Enhance hive writer (#1438)

* enhancement hive writer

* update

* update

Author: Yancey0623

pkg/sqlfs/hive_writer.go

@@ -65,6 +65,11 @@ func uploadCSVFile(csv *os.File, db *sql.DB, hivePath, table, user, passwd strin
 		if _, e := cmd.CombinedOutput(); e != nil {
 			return fmt.Errorf("failed %s: %v", cmd, e)
 		}
+		defer func() {
+			cmd = exec.Command("hdfs", "dfs", "-rm", "-r", "-f", hdfsPath)
+			cmd.Env = hdfsEnv
+			cmd.CombinedOutput()
+		}()
 
 		cmd = exec.Command("hdfs", "dfs", "-copyFromLocal", csv.Name(), hdfsPath)
 		cmd.Env = hdfsEnv

pkg/sqlfs/hive_writer_test.go

@@ -17,12 +17,23 @@ import (
 	"fmt"
 	"io"
 	"math/rand"
+	"os/exec"
+	"path"
 	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 )
 
+func hasHDFSDir(hdfsPath string) bool {
+	cmd := exec.Command("hdfs", "dfs", "-ls", hdfsPath)
+	out, _ := cmd.CombinedOutput()
+	if strings.Contains(string(out), "No such file or directory") {
+		return false
+	}
+	return true
+}
+
 func TestNewHiveWriter(t *testing.T) {
 	testDriver = getEnv("SQLFLOW_TEST_DB", "hive")
 	if testDriver != "hive" {
@@ -36,12 +47,13 @@ func TestNewHiveWriter(t *testing.T) {
 	w, e := newHiveWriter(testDB, "/hivepath", tbl, "", "", bufSize)
 	a.NoError(e)
 	a.NotNil(w)
-	defer w.Close()
 
 	has, e1 := hasTable(testDB, tbl)
 	a.NoError(e1)
 	a.True(has)
 
+	a.NoError(w.Close())
+	a.False(hasHDFSDir(path.Join("/hivepath", tbl)))
 	a.NoError(dropTable(testDB, tbl))
 }