[SPARK-32388][SQL] TRANSFORM with schema-less mode should keep the same with hive

sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala

@@ -111,15 +111,14 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
  .zip(outputFieldWriters)
  .map { case (data, writer) => writer(data) })
  } else {
- // In schema less mode, hive default serde will choose first two output column as output
- // if output column size less then 2, it will throw ArrayIndexOutOfBoundsException.
- // Here we change spark's behavior same as hive's default serde.
- // But in hive, TRANSFORM with schema less behavior like origin spark, we will fix this
- // to keep spark and hive behavior same in SPARK-32388
+ // In schema less mode, hive will choose first two output column as output.
+ // If output column size less then 2, it will return NULL for columns with missing values.
+ // Here we split row string and choose first 2 values, if values's size less then 2,
+ // we pad NULL value until 2 to make behavior same with hive.
  val kvWriter = CatalystTypeConverters.createToCatalystConverter(StringType)
  prevLine: String =>
  new GenericInternalRow(
- prevLine.split(outputRowFormat).slice(0, 2)
+ prevLine.split(outputRowFormat).slice(0, 2).padTo(2, null)
  .map(kvWriter))
  }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala

@@ -785,7 +785,9 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder(conf) {
  // Use default (serde) format.
  val name = conf.getConfString("hive.script.serde",
  "org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe")
- val props = Seq("field.delim" -> "\t")
+ val props = Seq(
+ "field.delim" -> "\t",
+ "serialization.last.column.takes.rest" -> "true")
  val recordHandler = Option(conf.getConfString(configKey, defaultConfigValue))
  (Nil, Option(name), props, recordHandler)
  }

sql/core/src/test/scala/org/apache/spark/sql/execution/BaseScriptTransformationSuite.scala

@@ -137,10 +137,7 @@ abstract class BaseScriptTransformationSuite extends SparkPlanTest with SQLTestU
  }
  }
 
