fix(spark): sporadic failures in Hive test suite (#540)

The builds were sporadically failing with the error:
io.substrait.spark.HiveTableSuite > Insert into Hive table FAILED
    org.scalatest.exceptions.TestFailedException: Expected 4, but got 5

It turns out that the row values were not being correctly
serialised / deserialised into the Hive table correctly, so even in
the cases were the test passed (based on number of rows
added), the values inserted were still incorrect.
This commit fixes that, and adds an extra check in the test
case(s) to assert the correct values are inserted.

Signed-off-by: Andrew Coleman <andrew_coleman@uk.ibm.com>

Author: andrew-coleman

spark/src/main/scala/io/substrait/spark/logical/ToLogicalPlan.scala

@@ -470,7 +470,7 @@ class ToLogicalPlan(spark: SparkSession = SparkSession.builder().getOrCreate())
       case WriteOp.INSERT if isHive =>
         withChild(child) {
           InsertIntoHiveTable(
-            table,
+            catalogTable(write.getNames.asScala, ToSparkType.toStructType(write.getTableSchema)),
             Map.empty,
             child,
             write.getCreateMode == CreateMode.REPLACE_IF_EXISTS,
@@ -554,8 +554,8 @@ class ToLogicalPlan(spark: SparkSession = SparkSession.builder().getOrCreate())
     val loc = spark.conf.get(StaticSQLConf.WAREHOUSE_PATH.key)
     val storage = CatalogStorageFormat(
       locationUri = Some(URI.create(f"$loc/$table")),
-      inputFormat = None,
-      outputFormat = None,
+      inputFormat = Some("org.apache.hadoop.mapred.TextInputFormat"),
+      outputFormat = Some("org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat"),
       serde = None,
       compressed = false,
       properties = Map.empty

spark/src/test/scala/io/substrait/spark/HiveTableSuite.scala

@@ -95,6 +95,8 @@ class HiveTableSuite extends SparkFunSuite {
     // and again...
     spark.sessionState.executePlan(plan).executedPlan.execute()
     assertResult(5)(spark.sql("select * from test").count())
+    // check it has inserted the correct values
+    assertResult(3)(spark.sql("select * from test where ID = 1003 and VALUE = 'again'").count())
   }
 
   test("Create Table As Select") {

spark/src/test/scala/io/substrait/spark/LocalFiles.scala

@@ -218,6 +218,8 @@ class LocalFiles extends SharedSparkSession {
     // and again...
     spark.sessionState.executePlan(plan).executedPlan.execute()
     assertResult(5)(spark.sql("select * from test").count())
+    // check it has inserted the correct values
+    assertResult(3)(spark.sql("select * from test where ID = 1003 and VALUE = 'again'").count())
   }
 
   test("Append to CSV file") {