This is an automated email from the ASF dual-hosted git repository.
gengliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 8502cb37b57 [SPARK-38795][SQL] Support INSERT INTO user specified
column lists with DEFAULT values
8502cb37b57 is described below
commit 8502cb37b574c4c9cfb4711ce399ffa873f48825
Author: Daniel Tenedorio <daniel.tenedo...@databricks.com>
AuthorDate: Mon Apr 11 15:00:16 2022 +0800
[SPARK-38795][SQL] Support INSERT INTO user specified column lists with
DEFAULT values
### What changes were proposed in this pull request?
Support INSERT INTO commands with user specified column lists with DEFAULT
values.
For example:
```
CREATE TABLE t (x INT DEFAULT 42, y STRING DEFAULT "abc") USING PARQUET;
INSERT INTO t (y) VALUES ("def");
SELECT x, y FROM t;
> 42, "def"
```
### Why are the changes needed?
This PR allows INSERT INTO commands with explicit user specified column
lists to benefit from DEFAULT column support, in addition to those without.
This expands the functionality provided by this feature.
### Does this PR introduce _any_ user-facing change?
Yes, more types of INSERT INTO commands may use DEFAULT values.
### How was this patch tested?
This PR adds additional unit test coverage.
Closes #36077 from dtenedor/insert-into-defaults-explicit-col-lists.
Authored-by: Daniel Tenedorio <daniel.tenedo...@databricks.com>
Signed-off-by: Gengliang Wang <gengli...@apache.org>
---
.../catalyst/analysis/ResolveDefaultColumns.scala | 144 +++++++++++-----
.../org/apache/spark/sql/sources/InsertSuite.scala | 184 +++++++++++++++++++--
2 files changed, 269 insertions(+), 59 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala
index f4502c92457..447769be300 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala
@@ -53,25 +53,53 @@ case class ResolveDefaultColumns(
// This field stores the enclosing INSERT INTO command, once we find one.
var enclosingInsert: Option[InsertIntoStatement] = None
+ // This field stores the schema of the target table of the above command.
+ var insertTableSchemaWithoutPartitionColumns: Option[StructType] = None
- override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(
- (_ => SQLConf.get.enableDefaultColumns), ruleId) {
- case i@InsertIntoStatement(_, _, _, _, _, _)
- if i.query.collectFirst { case u: UnresolvedInlineTable => u }.isDefined
=>
- enclosingInsert = Some(i)
- i
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ // Initialize by clearing our reference to the enclosing INSERT INTO
command.
+ enclosingInsert = None
+ insertTableSchemaWithoutPartitionColumns = None
+ // Traverse the logical query plan in preorder (top-down).
+ plan.resolveOperatorsWithPruning(
+ (_ => SQLConf.get.enableDefaultColumns), ruleId) {
+ case i@InsertIntoStatement(_, _, _, _, _, _)
+ if i.query.collectFirst { case u: UnresolvedInlineTable
+ if u.rows.nonEmpty && u.rows.forall(_.size == u.rows(0).size) => u
+ }.isDefined =>
+ enclosingInsert = Some(i)
+ insertTableSchemaWithoutPartitionColumns =
getInsertTableSchemaWithoutPartitionColumns
+ val regenerated: InsertIntoStatement = regenerateUserSpecifiedCols(i)
+ regenerated
+
+ case table: UnresolvedInlineTable
+ if enclosingInsert.isDefined =>
+ val expanded: UnresolvedInlineTable =
addMissingDefaultColumnValues(table).getOrElse(table)
+ val replaced: LogicalPlan =
+ replaceExplicitDefaultColumnValues(analyzer,
expanded).getOrElse(table)
+ replaced
+
+ case i@InsertIntoStatement(_, _, _, project: Project, _, _) =>
+ enclosingInsert = Some(i)
+ insertTableSchemaWithoutPartitionColumns =
getInsertTableSchemaWithoutPartitionColumns
+ val expanded: Project =
addMissingDefaultColumnValues(project).getOrElse(project)
+ val replaced: Option[LogicalPlan] =
replaceExplicitDefaultColumnValues(analyzer, expanded)
+ val updated: InsertIntoStatement =
+ if (replaced.isDefined) i.copy(query = replaced.get) else i
+ val regenerated: InsertIntoStatement =
regenerateUserSpecifiedCols(updated)
+ regenerated
+ }
+ }
- case table: UnresolvedInlineTable
- if enclosingInsert.isDefined &&
- table.rows.nonEmpty && table.rows.forall(_.size == table.rows(0).size)
=>
- val expanded: UnresolvedInlineTable =
addMissingDefaultColumnValues(table).getOrElse(table)
- replaceExplicitDefaultColumnValues(analyzer, expanded).getOrElse(table)
-
- case i@InsertIntoStatement(_, _, _, project: Project, _, _) =>
- enclosingInsert = Some(i)
- val expanded: Project =
addMissingDefaultColumnValues(project).getOrElse(project)
- val replaced: Option[LogicalPlan] =
replaceExplicitDefaultColumnValues(analyzer, expanded)
- if (replaced.isDefined) i.copy(query = replaced.get) else i
+ // Helper method to regenerate user-specified columns of an
InsertIntoStatement based on the names
+ // in the insertTableSchemaWithoutPartitionColumns field of this class.
+ private def regenerateUserSpecifiedCols(i: InsertIntoStatement):
InsertIntoStatement = {
+ if (i.userSpecifiedCols.nonEmpty &&
insertTableSchemaWithoutPartitionColumns.isDefined) {
+ i.copy(
+ userSpecifiedCols =
insertTableSchemaWithoutPartitionColumns.get.fields.map(_.name))
+ } else {
+ i
+ }
}
// Helper method to check if an expression is an explicit DEFAULT column
reference.
@@ -87,7 +115,7 @@ case class ResolveDefaultColumns(
table: UnresolvedInlineTable): Option[UnresolvedInlineTable] = {
assert(enclosingInsert.isDefined)
val numQueryOutputs: Int = table.rows(0).size
- val schema = getInsertTableSchemaWithoutPartitionColumns.getOrElse(return
None)
+ val schema = insertTableSchemaWithoutPartitionColumns.getOrElse(return
None)
val newDefaultExpressions: Seq[Expression] =
getDefaultExpressions(numQueryOutputs, schema)
val newNames: Seq[String] = schema.fields.drop(numQueryOutputs).map {
_.name }
if (newDefaultExpressions.nonEmpty) {
@@ -104,7 +132,7 @@ case class ResolveDefaultColumns(
*/
private def addMissingDefaultColumnValues(project: Project): Option[Project]
= {
val numQueryOutputs: Int = project.projectList.size
- val schema = getInsertTableSchemaWithoutPartitionColumns.getOrElse(return
None)
+ val schema = insertTableSchemaWithoutPartitionColumns.getOrElse(return
None)
val newDefaultExpressions: Seq[Expression] =
getDefaultExpressions(numQueryOutputs, schema)
val newAliases: Seq[NamedExpression] =
newDefaultExpressions.zip(schema.fields).map {
@@ -122,16 +150,21 @@ case class ResolveDefaultColumns(
*/
private def getDefaultExpressions(numQueryOutputs: Int, schema: StructType):
Seq[Expression] = {
val remainingFields: Seq[StructField] = schema.fields.drop(numQueryOutputs)
- val numDefaultExpressionsToAdd: Int = {
- if (SQLConf.get.useNullsForMissingDefaultColumnValues) {
- remainingFields.size
- } else {
-
remainingFields.takeWhile(_.metadata.contains(CURRENT_DEFAULT_COLUMN_METADATA_KEY)).size
- }
- }
+ val numDefaultExpressionsToAdd =
getStructFieldsForDefaultExpressions(remainingFields).size
Seq.fill(numDefaultExpressionsToAdd)(UnresolvedAttribute(CURRENT_DEFAULT_COLUMN_NAME))
}
+ /**
+ * This is a helper for the getDefaultExpressions methods above.
+ */
+ private def getStructFieldsForDefaultExpressions(fields: Seq[StructField]):
Seq[StructField] = {
+ if (SQLConf.get.useNullsForMissingDefaultColumnValues) {
+ fields
+ } else {
+
fields.takeWhile(_.metadata.contains(CURRENT_DEFAULT_COLUMN_METADATA_KEY))
+ }
+ }
+
/**
* Replaces unresolved DEFAULT column references with corresponding values
in a logical plan.
*/
@@ -139,7 +172,7 @@ case class ResolveDefaultColumns(
analyzer: Analyzer,
input: LogicalPlan): Option[LogicalPlan] = {
assert(enclosingInsert.isDefined)
- val schema = getInsertTableSchemaWithoutPartitionColumns.getOrElse(return
None)
+ val schema = insertTableSchemaWithoutPartitionColumns.getOrElse(return
None)
val columnNames: Seq[String] = schema.fields.map { _.name }
val defaultExpressions: Seq[Expression] = schema.fields.map {
case f if f.metadata.contains(CURRENT_DEFAULT_COLUMN_METADATA_KEY) =>
@@ -193,7 +226,7 @@ case class ResolveDefaultColumns(
}
/**
- * Replaces unresolved DEFAULT column references with corresponding values
in an projection.
+ * Replaces unresolved DEFAULT column references with corresponding values
in a projection.
*/
private def replaceExplicitDefaultColumnValues(
defaultExpressions: Seq[Expression],
@@ -230,21 +263,48 @@ case class ResolveDefaultColumns(
*/
private def getInsertTableSchemaWithoutPartitionColumns: Option[StructType]
= {
assert(enclosingInsert.isDefined)
- try {
- val tableName = enclosingInsert.get.table match {
- case r: UnresolvedRelation => TableIdentifier(r.name)
- case r: UnresolvedCatalogRelation => r.tableMeta.identifier
- case _ => return None
- }
- val lookup = catalog.lookupRelation(tableName)
- lookup match {
- case SubqueryAlias(_, r: UnresolvedCatalogRelation) =>
- Some(StructType(r.tableMeta.schema.fields.dropRight(
- enclosingInsert.get.partitionSpec.size)))
- case _ => None
- }
+ val tableName = enclosingInsert.get.table match {
+ case r: UnresolvedRelation => TableIdentifier(r.name)
+ case r: UnresolvedCatalogRelation => r.tableMeta.identifier
+ case _ => return None
+ }
+ // Lookup the relation from the catalog by name. This either succeeds or
returns some "not
+ // found" error. In the latter cases, return out of this rule without
changing anything and let
+ // the analyzer return a proper error message elsewhere.
+ val lookup: LogicalPlan = try {
+ catalog.lookupRelation(tableName)
} catch {
- case _: NoSuchTableException => None
+ case _: AnalysisException => return None
+ }
+ val schema: StructType = lookup match {
+ case SubqueryAlias(_, r: UnresolvedCatalogRelation) =>
+ StructType(r.tableMeta.schema.fields.dropRight(
+ enclosingInsert.get.partitionSpec.size))
+ case _ => return None
+ }
+ // Rearrange the columns in the result schema to match the order of the
explicit column list,
+ // if any.
+ val userSpecifiedCols: Seq[String] = enclosingInsert.get.userSpecifiedCols
+ if (userSpecifiedCols.isEmpty) {
+ return Some(schema)
}
+ def normalize(str: String) = {
+ if (SQLConf.get.caseSensitiveAnalysis) str else str.toLowerCase()
+ }
+ val colNamesToFields: Map[String, StructField] =
+ schema.fields.map {
+ field: StructField => normalize(field.name) -> field
+ }.toMap
+ val userSpecifiedFields: Seq[StructField] =
+ userSpecifiedCols.map {
+ name: String => colNamesToFields.getOrElse(normalize(name), return
None)
+ }
+ val userSpecifiedColNames: Set[String] = userSpecifiedCols.toSet
+ val nonUserSpecifiedFields: Seq[StructField] =
+ schema.fields.filter {
+ field => !userSpecifiedColNames.contains(field.name)
+ }
+ Some(StructType(userSpecifiedFields ++
+ getStructFieldsForDefaultExpressions(nonUserSpecifiedFields)))
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
index 2483055880e..222e195719d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
@@ -865,26 +865,26 @@ class InsertSuite extends DataSourceTest with
SharedSparkSession {
withTable("t") {
sql("create table t(i boolean, s bigint) using parquet")
sql("insert into t values(true)")
- checkAnswer(sql("select s from t where i = true"), Seq(Row(null)))
+ checkAnswer(spark.table("t"), Row(true, null))
}
}
// The default value for the DEFAULT keyword is the NULL literal.
withTable("t") {
sql("create table t(i boolean, s bigint) using parquet")
sql("insert into t values(true, default)")
- checkAnswer(sql("select s from t where i = true"), Seq(null).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(true, null))
}
// There is a complex expression in the default value.
withTable("t") {
sql("create table t(i boolean, s string default concat('abc', 'def'))
using parquet")
sql("insert into t values(true, default)")
- checkAnswer(sql("select s from t where i = true"), Seq("abcdef").map(i
=> Row(i)))
+ checkAnswer(spark.table("t"), Row(true, "abcdef"))
}
// The default value parses correctly and the provided value type is
different but coercible.
withTable("t") {
sql("create table t(i boolean, s bigint default 42) using parquet")
sql("insert into t values(false)")
- checkAnswer(sql("select s from t where i = false"), Seq(42L).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(false, 42L))
}
// There are two trailing default values referenced implicitly by the
INSERT INTO statement.
withTable("t") {
@@ -894,74 +894,74 @@ class InsertSuite extends DataSourceTest with
SharedSparkSession {
}
// The table has a partitioning column and a default value is injected.
withTable("t") {
- sql("create table t(i boolean, s bigint, q int default 42 ) using
parquet partitioned by (i)")
+ sql("create table t(i boolean, s bigint, q int default 42) using parquet
partitioned by (i)")
sql("insert into t partition(i='true') values(5, default)")
- checkAnswer(sql("select s from t where i = true"), Seq(5).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(5, 42, true))
}
// The table has a partitioning column and a default value is added per an
explicit reference.
withTable("t") {
sql("create table t(i boolean, s bigint default 42) using parquet
partitioned by (i)")
sql("insert into t partition(i='true') values(default)")
- checkAnswer(sql("select s from t where i = true"), Seq(42L).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(42L, true))
}
// The default value parses correctly as a constant but non-literal
expression.
withTable("t") {
sql("create table t(i boolean, s bigint default 41 + 1) using parquet")
sql("insert into t values(false, default)")
- checkAnswer(sql("select s from t where i = false"), Seq(42L).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(false, 42L))
}
// Explicit defaults may appear in different positions within the inline
table provided as input
// to the INSERT INTO statement.
withTable("t") {
sql("create table t(i boolean default false, s bigint default 42) using
parquet")
sql("insert into t values(false, default), (default, 42)")
- checkAnswer(sql("select s from t where i = false"), Seq(42L, 42L).map(i
=> Row(i)))
+ checkAnswer(spark.table("t"), Seq(Row(false, 42L), Row(false, 42L)))
}
// There is an explicit default value provided in the INSERT INTO
statement in the VALUES,
// with an alias over the VALUES.
withTable("t") {
sql("create table t(i boolean, s bigint default 42) using parquet")
sql("insert into t select * from values (false, default) as tab(col,
other)")
- checkAnswer(sql("select s from t where i = false"), Seq(42L).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(false, 42L))
}
// The explicit default value arrives first before the other value.
withTable("t") {
sql("create table t(i boolean default false, s bigint) using parquet")
sql("insert into t values (default, 43)")
- checkAnswer(sql("select s from t where i = false"), Seq(43L).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(false, 43L))
}
// The 'create table' statement provides the default parameter first.
withTable("t") {
sql("create table t(i boolean default false, s bigint) using parquet")
sql("insert into t values (default, 43)")
- checkAnswer(sql("select s from t where i = false"), Seq(43L).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(false, 43L))
}
// The explicit default value is provided in the wrong order (first
instead of second), but
// this is OK because the provided default value evaluates to literal NULL.
withTable("t") {
sql("create table t(i boolean, s bigint default 42) using parquet")
sql("insert into t values (default, 43)")
- checkAnswer(sql("select s from t where i is null"), Seq(43L).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(null, 43L))
}
// There is an explicit default value provided in the INSERT INTO
statement as a SELECT.
// This is supported.
withTable("t") {
sql("create table t(i boolean, s bigint default 42) using parquet")
sql("insert into t select false, default")
- checkAnswer(sql("select s from t where i = false"), Seq(42L).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(false, 42L))
}
// There is a complex query plan in the SELECT query in the INSERT INTO
statement.
withTable("t") {
sql("create table t(i boolean default false, s bigint default 42) using
parquet")
sql("insert into t select col, count(*) from values (default, default) "
+
"as tab(col, other) group by 1")
- checkAnswer(sql("select s from t where i = false"), Seq(1).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(false, 1))
}
// The explicit default reference resolves successfully with nested table
subqueries.
withTable("t") {
sql("create table t(i boolean default false, s bigint) using parquet")
sql("insert into t select * from (select * from values(default, 42))")
- checkAnswer(sql("select s from t where i = false"), Seq(42L).map(i =>
Row(i)))
+ checkAnswer(spark.table("t"), Row(false, 42L))
}
// There are three column types exercising various combinations of
implicit and explicit
// default column value references in the 'insert into' statements. Note
these tests depend on
@@ -1086,7 +1086,7 @@ class InsertSuite extends DataSourceTest with
SharedSparkSession {
}
// The table has a partitioning column with a default value; this is not
allowed.
withTable("t") {
- sql("create table t(i boolean default true, s bigint, q int default 42 )
" +
+ sql("create table t(i boolean default true, s bigint, q int default 42)
" +
"using parquet partitioned by (i)")
assert(intercept[ParseException] {
sql("insert into t partition(i=default) values(5, default)")
@@ -1105,6 +1105,156 @@ class InsertSuite extends DataSourceTest with
SharedSparkSession {
}
}
+ test("INSERT INTO with user specified columns and defaults: positive tests")
{
+ Seq(
+ "insert into t (i, s) values (true, default)",
+ "insert into t (s, i) values (default, true)",
+ "insert into t (i) values (true)",
+ "insert into t (i) values (default)",
+ "insert into t (s) values (default)",
+ "insert into t (s) select default from (select 1)",
+ "insert into t (i) select true from (select 1)"
+ ).foreach { insert =>
+ withTable("t") {
+ sql("create table t(i boolean default true, s bigint default 42) using
parquet")
+ sql(insert)
+ checkAnswer(spark.table("t"), Row(true, 42L))
+ }
+ }
+ // The table is partitioned and we insert default values with explicit
column names.
+ withTable("t") {
+ sql("create table t(i boolean, s bigint default 4, q int default 42)
using parquet " +
+ "partitioned by (i)")
+ sql("insert into t partition(i='true') (s) values(5)")
+ sql("insert into t partition(i='false') (q) select 43")
+ sql("insert into t partition(i='false') (q) select default")
+ checkAnswer(spark.table("t"),
+ Seq(Row(5, 42, true),
+ Row(4, 43, false),
+ Row(4, 42, false)))
+ }
+ // When the CASE_SENSITIVE configuration is disabled, then using different
cases for the
+ // required and provided column names is successful.
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ withTable("t") {
+ sql("create table t(i boolean, s bigint default 42, q int default 43)
using parquet")
+ sql("insert into t (I, Q) select true from (select 1)")
+ checkAnswer(spark.table("t"), Row(true, 42L, 43))
+ }
+ }
+ // When the USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES configuration is
enabled, and no
+ // explicit DEFAULT value is available when the INSERT INTO statement
provides fewer
+ // values than expected, NULL values are appended in their place.
+ withSQLConf(SQLConf.USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES.key ->
"true") {
+ withTable("t") {
+ sql("create table t(i boolean, s bigint) using parquet")
+ sql("insert into t (i) values (true)")
+ checkAnswer(spark.table("t"), Row(true, null))
+ }
+ withTable("t") {
+ sql("create table t(i boolean default true, s bigint) using parquet")
+ sql("insert into t (i) values (default)")
+ checkAnswer(spark.table("t"), Row(true, null))
+ }
+ withTable("t") {
+ sql("create table t(i boolean, s bigint default 42) using parquet")
+ sql("insert into t (s) values (default)")
+ checkAnswer(spark.table("t"), Row(null, 42L))
+ }
+ withTable("t") {
+ sql("create table t(i boolean, s bigint, q int) using parquet
partitioned by (i)")
+ sql("insert into t partition(i='true') (s) values(5)")
+ sql("insert into t partition(i='false') (q) select 43")
+ sql("insert into t partition(i='false') (q) select default")
+ checkAnswer(spark.table("t"),
+ Seq(Row(5, null, true),
+ Row(null, 43, false),
+ Row(null, null, false)))
+ }
+ }
+ }
+
+ test("INSERT INTO with user specified columns and defaults: negative tests")
{
+ val addOneColButExpectedTwo = "target table has 2 column(s) but the
inserted data has 1 col"
+ val addTwoColButExpectedThree = "target table has 3 column(s) but the
inserted data has 2 col"
+ // The missing columns in these INSERT INTO commands do not have explicit
default values.
+ withTable("t") {
+ sql("create table t(i boolean, s bigint) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t (i) values (true)")
+ }.getMessage.contains(addOneColButExpectedTwo))
+ }
+ withTable("t") {
+ sql("create table t(i boolean default true, s bigint) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t (i) values (default)")
+ }.getMessage.contains(addOneColButExpectedTwo))
+ }
+ withTable("t") {
+ sql("create table t(i boolean, s bigint default 42) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t (s) values (default)")
+ }.getMessage.contains(addOneColButExpectedTwo))
+ }
+ withTable("t") {
+ sql("create table t(i boolean, s bigint, q int default 43) using
parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t (i, q) select true from (select 1)")
+ }.getMessage.contains(addTwoColButExpectedThree))
+ }
+ // When the USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES configuration is
disabled, and no
+ // explicit DEFAULT value is available when the INSERT INTO statement
provides fewer
+ // values than expected, the INSERT INTO command fails to execute.
+ withSQLConf(SQLConf.USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES.key ->
"false") {
+ withTable("t") {
+ sql("create table t(i boolean, s bigint) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t (i) values (true)")
+ }.getMessage.contains(addOneColButExpectedTwo))
+ }
+ withTable("t") {
+ sql("create table t(i boolean default true, s bigint) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t (i) values (default)")
+ }.getMessage.contains(addOneColButExpectedTwo))
+ }
+ withTable("t") {
+ sql("create table t(i boolean, s bigint default 42) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t (s) values (default)")
+ }.getMessage.contains(addOneColButExpectedTwo))
+ }
+ withTable("t") {
+ sql("create table t(i boolean, s bigint, q int) using parquet
partitioned by (i)")
+ assert(intercept[AnalysisException] {
+ sql("insert into t partition(i='true') (s) values(5)")
+ }.getMessage.contains(addTwoColButExpectedThree))
+ }
+ withTable("t") {
+ sql("create table t(i boolean, s bigint, q int) using parquet
partitioned by (i)")
+ assert(intercept[AnalysisException] {
+ sql("insert into t partition(i='false') (q) select 43")
+ }.getMessage.contains(addTwoColButExpectedThree))
+ }
+ withTable("t") {
+ sql("create table t(i boolean, s bigint, q int) using parquet
partitioned by (i)")
+ assert(intercept[AnalysisException] {
+ sql("insert into t partition(i='false') (q) select default")
+ }.getMessage.contains(addTwoColButExpectedThree))
+ }
+ }
+ // When the CASE_SENSITIVE configuration is enabled, then using different
cases for the required
+ // and provided column names results in an analysis error.
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+ withTable("t") {
+ sql("create table t(i boolean default true, s bigint default 42) using
parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t (I) select true from (select 1)")
+ }.getMessage.contains("Cannot resolve column name I"))
+ }
+ }
+ }
+
test("Stop task set if FileAlreadyExistsException was thrown") {
Seq(true, false).foreach { fastFail =>
withSQLConf("fs.file.impl" ->
classOf[FileExistingTestFileSystem].getName,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