- test("SPARK-25990: TRANSFORM should handle schema less correctly (no serde)") {
- assume(TestUtils.testCommandAvailable("python"))
- val scriptFilePath = copyAndGetResourceFile("test_script.py", ".py").getAbsoluteFile
-
+ test("SPARK-32388: TRANSFORM should handle schema less correctly (no serde)") {
  withTempView("v") {
  val df = Seq(
  (1, "1", 1.0, BigDecimal(1.0), new Timestamp(1)),
@@ -157,7 +154,24 @@ abstract class BaseScriptTransformationSuite extends SparkPlanTest with SQLTestU
  df.col("c").expr,
  df.col("d").expr,
  df.col("e").expr),
- script = s"python $scriptFilePath",
+ script = "cat",
+ output = Seq(
+ AttributeReference("key", StringType)(),
+ AttributeReference("value", StringType)()),
+ child = child,
+ ioschema = defaultIOSchema.copy(schemaLess = true)
+ ),
+ df.select(
+ 'a.cast("string").as("key"),
+ 'b.cast("string").as("value")).collect())
+
+ checkAnswer(
+ df,
+ (child: SparkPlan) => createScriptTransformationExec(
+ input = Seq(
+ df.col("a").expr,
+ df.col("b").expr),
+ script = "cat",
  output = Seq(
  AttributeReference("key", StringType)(),
  AttributeReference("value", StringType)()),
@@ -167,6 +181,22 @@ abstract class BaseScriptTransformationSuite extends SparkPlanTest with SQLTestU
  df.select(
  'a.cast("string").as("key"),
  'b.cast("string").as("value")).collect())
+
+ checkAnswer(
+ df,
+ (child: SparkPlan) => createScriptTransformationExec(
+ input = Seq(
+ df.col("a").expr),
+ script = "cat",
+ output = Seq(
+ AttributeReference("key", StringType)(),
+ AttributeReference("value", StringType)()),
+ child = child,
+ ioschema = defaultIOSchema.copy(schemaLess = true)
+ ),
+ df.select(
+ 'a.cast("string").as("key"),
+ lit(null)).collect())
  }
  }
 

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationSuite.scala

@@ -156,10 +156,7 @@ class HiveScriptTransformationSuite extends BaseScriptTransformationSuite with T
  assert(uncaughtExceptionHandler.exception.isEmpty)
  }
 
- test("SPARK-25990: TRANSFORM should handle schema less correctly (hive serde)") {
- assume(TestUtils.testCommandAvailable("python"))
- val scriptFilePath = copyAndGetResourceFile("test_script.py", ".py").getAbsolutePath
-
+ test("SPARK-32388: TRANSFORM should handle schema less correctly (hive serde)") {
  withTempView("v") {
  val df = Seq(
  (1, "1", 1.0, BigDecimal(1.0), new Timestamp(1)),
@@ -168,21 +165,157 @@ class HiveScriptTransformationSuite extends BaseScriptTransformationSuite with T
  ).toDF("a", "b", "c", "d", "e") // Note column d's data type is Decimal(38, 18)
  df.createTempView("v")
 
- val query = sql(
- s"""
- |SELECT TRANSFORM(a, b, c, d, e)
- |USING 'python ${scriptFilePath}'
- |FROM v
- """.stripMargin)
+ // In hive default serde mode, if we don't define output schema,
+ // when output column size > 2 and don't specify serde,
+ // it will choose take rest columns in second column as output schema
+ // (key: String, value: String)
+ checkAnswer(
+ sql(
+ s"""
+ |SELECT TRANSFORM(a, b, c, d, e)
+ | USING 'cat'
+ |FROM v
+ """.stripMargin),
+ identity,
+ df.select(
+ 'a.cast("string").as("key"),
+ concat_ws("\t",
+ 'b.cast("string"),
+ 'c.cast("string"),
+ 'd.cast("string"),
+ 'e.cast("string")).as("value")).collect())
+
+ // In hive default serde mode, if we don't define output schema,
+ // when output column size > 2 and just specify serde,
+ // it will choose take rest columns in second column as output schema
+ // (key: String, value: String)
+ checkAnswer(
+ sql(
+ s"""
+ |SELECT TRANSFORM(a, b, c, d, e)
+ | ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ | WITH SERDEPROPERTIES (
+ | 'field.delim' = '\t'
+ | )
+ | USING 'cat'
+ | ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ | WITH SERDEPROPERTIES (
+ | 'field.delim' = '\t'
+ | )
+ |FROM v
+ """.stripMargin),
+ identity,
+ df.select(
+ 'a.cast("string").as("key"),
+ 'b.cast("string").as("value")).collect())
+
+
+ // In hive default serde mode, if we don't define output schema,
+ // when output column size > 2 and specify serde with
+ // 'serialization.last.column.takes.rest=true',
+ // it will choose take rest columns in second column as output schema
+ // (key: String, value: String)
+ checkAnswer(
+ sql(
+ s"""
+ |SELECT TRANSFORM(a, b, c, d, e)
+ | ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ | WITH SERDEPROPERTIES (
+ | 'field.delim' = '\t',
+ | 'serialization.last.column.takes.rest' = 'true'
+ | )
+ | USING 'cat'
+ | ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ | WITH SERDEPROPERTIES (
+ | 'field.delim' = '\t',
+ | 'serialization.last.column.takes.rest' = 'true'
+ | )
+ |FROM v
+ """.stripMargin),
+ identity,
+ df.select(
+ 'a.cast("string").as("key"),
+ concat_ws("\t",
+ 'b.cast("string"),
+ 'c.cast("string"),
+ 'd.cast("string"),
+ 'e.cast("string")).as("value")).collect())
+
+ // In hive default serde mode, if we don't define output schema,
+ // when output column size > 2 and specify serde
+ // with 'serialization.last.column.takes.rest=false',
+ // it will choose first two column as output schema (key: String, value: String)
+ checkAnswer(
+ sql(
+ s"""
+ |SELECT TRANSFORM(a, b, c, d, e)
+ | ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ | WITH SERDEPROPERTIES (
+ | 'field.delim' = '\t',
+ | 'serialization.last.column.takes.rest' = 'false'
+ | )
+ | USING 'cat'
+ | ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ | WITH SERDEPROPERTIES (
+ | 'field.delim' = '\t',
+ | 'serialization.last.column.takes.rest' = 'false'
+ | )
+ |FROM v
+ """.stripMargin),
+ identity,
+ df.select(
+ 'a.cast("string").as("key"),
+ 'b.cast("string").as("value")).collect())
 
- // In hive default serde mode, if we don't define output schema, it will choose first
- // two column as output schema (key: String, value: String)
+ // In hive default serde mode, if we don't define output schema,
+ // when output column size = 2 and specify serde, it will these two column as
+ // output schema (key: String, value: String)
  checkAnswer(
- query,
+ sql(
+ s"""
+ |SELECT TRANSFORM(a, b)
+ | ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ | WITH SERDEPROPERTIES (
+ | 'field.delim' = '\t',
+ | 'serialization.last.column.takes.rest' = 'true'
+ | )
+ | USING 'cat'
+ | ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ | WITH SERDEPROPERTIES (
+ | 'field.delim' = '\t',
+ | 'serialization.last.column.takes.rest' = 'true'
+ | )
+ |FROM v
+ """.stripMargin),
  identity,
  df.select(
  'a.cast("string").as("key"),
  'b.cast("string").as("value")).collect())
+
+ // In hive default serde mode, if we don't define output schema,
+ // when output column size < 2 and specify serde, it will return null for deficiency
+ // output schema (key: String, value: String)
+ checkAnswer(
+ sql(
+ s"""
+ |SELECT TRANSFORM(a)
+ | ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ | WITH SERDEPROPERTIES (
+ | 'field.delim' = '\t',
+ | 'serialization.last.column.takes.rest' = 'true'
+ | )
+ | USING 'cat'
+ | ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
+ | WITH SERDEPROPERTIES (
+ | 'field.delim' = '\t',
+ | 'serialization.last.column.takes.rest' = 'true'
+ | )
+ |FROM v
+ """.stripMargin),
+ identity,
+ df.select(
+ 'a.cast("string").as("key"),
+ lit(null)).collect())
  }
  }