You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2022/03/31 06:31:14 UTC
[spark] branch master updated: [SPARK-38336][SQL] Support INSERT INTO commands into tables with DEFAULT columns
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 a364cc0 [SPARK-38336][SQL] Support INSERT INTO commands into tables with DEFAULT columns
a364cc0 is described below
commit a364cc01716279d19eb5b43bde4961b04f5102af
Author: Daniel Tenedorio <da...@databricks.com>
AuthorDate: Thu Mar 31 14:29:58 2022 +0800
[SPARK-38336][SQL] Support INSERT INTO commands into tables with DEFAULT columns
### What changes were proposed in this pull request?
Extend INSERT INTO statements to support omitting default values or referring to them explicitly with the DEFAULT keyword, in which case the Spark analyzer will automatically insert the appropriate corresponding values in the right places.
Example:
```
CREATE TABLE T(a INT DEFAULT 4, b INT NOT NULL DEFAULT 5);
INSERT INTO T VALUES (1);
INSERT INTO T VALUES (1, DEFAULT);
INSERT INTO T VALUES (DEFAULT, 6);
SELECT * FROM T;
(1, 5)
(1, 5)
(4, 6)
```
### Why are the changes needed?
This helps users issue INSERT INTO statements with less effort, and helps people creating or updating tables to add custom optional columns for use in specific circumstances as desired.
### How was this patch tested?
This change is covered by new and existing unit test coverage as well as new INSERT INTO query test cases covering a variety of positive and negative scenarios.
Closes #35982 from dtenedor/default-columns-insert-into.
Authored-by: Daniel Tenedorio <da...@databricks.com>
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
.../spark/sql/catalyst/parser/SqlBaseParser.g4 | 1 +
.../spark/sql/catalyst/analysis/Analyzer.scala | 1 +
.../catalyst/analysis/ResolveDefaultColumns.scala | 250 ++++++++++++++++++++
.../spark/sql/catalyst/parser/AstBuilder.scala | 5 +
.../sql/catalyst/rules/RuleIdCollection.scala | 1 +
...lumns.scala => ResolveDefaultColumnsUtil.scala} | 52 +++--
.../spark/sql/errors/QueryParsingErrors.scala | 5 +
.../org/apache/spark/sql/internal/SQLConf.scala | 14 ++
.../org/apache/spark/sql/SQLInsertTestSuite.scala | 30 ++-
.../org/apache/spark/sql/sources/InsertSuite.scala | 258 ++++++++++++++++++++-
.../org/apache/spark/sql/hive/InsertSuite.scala | 91 ++++----
11 files changed, 638 insertions(+), 70 deletions(-)
diff --git a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4
index 17c3395..872ea53 100644
--- a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4
+++ b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4
@@ -322,6 +322,7 @@ partitionSpec
partitionVal
: identifier (EQ constant)?
+ | identifier EQ DEFAULT
;
namespace
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index f69f17d..bd437c3 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -311,6 +311,7 @@ class Analyzer(override val catalogManager: CatalogManager)
ResolveAggregateFunctions ::
TimeWindowing ::
SessionWindowing ::
+ ResolveDefaultColumns(this, v1SessionCatalog) ::
ResolveInlineTables ::
ResolveLambdaVariables ::
ResolveTimeZone ::
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
new file mode 100644
index 0000000..f4502c9
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala
@@ -0,0 +1,250 @@
+/*
+ * 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.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{SessionCatalog, UnresolvedCatalogRelation}
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+/**
+ * This is a rule to process DEFAULT columns in statements such as CREATE/REPLACE TABLE.
+ *
+ * Background: CREATE TABLE and ALTER TABLE invocations support setting column default values for
+ * later operations. Following INSERT, and INSERT MERGE commands may then reference the value
+ * using the DEFAULT keyword as needed.
+ *
+ * Example:
+ * CREATE TABLE T(a INT DEFAULT 4, b INT NOT NULL DEFAULT 5);
+ * INSERT INTO T VALUES (1, 2);
+ * INSERT INTO T VALUES (1, DEFAULT);
+ * INSERT INTO T VALUES (DEFAULT, 6);
+ * SELECT * FROM T;
+ * (1, 2)
+ * (1, 5)
+ * (4, 6)
+ *
+ * @param analyzer analyzer to use for processing DEFAULT values stored as text.
+ * @param catalog the catalog to use for looking up the schema of INSERT INTO table objects.
+ */
+case class ResolveDefaultColumns(
+ analyzer: Analyzer,
+ catalog: SessionCatalog) extends Rule[LogicalPlan] {
+
+ // This field stores the enclosing INSERT INTO command, once we find one.
+ var enclosingInsert: Option[InsertIntoStatement] = 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
+
+ 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 check if an expression is an explicit DEFAULT column reference.
+ private def isExplicitDefaultColumn(expr: Expression): Boolean = expr match {
+ case u: UnresolvedAttribute if u.name.equalsIgnoreCase(CURRENT_DEFAULT_COLUMN_NAME) => true
+ case _ => false
+ }
+
+ /**
+ * Updates an inline table to generate missing default column values.
+ */
+ private def addMissingDefaultColumnValues(
+ table: UnresolvedInlineTable): Option[UnresolvedInlineTable] = {
+ assert(enclosingInsert.isDefined)
+ val numQueryOutputs: Int = table.rows(0).size
+ val schema = getInsertTableSchemaWithoutPartitionColumns.getOrElse(return None)
+ val newDefaultExpressions: Seq[Expression] = getDefaultExpressions(numQueryOutputs, schema)
+ val newNames: Seq[String] = schema.fields.drop(numQueryOutputs).map { _.name }
+ if (newDefaultExpressions.nonEmpty) {
+ Some(table.copy(
+ names = table.names ++ newNames,
+ rows = table.rows.map { row => row ++ newDefaultExpressions }))
+ } else {
+ None
+ }
+ }
+
+ /**
+ * Adds a new expressions to a projection to generate missing default column values.
+ */
+ private def addMissingDefaultColumnValues(project: Project): Option[Project] = {
+ val numQueryOutputs: Int = project.projectList.size
+ val schema = getInsertTableSchemaWithoutPartitionColumns.getOrElse(return None)
+ val newDefaultExpressions: Seq[Expression] = getDefaultExpressions(numQueryOutputs, schema)
+ val newAliases: Seq[NamedExpression] =
+ newDefaultExpressions.zip(schema.fields).map {
+ case (expr, field) => Alias(expr, field.name)()
+ }
+ if (newDefaultExpressions.nonEmpty) {
+ Some(project.copy(projectList = project.projectList ++ newAliases))
+ } else {
+ None
+ }
+ }
+
+ /**
+ * This is a helper for the addMissingDefaultColumnValues methods above.
+ */
+ 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
+ }
+ }
+ Seq.fill(numDefaultExpressionsToAdd)(UnresolvedAttribute(CURRENT_DEFAULT_COLUMN_NAME))
+ }
+
+ /**
+ * Replaces unresolved DEFAULT column references with corresponding values in a logical plan.
+ */
+ private def replaceExplicitDefaultColumnValues(
+ analyzer: Analyzer,
+ input: LogicalPlan): Option[LogicalPlan] = {
+ assert(enclosingInsert.isDefined)
+ val schema = getInsertTableSchemaWithoutPartitionColumns.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) =>
+ analyze(analyzer, f, "INSERT")
+ case _ => Literal(null)
+ }
+ // Check the type of `input` and replace its expressions accordingly.
+ // If necessary, return a more descriptive error message if the user tries to nest the DEFAULT
+ // column reference inside some other expression, such as DEFAULT + 1 (this is not allowed).
+ //
+ // Note that we don't need to check if "SQLConf.get.useNullsForMissingDefaultColumnValues"
+ // after this point because this method only takes responsibility to replace *existing*
+ // DEFAULT references. In contrast, the "getDefaultExpressions" method will check that config
+ // and add new NULLs if needed.
+ input match {
+ case table: UnresolvedInlineTable =>
+ replaceExplicitDefaultColumnValues(defaultExpressions, table)
+ case project: Project =>
+ replaceExplicitDefaultColumnValues(defaultExpressions, columnNames, project)
+ }
+ }
+
+ /**
+ * Replaces unresolved DEFAULT column references with corresponding values in an inline table.
+ */
+ private def replaceExplicitDefaultColumnValues(
+ defaultExpressions: Seq[Expression],
+ table: UnresolvedInlineTable): Option[LogicalPlan] = {
+ var replaced = false
+ val newRows: Seq[Seq[Expression]] = {
+ table.rows.map { row: Seq[Expression] =>
+ for {
+ i <- 0 until row.size
+ expr = row(i)
+ defaultExpr = if (i < defaultExpressions.size) defaultExpressions(i) else Literal(null)
+ } yield expr match {
+ case u: UnresolvedAttribute if isExplicitDefaultColumn(u) =>
+ replaced = true
+ defaultExpr
+ case expr@_ if expr.find { isExplicitDefaultColumn }.isDefined =>
+ throw new AnalysisException(DEFAULTS_IN_EXPRESSIONS_ERROR)
+ case _ => expr
+ }
+ }
+ }
+ if (replaced) {
+ Some(table.copy(rows = newRows))
+ } else {
+ None
+ }
+ }
+
+ /**
+ * Replaces unresolved DEFAULT column references with corresponding values in an projection.
+ */
+ private def replaceExplicitDefaultColumnValues(
+ defaultExpressions: Seq[Expression],
+ colNames: Seq[String],
+ project: Project): Option[LogicalPlan] = {
+ var replaced = false
+ val updated: Seq[NamedExpression] = {
+ for {
+ i <- 0 until project.projectList.size
+ projectExpr = project.projectList(i)
+ defaultExpr = if (i < defaultExpressions.size) defaultExpressions(i) else Literal(null)
+ colName = if (i < colNames.size) colNames(i) else ""
+ } yield projectExpr match {
+ case Alias(u: UnresolvedAttribute, _) if isExplicitDefaultColumn(u) =>
+ replaced = true
+ Alias(defaultExpr, colName)()
+ case u: UnresolvedAttribute if isExplicitDefaultColumn(u) =>
+ replaced = true
+ Alias(defaultExpr, colName)()
+ case expr@_ if expr.find { isExplicitDefaultColumn }.isDefined =>
+ throw new AnalysisException(DEFAULTS_IN_EXPRESSIONS_ERROR)
+ case _ => projectExpr
+ }
+ }
+ if (replaced) {
+ Some(project.copy(projectList = updated))
+ } else {
+ None
+ }
+ }
+
+ /**
+ * Looks up the schema for the table object of an INSERT INTO statement from the catalog.
+ */
+ 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
+ }
+ } catch {
+ case _: NoSuchTableException => None
+ }
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
index c7925c9..c435e31 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -480,6 +480,11 @@ class AstBuilder extends SqlBaseParserBaseVisitor[AnyRef] with SQLConfHelper wit
val legacyNullAsString =
conf.getConf(SQLConf.LEGACY_PARSE_NULL_PARTITION_SPEC_AS_STRING_LITERAL)
val parts = ctx.partitionVal.asScala.map { pVal =>
+ // Check if the query attempted to refer to a DEFAULT column value within the PARTITION clause
+ // and return a specific error to help guide the user, since this is not allowed.
+ if (pVal.DEFAULT != null) {
+ throw QueryParsingErrors.defaultColumnReferencesNotAllowedInPartitionSpec(ctx)
+ }
val name = pVal.identifier.getText
val value = Option(pVal.constant).map(v => visitStringConstant(v, legacyNullAsString))
name -> value
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
index e36a76b..1204fa8 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
@@ -81,6 +81,7 @@ object RuleIdCollection {
"org.apache.spark.sql.catalyst.analysis.DeduplicateRelations" ::
"org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases" ::
"org.apache.spark.sql.catalyst.analysis.EliminateUnions" ::
+ "org.apache.spark.sql.catalyst.analysis.ResolveDefaultColumns" ::
"org.apache.spark.sql.catalyst.analysis.ResolveExpressionsWithNamePlaceholders" ::
"org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveCoalesceHints" ::
"org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveJoinStrategyHints" ::
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumns.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala
similarity index 78%
rename from sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumns.scala
rename to sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala
index 1e40756..b7db9be 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumns.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala
@@ -38,20 +38,28 @@ object ResolveDefaultColumns {
val CURRENT_DEFAULT_COLUMN_METADATA_KEY = "CURRENT_DEFAULT"
// This column metadata represents the default value for all existing rows in a table after a
// column has been added. This value is determined at time of CREATE TABLE, REPLACE TABLE, or
- // ALTER TABLE ADD COLUMN, and never changes thereafter. The intent is for this "exist default"
- // to be used by any scan when the columns in the source row are missing data. For example,
- // consider the following sequence:
+ // ALTER TABLE ADD COLUMN, and never changes thereafter. The intent is for this "exist default" to
+ // be used by any scan when the columns in the source row are missing data. For example, consider
+ // the following sequence:
// CREATE TABLE t (c1 INT)
// INSERT INTO t VALUES (42)
// ALTER TABLE t ADD COLUMNS (c2 INT DEFAULT 43)
// SELECT c1, c2 FROM t
// In this case, the final query is expected to return 42, 43. The ALTER TABLE ADD COLUMNS command
- // executed after there was already data in the table, so in order to enforce this invariant,
- // we need either (1) an expensive backfill of value 43 at column c2 into all previous rows, or
- // (2) indicate to each data source that selected columns missing data are to generate the
+ // executed after there was already data in the table, so in order to enforce this invariant, we
+ // need either (1) an expensive backfill of value 43 at column c2 into all previous rows, or (2)
+ // indicate to each data source that selected columns missing data are to generate the
// corresponding DEFAULT value instead. We choose option (2) for efficiency, and represent this
// value as the text representation of a folded constant in the "EXISTS_DEFAULT" column metadata.
val EXISTS_DEFAULT_COLUMN_METADATA_KEY = "EXISTS_DEFAULT"
+ // Name of attributes representing explicit references to the value stored in the above
+ // CURRENT_DEFAULT_COLUMN_METADATA.
+ val CURRENT_DEFAULT_COLUMN_NAME = "DEFAULT"
+ // Return a more descriptive error message if the user tries to nest the DEFAULT column reference
+ // inside some other expression, such as DEFAULT + 1 (this is not allowed).
+ val DEFAULTS_IN_EXPRESSIONS_ERROR = "Failed to execute INSERT INTO command because the " +
+ "VALUES list contains a DEFAULT column reference as part of another expression; this is " +
+ "not allowed"
/**
* Finds "current default" expressions in CREATE/REPLACE TABLE columns and constant-folds them.
@@ -75,8 +83,8 @@ object ResolveDefaultColumns {
* data source then takes responsibility to provide the constant-folded value in the
* EXISTS_DEFAULT metadata for such columns where the value is not present in storage.
*
- * @param analyzer used for analyzing the result of parsing the column expression stored as text.
- * @param tableSchema represents the names and types of the columns of the statement to process.
+ * @param analyzer used for analyzing the result of parsing the expression stored as text.
+ * @param tableSchema represents the names and types of the columns of the statement to process.
* @param statementType name of the statement being processed, such as INSERT; useful for errors.
* @return a copy of `tableSchema` with field metadata updated with the constant-folded values.
*/
@@ -84,26 +92,28 @@ object ResolveDefaultColumns {
analyzer: Analyzer,
tableSchema: StructType,
statementType: String): StructType = {
- if (!SQLConf.get.enableDefaultColumns) {
- return tableSchema
- }
- val newFields: Seq[StructField] = tableSchema.fields.map { field =>
- if (field.metadata.contains(CURRENT_DEFAULT_COLUMN_METADATA_KEY)) {
- val analyzed: Expression = analyze(analyzer, field, statementType)
- val newMetadata: Metadata = new MetadataBuilder().withMetadata(field.metadata)
- .putString(EXISTS_DEFAULT_COLUMN_METADATA_KEY, analyzed.sql).build()
- field.copy(metadata = newMetadata)
- } else {
- field
+ if (SQLConf.get.enableDefaultColumns) {
+ val newFields: Seq[StructField] = tableSchema.fields.map { field =>
+ if (field.metadata.contains(CURRENT_DEFAULT_COLUMN_METADATA_KEY)) {
+ val analyzed: Expression = analyze(analyzer, field, statementType)
+ val newMetadata: Metadata = new MetadataBuilder().withMetadata(field.metadata)
+ .putString(EXISTS_DEFAULT_COLUMN_METADATA_KEY, analyzed.sql).build()
+ field.copy(metadata = newMetadata)
+ } else {
+ field
+ }
}
+ StructType(newFields)
+ } else {
+ tableSchema
}
- StructType(newFields)
}
/**
* Parses and analyzes the DEFAULT column text in `field`, returning an error upon failure.
*
- * @param field represents the DEFAULT column value whose "default" metadata to parse and analyze.
+ * @param field represents the DEFAULT column value whose "default" metadata to parse
+ * and analyze.
* @param statementType which type of statement we are running, such as INSERT; useful for errors.
* @return Result of the analysis and constant-folding operation.
*/
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
index b13a530..69e118d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
@@ -452,4 +452,9 @@ object QueryParsingErrors {
def defaultColumnNotEnabledError(ctx: ParserRuleContext): Throwable = {
new ParseException("Support for DEFAULT column values is not allowed", ctx)
}
+
+ def defaultColumnReferencesNotAllowedInPartitionSpec(ctx: ParserRuleContext): Throwable = {
+ new ParseException(
+ "References to DEFAULT column values are not allowed within the PARTITION clause", ctx)
+ }
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 3e1872a..9aad649 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -2816,6 +2816,17 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES =
+ buildConf("spark.sql.defaultColumn.useNullsForMissingDefautValues")
+ .internal()
+ .doc("When true, and DEFAULT columns are enabled, allow column definitions lacking " +
+ "explicit default values to behave as if they had specified DEFAULT NULL instead. " +
+ "For example, this allows most INSERT INTO statements to specify only a prefix of the " +
+ "columns in the target table, and the remaining columns will receive NULL values.")
+ .version("3.4.0")
+ .booleanConf
+ .createWithDefault(false)
+
val ENFORCE_RESERVED_KEYWORDS = buildConf("spark.sql.ansi.enforceReservedKeywords")
.doc(s"When true and '${ANSI_ENABLED.key}' is true, the Spark SQL parser enforces the ANSI " +
"reserved keywords and forbids SQL queries that use reserved keywords as alias names " +
@@ -4318,6 +4329,9 @@ class SQLConf extends Serializable with Logging {
def enableDefaultColumns: Boolean = getConf(SQLConf.ENABLE_DEFAULT_COLUMNS)
+ def useNullsForMissingDefaultColumnValues: Boolean =
+ getConf(SQLConf.USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES)
+
def enforceReservedKeywords: Boolean = ansiEnabled && getConf(ENFORCE_RESERVED_KEYWORDS)
def strictIndexOperator: Boolean = ansiEnabled && getConf(ANSI_STRICT_INDEX_OPERATOR)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLInsertTestSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLInsertTestSuite.scala
index fad01db..274da57 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLInsertTestSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLInsertTestSuite.scala
@@ -171,25 +171,36 @@ trait SQLInsertTestSuite extends QueryTest with SQLTestUtils {
}
test("insert with column list - mismatched column list size") {
- val msg = "Cannot write to table due to mismatched user specified column size"
- withTable("t1") {
- val cols = Seq("c1", "c2", "c3")
- createTable("t1", cols, Seq("int", "long", "string"))
- val e1 = intercept[AnalysisException](sql(s"INSERT INTO t1 (c1, c2) values(1, 2, 3)"))
- assert(e1.getMessage.contains(msg))
- val e2 = intercept[AnalysisException](sql(s"INSERT INTO t1 (c1, c2, c3) values(1, 2)"))
- assert(e2.getMessage.contains(msg))
+ val msgs = Seq("Cannot write to table due to mismatched user specified column size",
+ "expected 3 columns but found")
+ def test: Unit = {
+ withTable("t1") {
+ val cols = Seq("c1", "c2", "c3")
+ createTable("t1", cols, Seq("int", "long", "string"))
+ val e1 = intercept[AnalysisException](sql(s"INSERT INTO t1 (c1, c2) values(1, 2, 3)"))
+ assert(e1.getMessage.contains(msgs(0)) || e1.getMessage.contains(msgs(1)))
+ val e2 = intercept[AnalysisException](sql(s"INSERT INTO t1 (c1, c2, c3) values(1, 2)"))
+ assert(e2.getMessage.contains(msgs(0)) || e2.getMessage.contains(msgs(1)))
+ }
+ }
+ withSQLConf(SQLConf.ENABLE_DEFAULT_COLUMNS.key -> "false") {
+ test
+ }
+ withSQLConf(SQLConf.ENABLE_DEFAULT_COLUMNS.key -> "true") {
+ test
}
}
test("insert with column list - mismatched target table out size after rewritten query") {
- val v2Msg = "Cannot write to 'testcat.t1', not enough data columns:"
+ val v2Msg = "expected 2 columns but found"
val cols = Seq("c1", "c2", "c3", "c4")
withTable("t1") {
createTable("t1", cols, Seq.fill(4)("int"))
val e1 = intercept[AnalysisException](sql(s"INSERT INTO t1 (c1) values(1)"))
assert(e1.getMessage.contains("target table has 4 column(s) but the inserted data has 1") ||
+ e1.getMessage.contains("expected 4 columns but found 1") ||
+ e1.getMessage.contains("not enough data columns") ||
e1.getMessage.contains(v2Msg))
}
@@ -199,6 +210,7 @@ trait SQLInsertTestSuite extends QueryTest with SQLTestUtils {
sql(s"INSERT INTO t1 partition(c3=3, c4=4) (c1) values(1)")
}
assert(e1.getMessage.contains("target table has 4 column(s) but the inserted data has 3") ||
+ e1.getMessage.contains("not enough data columns") ||
e1.getMessage.contains(v2Msg))
}
}
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 1fb4737..2483055 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
@@ -27,7 +27,9 @@ import org.apache.spark.SparkException
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns
import org.apache.spark.sql.execution.datasources.DataSourceUtils
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.PartitionOverwriteMode
@@ -405,7 +407,7 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
}
test("SPARK-15824 - Execute an INSERT wrapped in a WITH statement immediately") {
- withTable("target", "target2") {
+ def test: Unit = withTable("target", "target2") {
sql(s"CREATE TABLE target(a INT, b STRING) USING JSON")
sql("WITH tbl AS (SELECT * FROM jt) INSERT OVERWRITE TABLE target SELECT a, b FROM tbl")
checkAnswer(
@@ -426,6 +428,12 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
sql("SELECT a, b FROM jt")
)
}
+ withSQLConf(SQLConf.ENABLE_DEFAULT_COLUMNS.key -> "true") {
+ test
+ }
+ withSQLConf(SQLConf.ENABLE_DEFAULT_COLUMNS.key -> "false") {
+ test
+ }
}
test("SPARK-21203 wrong results of insertion of Array of Struct") {
@@ -849,6 +857,254 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
}
}
+ test("INSERT INTO statements with tables with default columns: positive tests") {
+ // 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 values(true)")
+ checkAnswer(sql("select s from t where i = true"), Seq(Row(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)))
+ }
+ // 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)))
+ }
+ // 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)))
+ }
+ // There are two trailing default values referenced implicitly by the INSERT INTO statement.
+ withTable("t") {
+ sql("create table t(i int, s bigint default 42, x bigint default 43) using parquet")
+ sql("insert into t values(1)")
+ checkAnswer(sql("select s + x from t where i = 1"), Seq(85L).map(i => Row(i)))
+ }
+ // 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("insert into t partition(i='true') values(5, default)")
+ checkAnswer(sql("select s from t where i = true"), Seq(5).map(i => Row(i)))
+ }
+ // 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)))
+ }
+ // 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)))
+ }
+ // 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)))
+ }
+ // 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)))
+ }
+ // 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)))
+ }
+ // 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)))
+ }
+ // 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)))
+ }
+ // 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)))
+ }
+ // 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)))
+ }
+ // 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)))
+ }
+ // 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
+ // enabling the configuration to use NULLs for missing DEFAULT column values.
+ withSQLConf(SQLConf.USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES.key -> "true") {
+ withTable("t1", "t2") {
+ sql("create table t1(j int, s bigint default 42, x bigint default 43) using parquet")
+ sql("insert into t1 values(1)")
+ sql("insert into t1 values(2, default)")
+ sql("insert into t1 values(3, default, default)")
+ sql("insert into t1 values(4, 44)")
+ sql("insert into t1 values(5, 44, 45)")
+ sql("create table t2(j int, s bigint default 42, x bigint default 43) using parquet")
+ sql("insert into t2 select j from t1 where j = 1")
+ sql("insert into t2 select j, default from t1 where j = 2")
+ sql("insert into t2 select j, default, default from t1 where j = 3")
+ sql("insert into t2 select j, s from t1 where j = 4")
+ sql("insert into t2 select j, s, default from t1 where j = 5")
+ val resultSchema = new StructType()
+ .add("s", LongType, false)
+ .add("x", LongType, false)
+ checkAnswer(
+ sql("select j, s, x from t2 order by j, s, x"),
+ Seq(
+ new GenericRowWithSchema(Array(1, 42L, 43L), resultSchema),
+ new GenericRowWithSchema(Array(2, 42L, 43L), resultSchema),
+ new GenericRowWithSchema(Array(3, 42L, 43L), resultSchema),
+ new GenericRowWithSchema(Array(4, 44L, 43L), resultSchema),
+ new GenericRowWithSchema(Array(5, 44L, 43L), resultSchema)))
+ }
+ }
+ }
+
+ test("INSERT INTO statements with tables with default columns: negative tests") {
+ object Errors {
+ val COMMON_SUBSTRING = " has a DEFAULT value"
+ val COLUMN_DEFAULT_NOT_FOUND = "Column 'default' does not exist"
+ val BAD_SUBQUERY = "cannot evaluate expression scalarsubquery() in inline table definition"
+ }
+ // The default value fails to analyze.
+ withTable("t") {
+ sql("create table t(i boolean, s bigint default badvalue) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t values (default, default)")
+ }.getMessage.contains(Errors.COMMON_SUBSTRING))
+ }
+ // The default value analyzes to a table not in the catalog.
+ withTable("t") {
+ sql("create table t(i boolean, s bigint default (select min(x) from badtable)) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t values (default, default)")
+ }.getMessage.contains(Errors.COMMON_SUBSTRING))
+ }
+ // The default value parses but refers to a table from the catalog.
+ withTable("t", "other") {
+ sql("create table other(x string) using parquet")
+ sql("create table t(i boolean, s bigint default (select min(x) from other)) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t values (default, default)")
+ }.getMessage.contains(Errors.COMMON_SUBSTRING))
+ }
+ // The default value has an explicit alias. It fails to evaluate when inlined into the VALUES
+ // list at the INSERT INTO time.
+ withTable("t") {
+ sql("create table t(i boolean default (select false as alias), s bigint) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t values (default, default)")
+ }.getMessage.contains(Errors.BAD_SUBQUERY))
+ }
+ // Explicit default values may not participate in complex expressions in the VALUES list.
+ withTable("t") {
+ sql("create table t(i boolean, s bigint default 42) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t values(false, default + 1)")
+ }.getMessage.contains(ResolveDefaultColumns.DEFAULTS_IN_EXPRESSIONS_ERROR))
+ }
+ // Explicit default values may not participate in complex expressions in the SELECT query.
+ withTable("t") {
+ sql("create table t(i boolean, s bigint default 42) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t select false, default + 1")
+ }.getMessage.contains(ResolveDefaultColumns.DEFAULTS_IN_EXPRESSIONS_ERROR))
+ }
+ // Explicit default values have a reasonable error path if the table is not found.
+ withTable("t") {
+ assert(intercept[AnalysisException] {
+ sql("insert into t values(false, default)")
+ }.getMessage.contains(Errors.COLUMN_DEFAULT_NOT_FOUND))
+ }
+ // The default value parses but the type is not coercible.
+ withTable("t") {
+ sql("create table t(i boolean, s bigint default false) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t values (default, default)")
+ }.getMessage.contains("provided a value of incompatible type"))
+ }
+ // The number of columns in the INSERT INTO statement is greater than the number of columns in
+ // the table.
+ withTable("t") {
+ sql("create table num_data(id int, val decimal(38,10)) using parquet")
+ sql("create table t(id1 int, int2 int, result decimal(38,10)) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t select t1.id, t2.id, t1.val, t2.val, t1.val * t2.val " +
+ "from num_data t1, num_data t2")
+ }.getMessage.contains(
+ "requires that the data to be inserted have the same number of columns as the target"))
+ }
+ // The default value is disabled per configuration.
+ withTable("t") {
+ withSQLConf(SQLConf.ENABLE_DEFAULT_COLUMNS.key -> "false") {
+ assert(intercept[AnalysisException] {
+ sql("create table t(i boolean, s bigint default 42L) using parquet")
+ }.getMessage.contains("Support for DEFAULT column values is not allowed"))
+ }
+ }
+ // There is one trailing default value referenced implicitly by the INSERT INTO statement.
+ withTable("t") {
+ sql("create table t(i int, s bigint default 42, x bigint) using parquet")
+ assert(intercept[AnalysisException] {
+ sql("insert into t values(1)")
+ }.getMessage.contains("expected 3 columns but found"))
+ }
+ // 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 ) " +
+ "using parquet partitioned by (i)")
+ assert(intercept[ParseException] {
+ sql("insert into t partition(i=default) values(5, default)")
+ }.getMessage.contains(
+ "References to DEFAULT column values are not allowed within the PARTITION clause"))
+ }
+ // The configuration option to append missing NULL values to the end of the INSERT INTO
+ // statement is not enabled.
+ 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 values(true)")
+ }.getMessage.contains("target table has 2 column(s) but the inserted data has 1 column(s)"))
+ }
+ }
+ }
+
test("Stop task set if FileAlreadyExistsException was thrown") {
Seq(true, false).foreach { fastFail =>
withSQLConf("fs.file.impl" -> classOf[FileExistingTestFileSystem].getName,
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala
index 4dc08a0e..a7148e9 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala
@@ -857,52 +857,65 @@ class InsertSuite extends QueryTest with TestHiveSingleton with BeforeAndAfter
}
test("SPARK-35531: Insert data with different cases of bucket column") {
- withTable("test1") {
- Seq(true, false).foreach { isHiveTable =>
- val createSpark = if (isHiveTable) {
- """
- |CREATE TABLE TEST1(
- |v1 BIGINT,
- |s1 INT)
- |PARTITIONED BY (pk BIGINT)
- |CLUSTERED BY (v1)
- |SORTED BY (s1)
- |INTO 200 BUCKETS
- |STORED AS PARQUET
- """.stripMargin
- } else {
- """
- |CREATE TABLE test1(
- |v1 BIGINT,
- |s1 INT)
- |USING PARQUET
- |PARTITIONED BY (pk BIGINT)
- |CLUSTERED BY (v1)
- |SORTED BY (s1)
- |INTO 200 BUCKETS
- """.stripMargin
- }
+ def testDefaultColumn: Unit = {
+ withTable("test1") {
+ Seq(true, false).foreach { isHiveTable =>
+ val createSpark = if (isHiveTable) {
+ """
+ |CREATE TABLE TEST1(
+ |v1 BIGINT,
+ |s1 INT)
+ |PARTITIONED BY (pk BIGINT)
+ |CLUSTERED BY (v1)
+ |SORTED BY (s1)
+ |INTO 200 BUCKETS
+ |STORED AS PARQUET
+ """.stripMargin
+ } else {
+ """
+ |CREATE TABLE test1(
+ |v1 BIGINT,
+ |s1 INT)
+ |USING PARQUET
+ |PARTITIONED BY (pk BIGINT)
+ |CLUSTERED BY (v1)
+ |SORTED BY (s1)
+ |INTO 200 BUCKETS
+ """.stripMargin
+ }
- val insertString =
- """
- |INSERT INTO test1
- |SELECT * FROM VALUES(1,1,1)
- """.stripMargin
+ val insertString =
+ """
+ |INSERT INTO test1
+ |SELECT * FROM VALUES(1,1,1)
+ """.stripMargin
- val dropString = "DROP TABLE IF EXISTS test1"
+ val dropString = "DROP TABLE IF EXISTS test1"
- sql(dropString)
- sql(createSpark.toLowerCase(Locale.ROOT))
+ sql(dropString)
+ sql(createSpark.toLowerCase(Locale.ROOT))
- sql(insertString.toLowerCase(Locale.ROOT))
- sql(insertString.toUpperCase(Locale.ROOT))
+ sql(insertString.toLowerCase(Locale.ROOT))
+ sql(insertString.toUpperCase(Locale.ROOT))
- sql(dropString)
- sql(createSpark.toUpperCase(Locale.ROOT))
+ sql(dropString)
+ sql(createSpark.toUpperCase(Locale.ROOT))
- sql(insertString.toLowerCase(Locale.ROOT))
- sql(insertString.toUpperCase(Locale.ROOT))
+ sql(insertString.toLowerCase(Locale.ROOT))
+ sql(insertString.toUpperCase(Locale.ROOT))
+ }
}
}
+ withSQLConf(SQLConf.ENABLE_DEFAULT_COLUMNS.key -> "false") {
+ testDefaultColumn
+ }
+ withSQLConf(SQLConf.ENABLE_DEFAULT_COLUMNS.key -> "true",
+ SQLConf.USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES.key -> "false") {
+ testDefaultColumn
+ }
+ withSQLConf(SQLConf.ENABLE_DEFAULT_COLUMNS.key -> "true",
+ SQLConf.USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES.key -> "true") {
+ testDefaultColumn
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org