This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 2b7fa93cbedf [SPARK-57644][SQL] Support generated column values on V2
table writes
2b7fa93cbedf is described below
commit 2b7fa93cbedf77dae7afd20653d6b072486cb52e
Author: Szehon Ho <[email protected]>
AuthorDate: Mon Jul 6 12:46:48 2026 -0700
[SPARK-57644][SQL] Support generated column values on V2 table writes
### What changes were proposed in this pull request?
Add support for auto-filling generated column values and enforcing
generated column constraints during V2 table writes (INSERT).
When a catalog declares `SUPPORT_GENERATED_COLUMN_ON_WRITE`:
- Missing generated columns are auto-filled using the generation expression
via a `Project` node added during analysis
- User-provided generated column values are validated against the
generation expression via `CheckInvariant(EqualNullSafe(col, genExpr))`
- MERGE, UPDATE, and streaming writes with generated columns are blocked
until support is implemented
**Key changes:**
- `TableCatalogCapability`: New `SUPPORT_GENERATED_COLUMN_ON_WRITE`
capability
- `CatalogV2Util`: Encode generation expression in V2-to-StructField
metadata roundtrip
- `TableOutputResolver`: Fill missing generated columns in by-name and
by-position write paths (checking generation expression before null defaults)
- `ResolveTableConstraints`: Add generated column constraints using V2
expression with SQL parser fallback, scoped to user-provided columns only
- `Analyzer` (`ResolveOutputRelation`): Gate on capability; strip
generation expression metadata from auto-filled columns so constraints are
scoped correctly
- `RewriteRowLevelCommand`: Block MERGE/UPDATE with generated columns
- `ResolveWriteToStream`: Block streaming writes with generated columns
- `SQLConf`: New `spark.sql.generatedColumn.allowNullableIngest.enabled`
config
### Why are the changes needed?
Generated columns (defined with `GENERATED ALWAYS AS (expr)`) currently
only support DDL — the generation expression is stored but never evaluated
during writes. This means connectors must handle generated column values
themselves (e.g., Delta Lake does this in its own execution layer).
This PR adds Spark-native support so that any V2 catalog can opt in to
having Spark auto-compute generated column values during INSERT. This brings
generated columns closer to parity with default column values, which Spark
already handles during writes.
### Does this PR introduce _any_ user-facing change?
Yes. For catalogs that declare `SUPPORT_GENERATED_COLUMN_ON_WRITE`:
- Users can omit generated columns from INSERT statements and the values
are auto-computed
- Users who explicitly provide generated column values will get a
`CHECK_CONSTRAINT_VIOLATION` error if the value doesn't match the generation
expression
- MERGE, UPDATE, and streaming writes to tables with generated columns will
fail with `UNSUPPORTED_FEATURE.TABLE_OPERATION`
- New config `spark.sql.generatedColumn.allowNullableIngest.enabled`
(default `true`) controls whether nullable non-generated columns can be omitted
when writing to tables with generated columns
### How was this patch tested?
New `GeneratedColumnWriteSuite` with 30 tests covering:
- Auto-fill by name and by position
- Constraint validation (matching/non-matching/NULL values)
- Multiple generated columns, multi-column expressions, complex expressions
(CAST)
- Type coercion, column reordering, case-insensitive matching
- INSERT OVERWRITE, INSERT SELECT, CTAS, DataFrame API
- Generated column in middle of schema, non-trailing by position (error)
- Generated partition columns
- Config for nullable ingest
- Combined with table CHECK constraints
- Plan inspection (CheckInvariant present for user-provided, absent for
auto-filled)
- Streaming write blocking
- Mix of auto-filled and user-provided generated columns
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-opus-4-6)
Closes #56712 from szehon-ho/generated-column-write.
Authored-by: Szehon Ho <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit 4553429c621044bb30d20b901f61bed8b2089817)
Signed-off-by: Szehon Ho <[email protected]>
---
.../connector/catalog/TableCatalogCapability.java | 20 +-
.../spark/sql/catalyst/analysis/Analyzer.scala | 32 +-
.../analysis/ResolveTableConstraints.scala | 84 +-
.../catalyst/analysis/RewriteMergeIntoTable.scala | 3 +
.../catalyst/analysis/RewriteRowLevelCommand.scala | 24 +-
.../sql/catalyst/analysis/RewriteUpdateTable.scala | 1 +
.../catalyst/analysis/TableOutputResolver.scala | 181 ++-
.../plans/logical/GeneratedColumnExpression.scala | 7 +-
.../spark/sql/catalyst/util/GeneratedColumn.scala | 68 +-
.../sql/connector/catalog/CatalogV2Util.scala | 6 +
.../org/apache/spark/sql/internal/SQLConf.scala | 13 +
.../connector/catalog/InMemoryTableCatalog.scala | 3 +-
.../streaming/runtime/ResolveWriteToStream.scala | 9 +
.../sql/connector/GeneratedColumnWriteSuite.scala | 1177 ++++++++++++++++++++
14 files changed, 1569 insertions(+), 59 deletions(-)
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalogCapability.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalogCapability.java
index a60c827d5ace..ac829395b9bc 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalogCapability.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalogCapability.java
@@ -92,5 +92,23 @@ public enum TableCatalogCapability {
* {@link TableCatalog#createTable}.
* See {@link Column#identityColumnSpec()}.
*/
- SUPPORTS_CREATE_TABLE_WITH_IDENTITY_COLUMNS
+ SUPPORTS_CREATE_TABLE_WITH_IDENTITY_COLUMNS,
+
+ /**
+ * Signals that the TableCatalog supports Spark auto-filling generated
column values and
+ * enforcing generated column constraints during writes.
+ * <p>
+ * When this capability is present, Spark will:
+ * <ul>
+ * <li>Auto-compute missing generated column values using the generation
expression.</li>
+ * <li>Validate explicitly-provided generated column values against the
generation
+ * expression.</li>
+ * </ul>
+ * <p>
+ * Without this capability, the connector is responsible for handling
generated column values
+ * during writes.
+ *
+ * @since 4.3.0
+ */
+ SUPPORT_GENERATED_COLUMN_ON_WRITE
}
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 7fb7086449f6..870a99b26ad6 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
@@ -49,7 +49,7 @@ import org.apache.spark.sql.catalyst.trees.AlwaysProcess
import org.apache.spark.sql.catalyst.trees.CurrentOrigin.withOrigin
import org.apache.spark.sql.catalyst.trees.TreePattern._
import org.apache.spark.sql.catalyst.types.DataTypeUtils
-import org.apache.spark.sql.catalyst.util.{toPrettySQL,
trimTempResolvedColumn, CharVarcharUtils}
+import org.apache.spark.sql.catalyst.util.{toPrettySQL,
trimTempResolvedColumn, CharVarcharUtils, GeneratedColumn}
import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
// `View` is aliased to `V2View` to avoid clashing with the logical-plan
`View` imported via
// `org.apache.spark.sql.catalyst.plans.logical._`.
@@ -3916,13 +3916,35 @@ class Analyzer(
val defaultValueFillMode =
if (conf.coerceInsertNestedTypes && v2Write.schemaEvolutionEnabled)
RECURSE
else FILL
- val projection = TableOutputResolver.resolveOutputColumns(
- v2Write.table.name, v2Write.table.output, v2Write.query,
v2Write.isByName, conf,
- defaultValueFillMode)
+ // Only let TableOutputResolver see generation expression metadata if
the catalog
+ // supports auto-filling generated columns on write.
+ val expected = v2Write.table match {
+ case r: DataSourceV2Relation
+ if !GeneratedColumn.supportsGeneratedColumnsOnWrite(r.catalog) =>
+ r.output.map(GeneratedColumn.removeGenerationExpressionMetadata)
+ case _ => v2Write.table.output
+ }
+ val (projection, autoFilledGenCols) =
+ TableOutputResolver.resolveOutputColumnsWithGeneratedInfo(
+ v2Write.table.name, expected, v2Write.query, v2Write.isByName,
conf,
+ defaultValueFillMode)
if (projection != v2Write.query) {
val cleanedTable = v2Write.table match {
case r: DataSourceV2Relation =>
- r.copy(output = r.output.map(CharVarcharUtils.cleanAttrMetadata))
+ r.copy(output = r.output.map { attr =>
+ val cleaned = CharVarcharUtils.cleanAttrMetadata(attr)
+ // Strip the generation expression metadata from columns Spark
auto-filled, so
+ // ResolveTableConstraints does not add a (redundant)
CheckInvariant for them:
+ // their values were computed from the generation expression
and are correct by
+ // construction. User-provided generated columns keep the
metadata so their
+ // values are still validated.
+ if (autoFilledGenCols.contains(attr.name)) {
+ GeneratedColumn.removeGenerationExpressionMetadata(cleaned)
+ .asInstanceOf[AttributeReference]
+ } else {
+ cleaned
+ }
+ })
case other => other
}
v2Write.withNewQuery(projection).withNewTable(cleanedTable)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala
index 41631b24a83e..de4490143a83 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala
@@ -18,11 +18,12 @@ package org.apache.spark.sql.catalyst.analysis
import scala.collection.mutable
-import org.apache.spark.sql.catalyst.expressions.{And, CheckInvariant,
Expression, V2ExpressionUtils}
+import org.apache.spark.sql.catalyst.expressions.{And, CheckInvariant,
EqualNullSafe, Expression, V2ExpressionUtils}
import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan,
V2WriteCommand, WriteDelta}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.COMMAND
-import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.catalyst.util.GeneratedColumn
+import org.apache.spark.sql.connector.catalog.{CatalogManager,
GenerationExpression}
import org.apache.spark.sql.connector.catalog.constraints.Check
import org.apache.spark.sql.connector.write.RowLevelOperation
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
@@ -39,18 +40,24 @@ class ResolveTableConstraints(val catalogManager:
CatalogManager) extends Rule[L
if v2Write.table.resolved && v2Write.query.resolved &&
!containsCheckInvariant(v2Write.query) && v2Write.outputResolved =>
v2Write.table match {
- case r: DataSourceV2Relation
- if r.table.constraints != null && r.table.constraints.nonEmpty =>
- // Check constraint is the only enforced constraint for DSV2 tables.
- val checkInvariants = r.table.constraints.collect {
- case c: Check =>
- val unresolvedExpr = buildCatalystExpression(c)
- val columnExtractors = mutable.Map[String, Expression]()
- buildColumnExtractors(unresolvedExpr, columnExtractors)
- CheckInvariant(unresolvedExpr, columnExtractors.toSeq, c.name,
c.predicateSql)
- }
+ case r: DataSourceV2Relation =>
+ val tableCheckInvariants =
+ if (r.table.constraints != null && r.table.constraints.nonEmpty) {
+ // Check constraint is the only enforced constraint for DSV2
tables.
+ r.table.constraints.collect {
+ case c: Check =>
+ val unresolvedExpr = buildCatalystExpression(c)
+ val columnExtractors = mutable.Map[String, Expression]()
+ buildColumnExtractors(unresolvedExpr, columnExtractors)
+ CheckInvariant(unresolvedExpr, columnExtractors.toSeq,
c.name, c.predicateSql)
+ }.toSeq
+ } else {
+ Seq.empty
+ }
+ val genColInvariants = buildGeneratedColumnConstraints(r, v2Write)
+ val allInvariants = tableCheckInvariants ++ genColInvariants
// Combine the check invariants into a single expression using
conjunctive AND.
- checkInvariants.reduceOption(And).fold(v2Write)(
+ allInvariants.reduceOption(And).fold(v2Write)(
condition => v2Write.withNewQuery(Filter(condition,
v2Write.query)))
case _ =>
v2Write
@@ -72,6 +79,57 @@ class ResolveTableConstraints(val catalogManager:
CatalogManager) extends Rule[L
.getOrElse(catalogManager.v1SessionCatalog.parser.parseExpression(c.predicateSql()))
}
+ /**
+ * For each user-provided generated column, add a CheckInvariant that
validates the column
+ * value matches the generation expression. Auto-filled generated columns
are excluded:
+ * ResolveOutputRelation strips their GENERATION_EXPRESSION metadata from
the table output,
+ * so they won't appear here.
+ */
+ private def buildGeneratedColumnConstraints(
+ r: DataSourceV2Relation,
+ v2Write: V2WriteCommand): Seq[Expression] = {
+ if (!GeneratedColumn.supportsGeneratedColumnsOnWrite(r.catalog,
r.table.columns())) {
+ return Seq.empty
+ }
+
+ // Use V2 columns from the table to access both V2 expressions and SQL
strings.
+ // Only add constraints for generated columns whose GENERATION_EXPRESSION
metadata
+ // is still present in the table output -- ResolveOutputRelation strips
the metadata
+ // from auto-filled columns so they are excluded here.
+ val v2Columns = r.table.columns()
+ val resolver = catalogManager.v1SessionCatalog.conf.resolver
+ val userProvidedGenCols = r.output
+ .filter(attr => GeneratedColumn.isGeneratedColumn(attr.metadata))
+ .map(_.name)
+ .toSet
+
+ v2Columns.flatMap { col =>
+ Option(col.columnGenerationExpression())
+ .filter(_ => userProvidedGenCols.exists(n => resolver(n, col.name)))
+ .map { genExpr =>
+ val catalystExpr = buildGenerationCatalystExpression(genExpr)
+ val colRef = UnresolvedAttribute.quoted(col.name)
+ val columnExtractors = Seq(col.name -> colRef)
+ val genExprSql = Option(genExpr.getSql()).getOrElse(catalystExpr.sql)
+ CheckInvariant(
+ EqualNullSafe(colRef, catalystExpr),
+ columnExtractors,
+ "Generated Column",
+ s"${col.name} <=> $genExprSql")
+ }
+ }.toSeq
+ }
+
+ /**
+ * Convert a V2 GenerationExpression to a Catalyst expression.
+ * Try the V2 expression first, fall back to parsing the SQL string.
+ */
+ private def buildGenerationCatalystExpression(genExpr:
GenerationExpression): Expression = {
+ Option(genExpr.getExpression)
+ .flatMap(V2ExpressionUtils.toCatalyst)
+
.getOrElse(catalogManager.v1SessionCatalog.parser.parseExpression(genExpr.getSql))
+ }
+
private def buildColumnExtractors(
expr: Expression,
columnExtractors: mutable.Map[String, Expression]): Unit = {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
index 8281f89bd2e8..e11a985d85dd 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
@@ -52,6 +52,7 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand
with PredicateHelper
EliminateSubqueryAliases(aliasedTable) match {
case r: DataSourceV2Relation =>
+ checkNoGeneratedColumns(r, MERGE)
validateMergeIntoConditions(m)
// NOT MATCHED conditions may only refer to columns in source so
they can be pushed down
@@ -85,6 +86,7 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand
with PredicateHelper
EliminateSubqueryAliases(aliasedTable) match {
case r: DataSourceV2Relation =>
+ checkNoGeneratedColumns(r, MERGE)
validateMergeIntoConditions(m)
// there are only NOT MATCHED actions, use a left anti join to
remove any matching rows
@@ -124,6 +126,7 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand
with PredicateHelper
notMatchedBySourceActions, _) if m.resolved && m.rewritable &&
m.aligned =>
EliminateSubqueryAliases(aliasedTable) match {
case r @ ExtractV2Table(tbl: SupportsRowLevelOperations) =>
+ checkNoGeneratedColumns(r, MERGE)
validateMergeIntoConditions(m)
val table = buildOperationTable(tbl, MERGE,
CaseInsensitiveStringMap.empty())
table.operation match {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala
index 48c48eb323bd..60143ae46c3c 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala
@@ -24,7 +24,8 @@ import org.apache.spark.sql.catalyst.ProjectingInternalRow
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
AttributeReference, AttributeSet, Expression, ExprId, If, Literal,
MetadataAttribute, NamedExpression, V2ExpressionUtils}
import org.apache.spark.sql.catalyst.plans.logical.{Assignment, Expand,
LogicalPlan, MergeRows, Project, Union}
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.catalyst.util.{ReplaceDataProjections,
WriteDeltaProjections}
+import org.apache.spark.sql.catalyst.util.{GeneratedColumn,
ReplaceDataProjections,
+ WriteDeltaProjections}
import org.apache.spark.sql.catalyst.util.RowDeltaUtils._
import org.apache.spark.sql.connector.catalog.SupportsRowLevelOperations
import org.apache.spark.sql.connector.expressions.FieldReference
@@ -47,6 +48,27 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
protected def groupFilterEnabled: Boolean =
conf.runtimeRowLevelOperationGroupFilterEnabled
+ /**
+ * Throws if the catalog supports auto-filling generated columns on write
and the table
+ * has generated columns. MERGE and UPDATE with generated columns are not
yet supported.
+ *
+ * This is intentionally coarse-grained: the whole statement is rejected
whenever the target
+ * has any generated column, even if the operation does not assign to a
generated column. This
+ * is because Spark cannot yet recompute generated column values for the
rewritten rows, so it
+ * fails fast rather than risk writing stale values. It can be relaxed once
recomputation is
+ * implemented for these operations.
+ */
+ protected def checkNoGeneratedColumns(
+ relation: DataSourceV2Relation,
+ command: Command): Unit = {
+ if (GeneratedColumn.supportsGeneratedColumnsOnWrite(
+ relation.catalog, relation.table.columns())) {
+ throw QueryCompilationErrors.unsupportedTableOperationError(
+ relation.catalog.get, relation.identifier.get,
+ s"${command.toString} with generated columns")
+ }
+ }
+
protected def buildOperationTable(
table: SupportsRowLevelOperations,
command: Command,
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
index f235374bd5d6..5664e4b3a24c 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
@@ -41,6 +41,7 @@ object RewriteUpdateTable extends RewriteRowLevelCommand {
EliminateSubqueryAliases(aliasedTable) match {
case r @ ExtractV2Table(tbl: SupportsRowLevelOperations) =>
+ checkNoGeneratedColumns(r, UPDATE)
val table = buildOperationTable(tbl, UPDATE,
CaseInsensitiveStringMap.empty())
val updateCond = cond.getOrElse(TrueLiteral)
table.operation match {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
index aa379a63a2af..5526d2667c86 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
@@ -25,10 +25,11 @@ import org.apache.spark.sql.catalyst.SQLConfHelper
import
org.apache.spark.sql.catalyst.analysis.TableOutputResolver.DefaultValueFillMode.{FILL,
NONE, RECURSE}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
-import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, GeneratedColumn}
import
org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getDefaultValueExprOrNullLit
import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLId
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
@@ -94,6 +95,36 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
byName: Boolean,
conf: SQLConf,
defaultValueFillMode: DefaultValueFillMode.Value = NONE): LogicalPlan = {
+ resolveOutputColumnsInternal(
+ tableName, expected, query, byName, conf, defaultValueFillMode)._1
+ }
+
+ /**
+ * @return a tuple of:
+ * - the resolved write plan (same as [[resolveOutputColumns]]); and
+ * - the names of the generated columns that were auto-filled, i.e.
computed from their
+ * generation expression because the user did not provide a value for
them.
+ */
+ def resolveOutputColumnsWithGeneratedInfo(
+ tableName: String,
+ expected: Seq[Attribute],
+ query: LogicalPlan,
+ byName: Boolean,
+ conf: SQLConf,
+ defaultValueFillMode: DefaultValueFillMode.Value = NONE
+ ): (LogicalPlan, Set[String]) = {
+ resolveOutputColumnsInternal(
+ tableName, expected, query, byName, conf, defaultValueFillMode)
+ }
+
+ private def resolveOutputColumnsInternal(
+ tableName: String,
+ expected: Seq[Attribute],
+ query: LogicalPlan,
+ byName: Boolean,
+ conf: SQLConf,
+ defaultValueFillMode: DefaultValueFillMode.Value
+ ): (LogicalPlan, Set[String]) = {
if (expected.size < query.output.size) {
throw QueryCompilationErrors.cannotWriteTooManyColumnsToTableError(
@@ -105,7 +136,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
// an error.
val fillDefaultValue = defaultValueFillMode == RECURSE
val errors = new mutable.ArrayBuffer[String]()
- val resolved: Seq[NamedExpression] = if (byName) {
+ // The resolver also reports which generated columns it auto-filled (not
provided by the user).
+ val (resolved, autoFilledGenCols) = if (byName) {
// By-name resolution: the defaultValueFillMode is passed through to
control whether
// missing top-level columns are filled (FILL/RECURSE) and whether
missing nested
// struct fields are also filled (RECURSE only).
@@ -124,7 +156,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
tableName, expected.map(_.name), query.output)
}
resolveColumnsByPosition(
- tableName, query.output, expected, conf, errors += _, fillDefaultValue
= fillDefaultValue)
+ tableName, query.output, expected, conf, errors += _,
+ fillDefaultValue = fillDefaultValue)
}
if (errors.nonEmpty) {
@@ -132,11 +165,12 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
tableName, expected.map(_.name).map(toSQLId).mkString(", "))
}
- if (resolved == query.output) {
+ val plan = if (resolved == query.output) {
query
} else {
Project(resolved, query)
}
+ (plan, autoFilledGenCols)
}
def resolveUpdate(
@@ -337,23 +371,41 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
addError: String => Unit,
colPath: Seq[String] = Nil,
defaultValueFillMode: DefaultValueFillMode.Value,
- enforceFullOutput: Boolean = false): Seq[NamedExpression] = {
+ enforceFullOutput: Boolean = false): (Seq[NamedExpression], Set[String])
= {
+ // Names of generated columns that were auto-filled (not provided by the
user). Only populated
+ // for top-level columns, since generated columns cannot be nested.
+ val autoFilledGenCols = mutable.Set.empty[String]
val matchedCols = mutable.HashSet.empty[String]
val reordered = expectedCols.flatMap { expectedCol =>
val matched = inputCols.filter(col => conf.resolver(col.name,
expectedCol.name))
val newColPath = colPath :+ expectedCol.name
if (matched.isEmpty) {
- val defaultExpr = if (Set(FILL,
RECURSE).contains(defaultValueFillMode)) {
- getDefaultValueExprOrNullLit(expectedCol,
conf.useNullsForMissingDefaultColumnValues)
- } else {
- None
+ // Check for generated column expression first, before falling back to
defaults,
+ // since getDefaultValueExprOrNullLit may return null for nullable
columns.
+ GeneratedColumn.getGenerationExpression(expectedCol.metadata) match {
+ case Some(genExprSql) =>
+ autoFilledGenCols += expectedCol.name
+ // The parsed references here are placeholders: they are resolved
against the post-cast
+ // stored columns after this loop (see below), so the generated
value is computed from
+ // the values as they will be written. References to other
generated columns are not
+ // allowed.
+ val genExpr = CatalystSqlParser.parseExpression(genExprSql)
+ Some(applyColumnMetadata(genExpr, expectedCol))
+ case None =>
+ val useNullAsDefault =
useNullAsDefaultForMissingColumn(expectedCols, conf)
+ val defaultExpr = if (Set(FILL,
RECURSE).contains(defaultValueFillMode)) {
+ getDefaultValueExprOrNullLit(expectedCol, useNullAsDefault)
+ } else {
+ None
+ }
+ if (defaultExpr.isDefined) {
+ Some(applyDefaultWithLengthCheck(defaultExpr.get, expectedCol,
conf))
+ } else {
+ throw
QueryCompilationErrors.incompatibleDataToTableCannotFindDataError(
+ tableName, newColPath.quoted
+ )
+ }
}
- if (defaultExpr.isEmpty) {
- throw
QueryCompilationErrors.incompatibleDataToTableCannotFindDataError(
- tableName, newColPath.quoted
- )
- }
- Some(applyDefaultWithLengthCheck(defaultExpr.get, expectedCol, conf))
} else if (matched.length > 1) {
throw
QueryCompilationErrors.incompatibleDataToTableAmbiguousColumnNameError(
tableName, newColPath.quoted
@@ -385,7 +437,24 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
}
}
- if (reordered.length == expectedCols.length) {
+ // Resolve auto-filled generated column references against the post-cast
stored columns, so the
+ // generated value is computed from the values as they will be written
(after any
+ // store-assignment cast). Generated columns only exist at the top level,
+ // so this only applies when colPath is empty.
+ val finalCols = if (colPath.isEmpty && autoFilledGenCols.nonEmpty) {
+ val storedCols = reordered.filterNot(col =>
autoFilledGenCols.contains(col.name))
+ reordered.map { col =>
+ if (autoFilledGenCols.contains(col.name)) {
+ resolveGenerationExprReferences(col, storedCols,
conf).asInstanceOf[NamedExpression]
+ } else {
+ col
+ }
+ }
+ } else {
+ reordered
+ }
+
+ val output = if (reordered.length == expectedCols.length) {
if (matchedCols.size < inputCols.length) {
val extraCols = inputCols.filterNot(col =>
matchedCols.contains(col.name))
.map(col => s"${toSQLId(col.name)}").mkString(", ")
@@ -397,7 +466,7 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
tableName, colPath.quoted, extraCols)
}
} else {
- reordered
+ finalCols
}
} else if (enforceFullOutput) {
val colName =
@@ -407,6 +476,7 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
} else {
Nil
}
+ (output, autoFilledGenCols.toSet)
}
private def resolveColumnsByPosition(
@@ -416,7 +486,10 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
conf: SQLConf,
addError: String => Unit,
colPath: Seq[String] = Nil,
- fillDefaultValue: Boolean = false): Seq[NamedExpression] = {
+ fillDefaultValue: Boolean = false): (Seq[NamedExpression], Set[String])
= {
+ // Names of generated columns that were auto-filled (not provided by the
user). Only populated
+ // for top-level columns, since generated columns cannot be nested.
+ val autoFilledGenCols = mutable.Set.empty[String]
val actualExpectedCols = expectedCols.map { attr =>
attr.withDataType {
CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType) }
}
@@ -433,7 +506,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
)
}
} else if (inputCols.size < actualExpectedCols.size && !fillDefaultValue) {
- val missingColsStr =
actualExpectedCols.takeRight(actualExpectedCols.size - inputCols.size)
+ val missingCols = actualExpectedCols.drop(inputCols.size)
+ val missingColsStr = missingCols
.map(col => toSQLId(col.name))
.mkString(", ")
if (colPath.isEmpty) {
@@ -466,15 +540,34 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
}
}
- val defaults = if (fillDefaultValue) {
- actualExpectedCols.drop(inputCols.size).map { expectedCol =>
- val defaultExpr = getDefaultValueExprOrNullLit(
- expectedCol, conf.useNullsForMissingDefaultColumnValues)
- if (defaultExpr.isEmpty) {
- throw
QueryCompilationErrors.incompatibleDataToTableCannotFindDataError(
- tableName, (colPath :+ expectedCol.name).quoted)
+ val trailingCols = actualExpectedCols.drop(inputCols.size)
+ val defaults = if (fillDefaultValue || trailingCols.nonEmpty) {
+ trailingCols.map { expectedCol =>
+ // Check for generated column expression first, before falling back to
defaults.
+ GeneratedColumn.getGenerationExpression(expectedCol.metadata) match {
+ case Some(genExprSql) =>
+ autoFilledGenCols += expectedCol.name
+ val genExpr = CatalystSqlParser.parseExpression(genExprSql)
+ // For by-position, manually resolve references against matched
columns
+ // since the query may use different column names than the table.
References to
+ // other generated columns are not allowed.
+ val resolvedGenExpr = resolveGenerationExprReferences(genExpr,
matched, conf)
+ applyColumnMetadata(resolvedGenExpr, expectedCol)
+ case None =>
+ val useNullAsDefault =
useNullAsDefaultForMissingColumn(actualExpectedCols, conf)
+ val defaultExpr = if (fillDefaultValue) {
+ getDefaultValueExprOrNullLit(
+ expectedCol, useNullAsDefault)
+ } else {
+ None
+ }
+ if (defaultExpr.isDefined) {
+ applyDefaultWithLengthCheck(defaultExpr.get, expectedCol, conf)
+ } else {
+ throw
QueryCompilationErrors.incompatibleDataToTableCannotFindDataError(
+ tableName, (colPath :+ expectedCol.name).quoted)
+ }
}
- applyDefaultWithLengthCheck(defaultExpr.get, expectedCol, conf)
}
} else {
Nil
@@ -487,7 +580,7 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
else actualExpectedCols.map(_.name).map(toSQLId).mkString(", ")
throw
QueryCompilationErrors.incompatibleDataToTableCannotFindDataError(tableName,
colName)
}
- result
+ (result, autoFilledGenCols.toSet)
}
private[sql] def checkNullability(
@@ -527,7 +620,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
Alias(GetStructField(nullCheckedInput, i, Some(f.name)), f.name)()
}
val defaultValueMode = if (fillDefaultValue) RECURSE else NONE
- val resolved = if (byName) {
+ // Generated columns cannot be nested, so the auto-filled set is always
empty here.
+ val (resolved, _) = if (byName) {
reorderColumnsByName(tableName, fields, toAttributes(expectedType),
conf, addError, colPath,
defaultValueMode, enforceFullOutput)
} else {
@@ -568,7 +662,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
val param = NamedLambdaVariable("element", inputType.elementType,
inputType.containsNull)
val fakeAttr =
AttributeReference("element", expectedType.elementType,
expectedType.containsNull)()
- val res = if (byName) {
+ // Generated columns cannot be nested, so the auto-filled set is always
empty here.
+ val (res, _) = if (byName) {
val defaultValueMode = if (fillDefaultValue) RECURSE else NONE
reorderColumnsByName(tableName, Seq(param), Seq(fakeAttr), conf,
addError, colPath,
defaultValueMode, enforceFullOutput)
@@ -611,7 +706,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
val keyParam = NamedLambdaVariable("key", inputType.keyType, nullable =
false)
val fakeKeyAttr = AttributeReference("key", expectedType.keyType, nullable
= false)()
val defaultValueFillMode = if (fillDefaultValue) RECURSE else NONE
- val resKey = if (byName) {
+ // Generated columns cannot be nested, so the auto-filled set is always
empty here.
+ val (resKey, _) = if (byName) {
reorderColumnsByName(tableName, Seq(keyParam), Seq(fakeKeyAttr), conf,
addError, colPath,
defaultValueFillMode, enforceFullOutput)
} else {
@@ -623,7 +719,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
NamedLambdaVariable("value", inputType.valueType,
inputType.valueContainsNull)
val fakeValueAttr =
AttributeReference("value", expectedType.valueType,
expectedType.valueContainsNull)()
- val resValue = if (byName) {
+ // Generated columns cannot be nested, so the auto-filled set is always
empty here.
+ val (resValue, _) = if (byName) {
reorderColumnsByName(tableName, Seq(valueParam), Seq(fakeValueAttr),
conf, addError, colPath,
defaultValueFillMode, enforceFullOutput)
} else {
@@ -662,6 +759,26 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
}
// scalastyle:on argcount
+ private def useNullAsDefaultForMissingColumn(
+ expectedCols: Seq[Attribute],
+ conf: SQLConf): Boolean = {
+ conf.useNullsForMissingDefaultColumnValues &&
+ (conf.generatedColumnAllowNullableIngest ||
+ !expectedCols.exists(c =>
GeneratedColumn.isGeneratedColumn(c.metadata)))
+ }
+
+ private def resolveGenerationExprReferences(
+ genExpr: Expression,
+ matched: Seq[NamedExpression],
+ conf: SQLConf): Expression = {
+ genExpr.transform {
+ case u: UnresolvedAttribute if u.nameParts.size == 1 =>
+ matched.collectFirst {
+ case alias: Alias if conf.resolver(alias.name, u.nameParts.head) =>
alias.child
+ }.getOrElse(u)
+ }
+ }
+
// For table insertions, capture the overflow errors and show proper message.
// Without this method, the overflow errors of castings will show hints for
turning off ANSI SQL
// mode, which are not helpful since the behavior is controlled by the store
assignment policy.
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/GeneratedColumnExpression.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/GeneratedColumnExpression.scala
index 6253ccccafc3..19f1336559b2 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/GeneratedColumnExpression.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/GeneratedColumnExpression.scala
@@ -23,7 +23,7 @@ import
org.apache.spark.sql.catalyst.trees.TreePattern.{ANALYSIS_AWARE_EXPRESSIO
import org.apache.spark.sql.catalyst.util.V2ExpressionBuilder
import org.apache.spark.sql.connector.catalog.GenerationExpression
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types.{DataType, TimeType}
import org.apache.spark.sql.util.SchemaUtils
/**
@@ -109,6 +109,11 @@ case class GeneratedColumnExpression(
throw unsupportedExpressionError("generation expression is not
deterministic")
}
+ if (targetDataType.existsRecursively(_.isInstanceOf[TimeType]) ||
+ child.exists(_.dataType.existsRecursively(_.isInstanceOf[TimeType]))) {
+ throw unsupportedExpressionError("TIME type is not supported in
generated columns")
+ }
+
if (!Cast.canUpCast(child.dataType, targetDataType)) {
throw unsupportedExpressionError(
s"generation expression data type ${child.dataType.simpleString} " +
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/GeneratedColumn.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/GeneratedColumn.scala
index 2f92428cc53e..0469cfc7678e 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/GeneratedColumn.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/GeneratedColumn.scala
@@ -17,10 +17,12 @@
package org.apache.spark.sql.catalyst.util
+import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.plans.logical.ColumnDefinition
-import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog,
TableCatalogCapability}
+import org.apache.spark.sql.connector.catalog.{CatalogPlugin, Column,
Identifier, TableCatalog,
+ TableCatalogCapability}
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.types.{StructField, StructType}
+import org.apache.spark.sql.types.{Metadata, MetadataBuilder, StructField,
StructType}
/**
* This object contains utility methods and values for Generated Columns
@@ -37,15 +39,29 @@ object GeneratedColumn {
* Whether the given `field` is a generated column
*/
def isGeneratedColumn(field: StructField): Boolean = {
- field.metadata.contains(GENERATION_EXPRESSION_METADATA_KEY)
+ isGeneratedColumn(field.metadata)
+ }
+
+ /**
+ * Whether the given metadata indicates a generated column
+ */
+ def isGeneratedColumn(metadata: Metadata): Boolean = {
+ metadata.contains(GENERATION_EXPRESSION_METADATA_KEY)
}
/**
* Returns the generation expression stored in the column metadata if it
exists
*/
def getGenerationExpression(field: StructField): Option[String] = {
- if (isGeneratedColumn(field)) {
- Some(field.metadata.getString(GENERATION_EXPRESSION_METADATA_KEY))
+ getGenerationExpression(field.metadata)
+ }
+
+ /**
+ * Returns the generation expression stored in the metadata if it exists
+ */
+ def getGenerationExpression(metadata: Metadata): Option[String] = {
+ if (isGeneratedColumn(metadata)) {
+ Some(metadata.getString(GENERATION_EXPRESSION_METADATA_KEY))
} else {
None
}
@@ -74,4 +90,46 @@ object GeneratedColumn {
}
}
}
+
+ /**
+ * Whether the table catalog supports Spark auto-filling generated column
values and enforcing
+ * generated column constraints during writes (the
+ * [[TableCatalogCapability.SUPPORT_GENERATED_COLUMN_ON_WRITE]] capability).
Without it, the
+ * connector is responsible for handling generated column values.
+ */
+ def supportsGeneratedColumnsOnWrite(catalog: Option[CatalogPlugin]): Boolean
= {
+ catalog.exists {
+ case tc: TableCatalog =>
+
tc.capabilities().contains(TableCatalogCapability.SUPPORT_GENERATED_COLUMN_ON_WRITE)
+ case _ => false
+ }
+ }
+
+ /**
+ * Whether the catalog supports generated columns on write (see
+ * [[supportsGeneratedColumnsOnWrite]]) and the given columns include at
least one generated
+ * column.
+ */
+ def supportsGeneratedColumnsOnWrite(
+ catalog: Option[CatalogPlugin],
+ columns: Array[Column]): Boolean = {
+ supportsGeneratedColumnsOnWrite(catalog) &&
+ columns.exists(_.columnGenerationExpression() != null)
+ }
+
+ /**
+ * Returns an attribute with the generation expression metadata removed.
+ * Used when the catalog does not support auto-filling generated columns on
write.
+ */
+ def removeGenerationExpressionMetadata(attr: Attribute): Attribute = {
+ if (isGeneratedColumn(attr.metadata)) {
+ val cleaned = new MetadataBuilder()
+ .withMetadata(attr.metadata)
+ .remove(GENERATION_EXPRESSION_METADATA_KEY)
+ .build()
+ attr.withMetadata(cleaned)
+ } else {
+ attr
+ }
+ }
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
index aaf8310933f8..a4b136d33e6f 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala
@@ -671,6 +671,12 @@ private[sql] object CatalogV2Util {
Option(col.id()).foreach { id =>
f = f.withId(id)
}
+ Option(col.generationExpression()).foreach { genExpr =>
+ f = f.copy(metadata = new MetadataBuilder()
+ .withMetadata(f.metadata)
+ .putString(GeneratedColumn.GENERATION_EXPRESSION_METADATA_KEY, genExpr)
+ .build())
+ }
f
}
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 86a19e45a045..f41337682378 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
@@ -5440,6 +5440,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val GENERATED_COLUMN_ALLOW_NULLABLE_INGEST =
+ buildConf("spark.sql.generatedColumn.allowNullableIngest.enabled")
+ .doc("When true, writing to a table with generated columns allows
omitting nullable " +
+ "non-generated columns from the input. Missing nullable columns are
filled with null. " +
+ "When false, all non-generated columns must be provided.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(true)
+
val SKIP_TYPE_VALIDATION_ON_ALTER_PARTITION =
buildConf("spark.sql.legacy.skipTypeValidationOnAlterPartition")
.internal()
@@ -8664,6 +8674,9 @@ class SQLConf extends Serializable with Logging with
SqlApiConf {
def useNullsForMissingDefaultColumnValues: Boolean =
getConf(SQLConf.USE_NULLS_FOR_MISSING_DEFAULT_COLUMN_VALUES)
+ def generatedColumnAllowNullableIngest: Boolean =
+ getConf(SQLConf.GENERATED_COLUMN_ALLOW_NULLABLE_INGEST)
+
def unionIsResolvedWhenDuplicatesPerChildResolved: Boolean =
getConf(SQLConf.UNION_IS_RESOLVED_WHEN_DUPLICATES_PER_CHILD_RESOLVED)
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
index bb137ba4830d..6466e3b3da30 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
@@ -286,7 +286,8 @@ class InMemoryTableCatalog extends
BasicInMemoryTableCatalog with SupportsNamesp
TableCatalogCapability.SUPPORT_COLUMN_DEFAULT_VALUE,
TableCatalogCapability.SUPPORT_TABLE_CONSTRAINT,
TableCatalogCapability.SUPPORTS_CREATE_TABLE_WITH_GENERATED_COLUMNS,
- TableCatalogCapability.SUPPORTS_CREATE_TABLE_WITH_IDENTITY_COLUMNS
+ TableCatalogCapability.SUPPORTS_CREATE_TABLE_WITH_IDENTITY_COLUMNS,
+ TableCatalogCapability.SUPPORT_GENERATED_COLUMN_ON_WRITE
).asJava
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/ResolveWriteToStream.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/ResolveWriteToStream.scala
index 0be430591dbd..2afb2ad2f4eb 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/ResolveWriteToStream.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/ResolveWriteToStream.scala
@@ -28,6 +28,7 @@ import
org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.streaming.{FlowAssigned,
StreamingRelationV2, UserProvided, WriteToStream, WriteToStreamStatement}
+import org.apache.spark.sql.catalyst.util.GeneratedColumn
import org.apache.spark.sql.connector.catalog.SupportsWrite
import org.apache.spark.sql.errors.{QueryCompilationErrors,
QueryExecutionErrors}
import org.apache.spark.sql.execution.streaming.{ContinuousTrigger,
RealTimeTrigger}
@@ -64,6 +65,14 @@ object ResolveWriteToStream extends Rule[LogicalPlan] {
}
}
+ // Streaming writes with generated columns are not yet supported.
+ s.catalogAndIdent.foreach { case (catalog, ident) =>
+ if (GeneratedColumn.supportsGeneratedColumnsOnWrite(Some(catalog),
s.sink.columns())) {
+ throw QueryCompilationErrors.unsupportedTableOperationError(
+ catalog, ident, "streaming write with generated columns")
+ }
+ }
+
WriteToStream(
s.userSpecifiedName.orNull,
s.userSpecifiedSinkName,
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala
new file mode 100644
index 000000000000..67d253ea01c5
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala
@@ -0,0 +1,1177 @@
+/*
+ * 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.connector
+
+import org.apache.spark.SparkRuntimeException
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.QueryPlanningTracker
+import org.apache.spark.sql.catalyst.expressions.CheckInvariant
+import org.apache.spark.sql.connector.catalog.{InMemoryCatalog,
InMemoryRowLevelOperationTableCatalog,
+ TableCatalogCapability}
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Tests for generated column auto-fill and constraint enforcement during
writes.
+ */
+class GeneratedColumnWriteSuite extends QueryTest with DatasourceV2SQLBase {
+
+ private val rowLevelCat = "rowlevelcat"
+
+ private def withRowLevelCatalog(f: => Unit): Unit = {
+ withSQLConf(s"spark.sql.catalog.$rowLevelCat" ->
+ classOf[InMemoryRowLevelOperationTableCatalog].getName) {
+ f
+ }
+ }
+
+ // A catalog that supports creating generated columns but does NOT declare
+ // SUPPORT_GENERATED_COLUMN_ON_WRITE, so Spark must not auto-fill or enforce
them.
+ private val noWriteCapCat = "nowritecapcat"
+
+ private def withNoWriteCapCatalog(f: => Unit): Unit = {
+ withSQLConf(s"spark.sql.catalog.$noWriteCapCat" ->
+ classOf[InMemoryNoGenColWriteCatalog].getName) {
+ f
+ }
+ }
+
+ private def hasCheckInvariant(sqlText: String): Boolean = {
+ val parsed = spark.sessionState.sqlParser.parsePlan(sqlText)
+ val analyzed = spark.sessionState.analyzer.executeAndCheck(parsed, new
QueryPlanningTracker)
+ analyzed.exists { node =>
+ node.expressions.exists(_.exists(_.isInstanceOf[CheckInvariant]))
+ }
+ }
+
+ private def sqlDate(str: String): java.sql.Date = java.sql.Date.valueOf(str)
+
+ private def sqlTimestamp(str: String): java.sql.Timestamp =
java.sql.Timestamp.valueOf(str)
+
+ // Default Spark DSV2 generated-column write table used by the write-path
cases below.
+ private def createDefaultGeneratedColumnWriteTable(tableName: String): Unit
= {
+ sql(s"""CREATE TABLE testcat.$tableName(
+ | c1 BIGINT,
+ | c2_g BIGINT GENERATED ALWAYS AS (c1 + 10),
+ | c3_p STRING,
+ | c4_g_p DATE GENERATED ALWAYS AS (CAST(c5 AS DATE)),
+ | c5 TIMESTAMP,
+ | c6 INT,
+ | c7_g_p INT GENERATED ALWAYS AS (c6 * 10),
+ | c8 DATE
+ |) USING foo PARTITIONED BY (c3_p, c4_g_p, c7_g_p)""".stripMargin)
+ }
+
+ private def defaultGeneratedColumnWriteRow: Row = {
+ Row(1L, 11L, "foo", sqlDate("2020-10-11"),
+ sqlTimestamp("2020-10-11 12:30:30"), 100, 1000, sqlDate("2020-11-12"))
+ }
+
+ private def testGeneratedColumnWrite(
+ testName: String)(updateFunc: String => Seq[Row]): Unit = {
+ test(s"DSV2 generated column write - $testName") {
+ val tableName = testName
+ withTable(s"testcat.$tableName") {
+ createDefaultGeneratedColumnWriteTable(tableName)
+ val expected = updateFunc(s"testcat.$tableName")
+ checkAnswer(spark.table(s"testcat.$tableName"), expected)
+ }
+ }
+ }
+
+ private def testGeneratedColumnDynamicOverwrite(
+ testName: String)(updateFunc: String => Seq[Row]): Unit = {
+ testGeneratedColumnWrite(s"dpo_$testName") { table =>
+ withSQLConf(SQLConf.PARTITION_OVERWRITE_MODE.key ->
+ SQLConf.PartitionOverwriteMode.DYNAMIC.toString) {
+ updateFunc(table)
+ }
+ }
+ }
+
+ testGeneratedColumnWrite("append_data_v2") { table =>
+ import testImplicits._
+ Seq(Tuple5(1L, "foo", "2020-10-11 12:30:30", 100, "2020-11-12"))
+ .toDF("c1", "c3_p", "c5", "c6", "c8")
+ .withColumn("c5", $"c5".cast("timestamp"))
+ .withColumn("c8", $"c8".cast("date"))
+ .writeTo(table)
+ .append()
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnWrite("append_data_in_different_column_order_v2") { table
=>
+ import testImplicits._
+ Seq(Tuple5("2020-10-11 12:30:30", 100, "2020-11-12", 1L, "foo"))
+ .toDF("c5", "c6", "c8", "c1", "c3_p")
+ .withColumn("c5", $"c5".cast("timestamp"))
+ .withColumn("c8", $"c8".cast("date"))
+ .writeTo(table)
+ .append()
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnWrite("insert_into_values_provide_all_columns") { table =>
+ sql(s"""INSERT INTO $table VALUES (
+ | 1L, 11L, 'foo', DATE'2020-10-11',
+ | TIMESTAMP'2020-10-11 12:30:30', 100, 1000, DATE'2020-11-12'
+ |)""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnWrite("insert_into_by_name_provide_all_columns") { table
=>
+ sql(s"""INSERT INTO $table (c5, c6, c7_g_p, c8, c1, c2_g, c3_p, c4_g_p)
+ |VALUES (
+ | TIMESTAMP'2020-10-11 12:30:30', 100, 1000, DATE'2020-11-12',
+ | 1L, 11L, 'foo', DATE'2020-10-11'
+ |)""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+
testGeneratedColumnWrite("insert_into_by_name_not_provide_generated_columns") {
table =>
+ sql(s"""INSERT INTO $table (c6, c8, c1, c3_p, c5)
+ |VALUES (100, DATE'2020-11-12', 1L, 'foo',
+ | TIMESTAMP'2020-10-11 12:30:30')""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnWrite("insert_into_by_name_with_some_generated_columns")
{ table =>
+ sql(s"""INSERT INTO $table (c5, c6, c8, c1, c3_p, c4_g_p)
+ |VALUES (
+ | TIMESTAMP'2020-10-11 12:30:30', 100, DATE'2020-11-12',
+ | 1L, 'foo', DATE'2020-10-11'
+ |)""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnWrite("insert_into_select_provide_all_columns") { table =>
+ sql(s"""INSERT INTO $table SELECT
+ | 1L, 11L, 'foo', DATE'2020-10-11',
+ | TIMESTAMP'2020-10-11 12:30:30', 100, 1000, DATE'2020-11-12'
+ |""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnWrite("insert_into_by_name_not_provide_normal_columns") {
table =>
+ withSQLConf(SQLConf.GENERATED_COLUMN_ALLOW_NULLABLE_INGEST.key -> "false")
{
+ val ex = intercept[AnalysisException] {
+ sql(s"INSERT INTO $table (c6, c8, c1, c3_p) " +
+ s"VALUES (100, DATE'2020-11-12', 1L, 'foo')")
+ }
+ assert(ex.getMessage.contains("c5"))
+ }
+ Nil
+ }
+
+ testGeneratedColumnWrite("insert_overwrite_values_provide_all_columns") {
table =>
+ sql(s"""INSERT OVERWRITE $table VALUES (
+ | 1L, 11L, 'foo', DATE'2020-10-11',
+ | TIMESTAMP'2020-10-11 12:30:30', 100, 1000, DATE'2020-11-12'
+ |)""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnWrite("insert_overwrite_select_provide_all_columns") {
table =>
+ sql(s"""INSERT OVERWRITE $table SELECT
+ | 1L, 11L, 'foo', DATE'2020-10-11',
+ | TIMESTAMP'2020-10-11 12:30:30', 100, 1000, DATE'2020-11-12'
+ |""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnWrite("insert_overwrite_by_name_provide_all_columns") {
table =>
+ sql(s"""INSERT OVERWRITE $table (c5, c6, c7_g_p, c8, c1, c2_g, c3_p,
c4_g_p)
+ |VALUES (
+ | TIMESTAMP'2020-10-11 12:30:30', 100, 1000, DATE'2020-11-12',
+ | 1L, 11L, 'foo', DATE'2020-10-11'
+ |)""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+
testGeneratedColumnWrite("insert_overwrite_by_name_not_provide_generated_columns")
{ table =>
+ sql(s"""INSERT OVERWRITE $table (c6, c8, c1, c3_p, c5)
+ |VALUES (100, DATE'2020-11-12', 1L, 'foo',
+ | TIMESTAMP'2020-10-11 12:30:30')""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+
testGeneratedColumnWrite("insert_overwrite_by_name_with_some_generated_columns")
{ table =>
+ sql(s"""INSERT OVERWRITE $table (c5, c6, c8, c1, c3_p, c4_g_p)
+ |VALUES (
+ | TIMESTAMP'2020-10-11 12:30:30', 100, DATE'2020-11-12',
+ | 1L, 'foo', DATE'2020-10-11'
+ |)""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+
testGeneratedColumnWrite("insert_overwrite_by_name_not_provide_normal_columns")
{ table =>
+ withSQLConf(SQLConf.GENERATED_COLUMN_ALLOW_NULLABLE_INGEST.key -> "false")
{
+ val ex = intercept[AnalysisException] {
+ sql(s"INSERT OVERWRITE $table (c6, c8, c1, c3_p) " +
+ s"VALUES (100, DATE'2020-11-12', 1L, 'foo')")
+ }
+ assert(ex.getMessage.contains("c5"))
+ }
+ Nil
+ }
+
+
testGeneratedColumnDynamicOverwrite("insert_overwrite_values_provide_all_columns")
{ table =>
+ sql(s"""INSERT OVERWRITE $table VALUES (
+ | 1L, 11L, 'foo', DATE'2020-10-11',
+ | TIMESTAMP'2020-10-11 12:30:30', 100, 1000, DATE'2020-11-12'
+ |)""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+
testGeneratedColumnDynamicOverwrite("insert_overwrite_select_provide_all_columns")
{ table =>
+ sql(s"""INSERT OVERWRITE $table SELECT
+ | 1L, 11L, 'foo', DATE'2020-10-11',
+ | TIMESTAMP'2020-10-11 12:30:30', 100, 1000, DATE'2020-11-12'
+ |""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnDynamicOverwrite(
+ "insert_overwrite_by_name_values_provide_all_columns") { table =>
+ sql(s"""INSERT OVERWRITE $table (c5, c6, c7_g_p, c8, c1, c2_g, c3_p,
c4_g_p)
+ |VALUES (
+ | TIMESTAMP'2020-10-11 12:30:30', 100, 1000, DATE'2020-11-12',
+ | 1L, 11L, 'foo', DATE'2020-10-11'
+ |)""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnDynamicOverwrite(
+ "insert_overwrite_by_name_not_provide_generated_columns") { table =>
+ sql(s"""INSERT OVERWRITE $table (c6, c8, c1, c3_p, c5)
+ |VALUES (100, DATE'2020-11-12', 1L, 'foo',
+ | TIMESTAMP'2020-10-11 12:30:30')""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnDynamicOverwrite(
+ "insert_overwrite_by_name_with_some_generated_columns") { table =>
+ sql(s"""INSERT OVERWRITE $table (c5, c6, c8, c1, c3_p, c4_g_p)
+ |VALUES (
+ | TIMESTAMP'2020-10-11 12:30:30', 100, DATE'2020-11-12',
+ | 1L, 'foo', DATE'2020-10-11'
+ |)""".stripMargin)
+ defaultGeneratedColumnWriteRow :: Nil
+ }
+
+ testGeneratedColumnDynamicOverwrite(
+ "insert_overwrite_by_name_not_provide_normal_columns") { table =>
+ withSQLConf(SQLConf.GENERATED_COLUMN_ALLOW_NULLABLE_INGEST.key -> "false")
{
+ val ex = intercept[AnalysisException] {
+ sql(s"INSERT OVERWRITE $table (c6, c8, c1, c3_p) " +
+ s"VALUES (100, DATE'2020-11-12', 1L, 'foo')")
+ }
+ assert(ex.getMessage.contains("c5"))
+ }
+ Nil
+ }
+
+ test("INSERT by name auto-fills missing generated column") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | eventDate DATE,
+ | eventYear INT GENERATED ALWAYS AS (year(eventDate))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(eventDate) VALUES (DATE'2024-06-15')")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(java.sql.Date.valueOf("2024-06-15"), 2024))
+ }
+ }
+
+ test("INSERT by name computes generated column from the stored (post-cast)
value") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | id INT,
+ | doubled INT GENERATED ALWAYS AS (id * 2)
+ |) USING foo""".stripMargin)
+ // 2.9 is cast to INT (= 2) when stored in `id`. The generated column
must be computed from
+ // the stored value (2 * 2 = 4), not from the pre-cast value (2.9 * 2 =
5.8 -> 5). This
+ // mirrors the by-position path, which resolves the generation
expression against the
+ // post-cast columns.
+ sql(s"INSERT INTO testcat.$tblName(id) VALUES (2.9)")
+ checkAnswer(spark.table(s"testcat.$tblName"), Row(2, 4))
+ }
+ }
+
+ test("INSERT by position validates generated column against the stored
(post-cast) value") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | id INT,
+ | doubled INT GENERATED ALWAYS AS (id * 2)
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName VALUES (2.9, 4)")
+ checkAnswer(spark.table(s"testcat.$tblName"), Row(2, 4))
+ }
+ }
+
+ test("INSERT by name with matching explicit value succeeds") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | eventDate DATE,
+ | eventYear INT GENERATED ALWAYS AS (year(eventDate))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(eventDate, eventYear) VALUES
(DATE'2024-06-15', 2024)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(java.sql.Date.valueOf("2024-06-15"), 2024))
+ }
+ }
+
+ test("INSERT by name with non-matching explicit value fails") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | eventDate DATE,
+ | eventYear INT GENERATED ALWAYS AS (year(eventDate))
+ |) USING foo""".stripMargin)
+ val ex = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO testcat.$tblName(eventDate, eventYear) VALUES
(DATE'2024-06-15', 2025)")
+ }
+ assert(ex.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ }
+ }
+
+ test("INSERT by position does not auto-fill trailing generated column") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | eventDate DATE,
+ | eventYear INT GENERATED ALWAYS AS (year(eventDate))
+ |) USING foo""".stripMargin)
+ // By-position writes require values for all target columns. To omit
generated columns,
+ // use a by-name write with an explicit column list.
+ val ex = intercept[AnalysisException] {
+ sql(s"INSERT INTO testcat.$tblName VALUES (DATE'2024-06-15')")
+ }
+ assert(ex.getMessage.contains("not enough data columns"))
+ }
+ }
+
+ test("INSERT by position with matching explicit value succeeds") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | eventDate DATE,
+ | eventYear INT GENERATED ALWAYS AS (year(eventDate))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName VALUES (DATE'2024-06-15', 2024)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(java.sql.Date.valueOf("2024-06-15"), 2024))
+ }
+ }
+
+ test("INSERT by position with non-matching explicit value fails") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | eventDate DATE,
+ | eventYear INT GENERATED ALWAYS AS (year(eventDate))
+ |) USING foo""".stripMargin)
+ val ex = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO testcat.$tblName VALUES (DATE'2024-06-15', 2025)")
+ }
+ assert(ex.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ }
+ }
+
+ test("INSERT auto-fills multiple generated columns") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | eventDate DATE,
+ | eventYear INT GENERATED ALWAYS AS (year(eventDate)),
+ | eventMonth INT GENERATED ALWAYS AS (month(eventDate))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(eventDate) VALUES (DATE'2024-06-15')")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(java.sql.Date.valueOf("2024-06-15"), 2024, 6))
+ }
+ }
+
+ test("INSERT with expression referencing multiple columns") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT,
+ | c INT GENERATED ALWAYS AS (a + b)
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(a, b) VALUES (3, 5)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(3, 5, 8))
+ }
+ }
+
+ test("allowNullableIngest config controls missing non-generated columns") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b STRING,
+ | c INT GENERATED ALWAYS AS (a + 1)
+ |) USING foo""".stripMargin)
+ // Config ON (default): missing nullable column 'b' is filled with null
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES (1)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(1, null, 2))
+ // Config OFF: missing nullable column 'b' causes an error
+ withSQLConf(SQLConf.GENERATED_COLUMN_ALLOW_NULLABLE_INGEST.key ->
"false") {
+ val ex = intercept[AnalysisException] {
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES (2)")
+ }
+ assert(ex.getMessage.contains("b"))
+ }
+ }
+ }
+
+ test("DataFrame writeTo append auto-fills generated columns with input
columns reordered") {
+ import testImplicits._
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1),
+ | c STRING,
+ | d INT GENERATED ALWAYS AS (length(c))
+ |) USING foo""".stripMargin)
+ Seq(("hello", 7), ("spark", 3)).toDF("c",
"a").writeTo(s"testcat.$tblName").append()
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(7, 8, "hello", 5) :: Row(3, 4, "spark", 5) :: Nil)
+ }
+ }
+
+ test("DataFrame writeTo append obeys allowNullableIngest for missing
non-generated column") {
+ import testImplicits._
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | optional STRING,
+ | b INT GENERATED ALWAYS AS (a + 1)
+ |) USING foo""".stripMargin)
+
+ Seq((1, "one")).toDF("a",
"optional").writeTo(s"testcat.$tblName").append()
+ Seq(2).toDF("a").writeTo(s"testcat.$tblName").append()
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(1, "one", 2) :: Row(2, null, 3) :: Nil)
+
+ withSQLConf(SQLConf.GENERATED_COLUMN_ALLOW_NULLABLE_INGEST.key ->
"false") {
+ val ex = intercept[AnalysisException] {
+ Seq(3).toDF("a").writeTo(s"testcat.$tblName").append()
+ }
+ assert(ex.getMessage.contains("optional"))
+ }
+ }
+ }
+
+ test("missing NOT NULL column fails with generated columns") {
+ import testImplicits._
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT NOT NULL,
+ | required STRING NOT NULL,
+ | b INT GENERATED ALWAYS AS (a + 1) NOT NULL
+ |) USING foo""".stripMargin)
+
+ val ex = intercept[AnalysisException] {
+ Seq(1).toDF("a").writeTo(s"testcat.$tblName").append()
+ }
+ assert(ex.getMessage.contains("required"))
+ }
+ }
+
+ test("works alongside table CHECK constraints") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1)
+ |) USING foo""".stripMargin)
+ sql(s"ALTER TABLE testcat.$tblName ADD CONSTRAINT positive_a CHECK (a >
0)")
+
+ // Both constraints pass: a > 0 and b = a + 1 (auto-filled)
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES (5)")
+ checkAnswer(spark.table(s"testcat.$tblName"), Row(5, 6))
+
+ // Table CHECK constraint fails: a <= 0
+ val ex1 = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES (-1)")
+ }
+ assert(ex1.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ assert(ex1.getMessage.contains("positive_a"))
+
+ // Generated column constraint fails: user provides wrong b
+ val ex2 = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO testcat.$tblName(a, b) VALUES (5, 999)")
+ }
+ assert(ex2.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ assert(ex2.getMessage.contains("Generated Column"))
+ }
+ }
+
+ test("NULL input produces NULL generated column value") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | eventDate DATE,
+ | eventYear INT GENERATED ALWAYS AS (year(eventDate))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(eventDate) VALUES (NULL)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(null, null))
+ }
+ }
+
+ test("type coercion in generated column expression") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b LONG GENERATED ALWAYS AS (a + 1)
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES (100)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(100, 101L))
+ }
+ }
+
+ test("multiple rows each get their own generated value") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a * 10)
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES (1), (2), (3)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(1, 10) :: Row(2, 20) :: Row(3, 30) :: Nil)
+ }
+ }
+
+ test("generated column in the middle of schema") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1),
+ | c STRING
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(a, c) VALUES (5, 'hello')")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(5, 6, "hello"))
+ }
+ }
+
+ test("NULL explicit value matching NULL generation result succeeds") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | eventDate DATE,
+ | eventYear INT GENERATED ALWAYS AS (year(eventDate))
+ |) USING foo""".stripMargin)
+ // NULL <=> year(NULL) -> NULL <=> NULL -> true (EqualNullSafe)
+ sql(s"INSERT INTO testcat.$tblName(eventDate, eventYear) VALUES (NULL,
NULL)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(null, null))
+ }
+ }
+
+ test("NULL explicit value for generated column") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1)
+ |) USING foo""".stripMargin)
+ // NULL <=> (a + 1) where a=5 -> NULL <=> 6 -> false -> violation
+ val ex = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO testcat.$tblName(a, b) VALUES (5, NULL)")
+ }
+ assert(ex.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ }
+ }
+
+ test("INSERT OVERWRITE with generated columns") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1)
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES (1)")
+ sql(s"INSERT OVERWRITE testcat.$tblName(a) VALUES (10)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(10, 11))
+ }
+ }
+
+ test("INSERT OVERWRITE by name with all generated columns validates
constraints") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1),
+ | c STRING,
+ | d INT GENERATED ALWAYS AS (length(c))
+ |) USING foo""".stripMargin)
+
+ sql(s"INSERT INTO testcat.$tblName(a, c) VALUES (1, 'old')")
+ sql(s"INSERT OVERWRITE testcat.$tblName(d, c, b, a) VALUES (3, 'new', 8,
7)")
+ checkAnswer(spark.table(s"testcat.$tblName"), Row(7, 8, "new", 3))
+
+ val ex = intercept[SparkRuntimeException] {
+ sql(s"INSERT OVERWRITE testcat.$tblName(d, c, b, a) VALUES (3, 'bad',
99, 7)")
+ }
+ assert(ex.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ assert(ex.getMessage.contains("Generated Column"))
+ }
+ }
+
+ test("INSERT OVERWRITE by name with some generated columns auto-fills the
rest") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1),
+ | c INT GENERATED ALWAYS AS (a * 10)
+ |) USING foo""".stripMargin)
+
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES (1)")
+ sql(s"INSERT OVERWRITE testcat.$tblName(c, a) VALUES (70, 7)")
+ checkAnswer(spark.table(s"testcat.$tblName"), Row(7, 8, 70))
+
+ val ex = intercept[SparkRuntimeException] {
+ sql(s"INSERT OVERWRITE testcat.$tblName(c, a) VALUES (99, 7)")
+ }
+ assert(ex.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ }
+ }
+
+ test("INSERT OVERWRITE SELECT with all generated column values validates
constraints") {
+ val tblName = "my_tab"
+ val srcName = "src_tab"
+ withTable(s"testcat.$tblName", s"testcat.$srcName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1),
+ | c STRING
+ |) USING foo""".stripMargin)
+ sql(s"CREATE TABLE testcat.$srcName(a INT, b INT, c STRING) USING foo")
+ sql(s"INSERT INTO testcat.$tblName(a, c) VALUES (0, 'old')")
+ sql(s"INSERT INTO testcat.$srcName VALUES (1, 2, 'ok')")
+ sql(s"INSERT OVERWRITE testcat.$tblName SELECT a, b, c FROM
testcat.$srcName")
+ checkAnswer(spark.table(s"testcat.$tblName"), Row(1, 2, "ok"))
+
+ sql(s"INSERT OVERWRITE testcat.$srcName VALUES (2, 99, 'bad')")
+ val ex = intercept[SparkRuntimeException] {
+ sql(s"INSERT OVERWRITE testcat.$tblName SELECT a, b, c FROM
testcat.$srcName")
+ }
+ assert(ex.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ assert(ex.getMessage.contains("Generated Column"))
+ }
+ }
+
+ test("DataFrame writeTo append with missing generated column") {
+ import testImplicits._
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a * 3)
+ |) USING foo""".stripMargin)
+ Seq(4, 5).toDF("a").writeTo(s"testcat.$tblName").append()
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(4, 12) :: Row(5, 15) :: Nil)
+ }
+ }
+
+ test("INSERT by position does not auto-fill non-trailing generated column") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT GENERATED ALWAYS AS (b + 1),
+ | b INT
+ |) USING foo""".stripMargin)
+ // By-position writes require values for all target columns. To omit
generated columns,
+ // use a by-name write with an explicit column list.
+ val ex = intercept[AnalysisException] {
+ sql(s"INSERT INTO testcat.$tblName VALUES (10)")
+ }
+ assert(ex.getMessage.contains("not enough data columns"))
+ }
+ }
+
+ test("INSERT by name with columns in different order") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1),
+ | c STRING
+ |) USING foo""".stripMargin)
+ // Provide columns in reverse order
+ sql(s"INSERT INTO testcat.$tblName(c, a) VALUES ('hello', 7)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(7, 8, "hello"))
+ }
+ }
+
+ test("INSERT SELECT auto-fills generated columns") {
+ val tblName = "my_tab"
+ val srcName = "src_tab"
+ withTable(s"testcat.$tblName", s"testcat.$srcName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a * 10)
+ |) USING foo""".stripMargin)
+ sql(s"CREATE TABLE testcat.$srcName(a INT) USING foo")
+ sql(s"INSERT INTO testcat.$srcName VALUES (1), (2), (3)")
+ sql(s"INSERT INTO testcat.$tblName(a) SELECT a FROM testcat.$srcName")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(1, 10) :: Row(2, 20) :: Row(3, 30) :: Nil)
+ }
+ }
+
+ test("INSERT SELECT with all generated column values validates constraints")
{
+ val tblName = "my_tab"
+ val srcName = "src_tab"
+ withTable(s"testcat.$tblName", s"testcat.$srcName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1),
+ | c STRING
+ |) USING foo""".stripMargin)
+ sql(s"CREATE TABLE testcat.$srcName(a INT, b INT, c STRING) USING foo")
+ sql(s"INSERT INTO testcat.$srcName VALUES (1, 2, 'ok')")
+ sql(s"INSERT INTO testcat.$tblName SELECT a, b, c FROM testcat.$srcName")
+ checkAnswer(spark.table(s"testcat.$tblName"), Row(1, 2, "ok"))
+
+ sql(s"INSERT OVERWRITE testcat.$srcName VALUES (2, 99, 'bad')")
+ val ex = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO testcat.$tblName SELECT a, b, c FROM
testcat.$srcName")
+ }
+ assert(ex.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ assert(ex.getMessage.contains("Generated Column"))
+ }
+ }
+
+ test("INSERT with complex generation expression using CAST") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | ts TIMESTAMP,
+ | ts_date DATE GENERATED ALWAYS AS (CAST(ts AS DATE))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(ts) VALUES (TIMESTAMP'2024-06-15
10:30:00')")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(java.sql.Timestamp.valueOf("2024-06-15 10:30:00"),
+ java.sql.Date.valueOf("2024-06-15")))
+ }
+ }
+
+ test("INSERT with case-insensitive column matching") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | EventDate DATE,
+ | EventYear INT GENERATED ALWAYS AS (year(EventDate))
+ |) USING foo""".stripMargin)
+ // Use different case in INSERT column list
+ sql(s"INSERT INTO testcat.$tblName(eventdate) VALUES (DATE'2024-06-15')")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(java.sql.Date.valueOf("2024-06-15"), 2024))
+ }
+ }
+
+ test("INSERT missing required non-generated column fails") {
+ val tblName = "my_tab"
+ withSQLConf(SQLConf.GENERATED_COLUMN_ALLOW_NULLABLE_INGEST.key -> "false")
{
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b STRING,
+ | c INT GENERATED ALWAYS AS (a + 1)
+ |) USING foo""".stripMargin)
+ // Missing non-generated, non-nullable column 'b' should fail
+ // when allowNullableIngest is off
+ val ex = intercept[AnalysisException] {
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES (1)")
+ }
+ assert(ex.getMessage.contains("b"))
+ }
+ }
+ }
+
+ test("INSERT with generated partition column") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | eventDate DATE,
+ | eventYear INT GENERATED ALWAYS AS (year(eventDate))
+ |) USING foo PARTITIONED BY (eventYear)""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(eventDate) VALUES (DATE'2024-06-15')")
+ sql(s"INSERT INTO testcat.$tblName(eventDate) VALUES (DATE'2023-01-01')")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(java.sql.Date.valueOf("2024-06-15"), 2024) ::
+ Row(java.sql.Date.valueOf("2023-01-01"), 2023) :: Nil)
+ }
+ }
+
+ test("CTAS auto-fills generated columns") {
+ val src = "src_tab"
+ val tgt = "tgt_tab"
+ withTable(s"testcat.$src", s"testcat.$tgt") {
+ sql(s"CREATE TABLE testcat.$src(a INT, b INT) USING foo")
+ sql(s"INSERT INTO testcat.$src VALUES (3, 5)")
+ sql(s"""CREATE TABLE testcat.$tgt(
+ | a INT,
+ | b INT,
+ | c INT GENERATED ALWAYS AS (a + b)
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tgt(a, b) SELECT a, b FROM testcat.$src")
+ checkAnswer(
+ spark.table(s"testcat.$tgt"),
+ Row(3, 5, 8))
+ }
+ }
+
+ test("generated column constraint supports null values") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | c1 STRING,
+ | c2 STRING GENERATED ALWAYS AS (concat(c1, 'y'))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName VALUES ('x', 'xy')")
+ sql(s"INSERT INTO testcat.$tblName VALUES (NULL, NULL)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row("x", "xy") :: Row(null, null) :: Nil)
+
+ val ex1 = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO testcat.$tblName VALUES ('foo', NULL)")
+ }
+ assert(ex1.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+
+ val ex2 = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO testcat.$tblName VALUES (NULL, 'foo')")
+ }
+ assert(ex2.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ }
+ }
+
+ test("generated column with dotted column names and case-insensitive insert
columns") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | `a.b` STRING GENERATED ALWAYS AS (concat(`c.d`, 'y')),
+ | `c.d` STRING
+ |) USING foo""".stripMargin)
+
+ sql(s"INSERT INTO testcat.$tblName VALUES ('xy', 'x')")
+ sql(s"INSERT INTO testcat.$tblName(`c.D`, `a.B`) VALUES ('x', 'xy')")
+ sql(s"INSERT INTO testcat.$tblName(`c.D`) VALUES ('x')")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row("xy", "x") :: Row("xy", "x") :: Row("xy", "x") :: Nil)
+
+ val ex = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO testcat.$tblName(`a.B`) VALUES ('xy')")
+ }
+ assert(ex.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ }
+ }
+
+ test("generated column with complex type extractors") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | `a.b` STRING,
+ | a STRUCT<b: INT, c: STRING>,
+ | array ARRAY<INT>,
+ | c1 STRING GENERATED ALWAYS AS (concat(`a.b`, 'b')),
+ | c2 INT GENERATED ALWAYS AS (a.b + 100),
+ | c3 INT GENERATED ALWAYS AS (array[1])
+ |) USING foo""".stripMargin)
+ sql(s"""INSERT INTO testcat.$tblName(`a.b`, a, array)
+ |VALUES ('a', named_struct('b', 100, 'c', 'foo'), array(1000,
1001))
+ |""".stripMargin)
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row("a", Row(100, "foo"), Seq(1000, 1001), "ab", 200, 1001))
+ }
+ }
+
+ test("TIME type in generated columns is not supported") {
+ val tblName = "my_tab"
+ withSQLConf(SQLConf.TIME_TYPE_ENABLED.key -> "true") {
+ withTable(s"testcat.$tblName") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | id INT,
+ | event_time TIME GENERATED ALWAYS AS (CAST('12:00:00' AS
TIME))
+ |) USING foo""".stripMargin)
+ },
+ condition = "UNSUPPORTED_EXPRESSION_GENERATED_COLUMN",
+ parameters = Map(
+ "fieldName" -> "event_time",
+ "expressionStr" -> "CAST('12:00:00' AS TIME)",
+ "reason" -> "TIME type is not supported in generated columns"))
+ }
+ }
+ }
+
+ test("streaming write with generated columns is blocked") {
+ import testImplicits._
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ withTempDir { checkpointDir =>
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | id INT,
+ | doubled INT GENERATED ALWAYS AS (id * 2)
+ |) USING foo""".stripMargin)
+ val inputData = MemoryStream[Int]
+ val df = inputData.toDF().toDF("id")
+ val ex = intercept[AnalysisException] {
+ df.writeStream
+ .option("checkpointLocation", checkpointDir.getAbsolutePath)
+ .toTable(s"testcat.$tblName")
+ }
+ assert(ex.getCondition == "UNSUPPORTED_FEATURE.TABLE_OPERATION")
+ assert(ex.getMessage.contains("streaming"))
+ assert(ex.getMessage.contains("generated columns"))
+ }
+ }
+ }
+
+ test("INSERT by position validates generated column with type cast on source
column") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ // Table expects LONG for 'a', generation expression is a + 1
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a LONG,
+ | b LONG GENERATED ALWAYS AS (a + 1)
+ |) USING foo""".stripMargin)
+ // Insert INT values by position -- 'a' gets cast to LONG, then the
generated column
+ // constraint validates against the cast value.
+ sql(s"INSERT INTO testcat.$tblName VALUES (42, 43)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(42L, 43L))
+ }
+ }
+
+ test("mix of auto-filled and user-provided generated columns") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1),
+ | c INT GENERATED ALWAYS AS (a * 10)
+ |) USING foo""".stripMargin)
+ // Provide 'a' and 'c' (user-provided), omit 'b' (auto-filled)
+ // b is auto-filled with a + 1 = 6
+ // c is user-provided with correct value a * 10 = 50 -> passes constraint
+ sql(s"INSERT INTO testcat.$tblName(a, c) VALUES (5, 50)")
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row(5, 6, 50))
+
+ // Now provide wrong value for c -> constraint violation on c only
+ val ex = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO testcat.$tblName(a, c) VALUES (5, 999)")
+ }
+ assert(ex.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ }
+ }
+
+ test("plan has constraint for user-provided but not auto-filled generated
columns") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1),
+ | c INT GENERATED ALWAYS AS (a * 10)
+ |) USING foo""".stripMargin)
+
+ // Auto-filled: no CheckInvariant in analyzed plan
+ assert(!hasCheckInvariant(s"INSERT INTO testcat.$tblName(a) VALUES (5)"),
+ "Auto-filled generated columns should not have CheckInvariant in plan")
+
+ // User-provided: CheckInvariant should appear
+ assert(hasCheckInvariant(s"INSERT INTO testcat.$tblName(a, b) VALUES (5,
6)"),
+ "User-provided generated columns should have CheckInvariant in plan")
+ }
+ }
+
+ // MERGE/UPDATE with generated columns are blocked for now; DELETE is
allowed.
+
+ test("UPDATE with generated columns is blocked") {
+ withRowLevelCatalog {
+ val tblName = "my_tab"
+ withTable(s"$rowLevelCat.$tblName") {
+ sql(s"""CREATE TABLE $rowLevelCat.$tblName(
+ | id INT,
+ | data STRING,
+ | doubled INT GENERATED ALWAYS AS (id * 2)
+ |) USING foo""".stripMargin)
+ val ex = intercept[AnalysisException] {
+ sql(s"UPDATE $rowLevelCat.$tblName SET data = 'x' WHERE id = 1")
+ }
+ assert(ex.getCondition == "UNSUPPORTED_FEATURE.TABLE_OPERATION")
+ assert(ex.getMessage.contains("UPDATE with generated columns"))
+ }
+ }
+ }
+
+ test("MERGE with generated columns is blocked") {
+ withRowLevelCatalog {
+ val tblName = "my_tab"
+ withTable(s"$rowLevelCat.$tblName") {
+ sql(s"""CREATE TABLE $rowLevelCat.$tblName(
+ | id INT,
+ | data STRING,
+ | doubled INT GENERATED ALWAYS AS (id * 2)
+ |) USING foo""".stripMargin)
+ val ex = intercept[AnalysisException] {
+ sql(
+ s"""MERGE INTO $rowLevelCat.$tblName AS t
+ |USING source AS s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN UPDATE SET t.data = s.data
+ |WHEN NOT MATCHED THEN INSERT (id, data) VALUES (s.id, s.data)
+ |""".stripMargin)
+ }
+ assert(ex.getCondition == "UNSUPPORTED_FEATURE.TABLE_OPERATION")
+ assert(ex.getMessage.contains("MERGE with generated columns"))
+ }
+ }
+ }
+
+ test("MERGE with only NOT MATCHED actions and generated columns is blocked")
{
+ withRowLevelCatalog {
+ val tblName = "my_tab"
+ withTable(s"$rowLevelCat.$tblName") {
+ sql(s"""CREATE TABLE $rowLevelCat.$tblName(
+ | id INT,
+ | data STRING,
+ | doubled INT GENERATED ALWAYS AS (id * 2)
+ |) USING foo""".stripMargin)
+ val ex = intercept[AnalysisException] {
+ sql(
+ s"""MERGE INTO $rowLevelCat.$tblName AS t
+ |USING source AS s
+ |ON t.id = s.id
+ |WHEN NOT MATCHED THEN INSERT (id, data) VALUES (s.id, s.data)
+ |""".stripMargin)
+ }
+ assert(ex.getCondition == "UNSUPPORTED_FEATURE.TABLE_OPERATION")
+ assert(ex.getMessage.contains("MERGE with generated columns"))
+ }
+ }
+ }
+
+ test("DELETE with generated columns is allowed") {
+ withRowLevelCatalog {
+ val tblName = "my_tab"
+ withTable(s"$rowLevelCat.$tblName") {
+ // Partition by id so the metadata delete-by-filter path works on the
in-memory table.
+ sql(s"""CREATE TABLE $rowLevelCat.$tblName(
+ | id INT,
+ | data STRING,
+ | doubled INT GENERATED ALWAYS AS (id * 2)
+ |) USING foo PARTITIONED BY (id)""".stripMargin)
+ sql(s"INSERT INTO $rowLevelCat.$tblName(id, data) VALUES (1, 'a'), (2,
'b')")
+ // DELETE is not blocked (only MERGE/UPDATE are), and auto-fill still
applies on INSERT.
+ sql(s"DELETE FROM $rowLevelCat.$tblName WHERE id = 1")
+ checkAnswer(spark.table(s"$rowLevelCat.$tblName"), Row(2, "b", 4))
+ }
+ }
+ }
+
+ test("catalog without write capability does not auto-fill or enforce
generated columns") {
+ withNoWriteCapCatalog {
+ val tblName = "my_tab"
+ withTable(s"$noWriteCapCat.$tblName") {
+ sql(s"""CREATE TABLE $noWriteCapCat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (a + 1)
+ |) USING foo""".stripMargin)
+
+ // A user-provided value that does NOT match the generation expression
is written
+ // as-is: no constraint is enforced because the catalog does not opt
in.
+ sql(s"INSERT INTO $noWriteCapCat.$tblName(a, b) VALUES (5, 999)")
+ checkAnswer(spark.table(s"$noWriteCapCat.$tblName"), Row(5, 999))
+
+ // Omitting the generated column does not auto-fill; it is treated as
a regular
+ // nullable column and filled with null.
+ sql(s"INSERT INTO $noWriteCapCat.$tblName(a) VALUES (7)")
+ checkAnswer(
+ spark.table(s"$noWriteCapCat.$tblName"),
+ Row(5, 999) :: Row(7, null) :: Nil)
+
+ // No CheckInvariant is added to the plan for the generated column.
+ assert(!hasCheckInvariant(s"INSERT INTO $noWriteCapCat.$tblName(a, b)
VALUES (5, 6)"),
+ "No CheckInvariant should be added when the catalog lacks the write
capability")
+ }
+ }
+ }
+}
+
+/**
+ * A catalog that supports creating tables with generated columns but does NOT
declare
+ * [[TableCatalogCapability.SUPPORT_GENERATED_COLUMN_ON_WRITE]], so Spark
leaves generated
+ * column handling to the connector (no auto-fill, no constraint enforcement
on write).
+ */
+class InMemoryNoGenColWriteCatalog extends InMemoryCatalog {
+ override def capabilities: java.util.Set[TableCatalogCapability] = {
+ val caps = new
java.util.HashSet[TableCatalogCapability](super.capabilities)
+ caps.remove(TableCatalogCapability.SUPPORT_GENERATED_COLUMN_ON_WRITE)
+ caps
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
