You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "szehon-ho (via GitHub)" <gi...@apache.org> on 2026/03/09 23:41:49 UTC
[PR] [SPARk-55903][SQL] Simplify Merge Into Schema Evolution and Check Write Privileges [spark]
szehon-ho opened a new pull request, #54704:
URL: https://github.com/apache/spark/pull/54704
### What changes were proposed in this pull request?
Some simplification for Merge Into Schema Evolution and minor bug fixes
- The biggest cleanup is getting rid of 'needSchemaEvolution', 'canEvaluateSchemaEvolution', 'changesForSchemaEvolution'. The three-part state is because the rule to evaluate schema evolution needed the analyzer to resolve all the references it can (minus the un-resolved target references in the query that would be solved by schema evolution). Thus it needed a weird 'canEvaluateSchemaEvolution' state to block until that happened. Now, the code has a simple 'pendingChanges' that is empty in both cases 1) when the analyzer has not resolved enough references yet for the schema evolution code to proceed, and 2) we have enough information but decide we do not need schema evolution because there are no valid changes.
- Load the table from catalog with write Privileges, previously it was not doing so and could be performing schema evolution without the privilege
- Catch and wrap the exception properly
### Why are the changes needed?
Discussed with @aokolnychyi on the state of Spark 4.1 schema evolution , and he suggested these changes.
### Does this PR introduce _any_ user-facing change?
Write privilege is now enforced (if any system used DSV2 privileges). Error message are changed.
### How was this patch tested?
Run existing unit tests
### Was this patch authored or co-authored using generative AI tooling?
No
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #54704:
URL: https://github.com/apache/spark/pull/54704#discussion_r2916749779
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1106,11 +1096,43 @@ object MergeIntoTable {
// Example: UPDATE SET target.a = source.a
private def isEqual(assignment: Assignment, sourceFieldPath: Seq[String]): Boolean = {
// key must be a non-qualified field path that may be added to target schema via evolution
- val assignmenKeyExpr = extractFieldPath(assignment.key, allowUnresolved = true)
+ val assignmentKeyExpr = extractFieldPath(assignment.key, allowUnresolved = true)
// value should always be resolved (from source)
val assignmentValueExpr = extractFieldPath(assignment.value, allowUnresolved = false)
- assignmenKeyExpr == assignmentValueExpr &&
- assignmenKeyExpr == sourceFieldPath
+ assignmentKeyExpr == assignmentValueExpr &&
Review Comment:
Minor: Have these two on the same line?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARk-55903][SQL] Simplify Merge Into Schema Evolution and Check Write Privileges [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #54704:
URL: https://github.com/apache/spark/pull/54704#discussion_r2910352546
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
+ catalogName: String,
+ changes: Array[TableChange]): Throwable = {
+ val tableName = (catalogName +: ident.asMultipartIdentifier).map(_.replaceAll("\"", ""))
Review Comment:
What is this `replaceAll` doing and why is it needed here but not in other error classes?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -1601,7 +1599,7 @@ class Analyzer(
InsertAction(
resolvedInsertCondition,
resolveAssignments(assignments, m, MergeResolvePolicy.SOURCE,
- throws = throws))
+ throws))
Review Comment:
Move to line above?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
+ catalogName: String,
+ changes: Array[TableChange]): Throwable = {
+ val tableName = (catalogName +: ident.asMultipartIdentifier).map(_.replaceAll("\"", ""))
val changesDesc = changes.map(_.toString).mkString("; ")
new AnalysisException(
- errorClass = "UNSUPPORTED_TABLE_CHANGES_IN_AUTO_SCHEMA_EVOLUTION",
+ errorClass = "UNSUPPORTED_TABLE_CHANGES_IN_AUTO_SCHEMA_EVOLUTION.CATALOG_PARTIAL",
messageParameters = Map(
- "changes" -> changesDesc, "tableName" -> toSQLId(sanitizedTableName)))
+ "tableName" -> toSQLId(tableName), "changes" -> changesDesc))
+ }
+
+ def failedToEvolveSchemaDuringMergeError(
Review Comment:
What about making it more generic so that it also works for INSERT later?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
+ catalogName: String,
+ changes: Array[TableChange]): Throwable = {
+ val tableName = (catalogName +: ident.asMultipartIdentifier).map(_.replaceAll("\"", ""))
val changesDesc = changes.map(_.toString).mkString("; ")
Review Comment:
Can we simplify this to `changes.mkString("; ")` or is it not safe?
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -7660,8 +7660,20 @@
},
"UNSUPPORTED_TABLE_CHANGES_IN_AUTO_SCHEMA_EVOLUTION" : {
Review Comment:
I wonder whether we should reconsider the error message classes. What about below?
```
UNSUPPORTED_AUTO_SCHEMA_EVOLUTION_CHANGES.FAILED_EVOLUTION
UNSUPPORTED_AUTO_SCHEMA_EVOLUTION_CHANGES.PARTIAL_EVOLUTION
```
Or anything similar (shorter).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
+ catalogName: String,
+ changes: Array[TableChange]): Throwable = {
Review Comment:
Why not accept Seq instead of Array to avoid the conversion while calling?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala:
##########
@@ -45,8 +45,8 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand with PredicateHelper
override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
case m @ MergeIntoTable(aliasedTable, source, cond, matchedActions, notMatchedActions,
- notMatchedBySourceActions, _) if m.resolved && m.rewritable && m.aligned &&
- !m.needSchemaEvolution && matchedActions.isEmpty && notMatchedActions.size == 1 &&
Review Comment:
I know we talked about this. Just to confirm. Removing `!m.needSchemaEvolution` is safe as `aligned` would be false otherwise?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
+ catalogName: String,
+ changes: Array[TableChange]): Throwable = {
+ val tableName = (catalogName +: ident.asMultipartIdentifier).map(_.replaceAll("\"", ""))
val changesDesc = changes.map(_.toString).mkString("; ")
new AnalysisException(
- errorClass = "UNSUPPORTED_TABLE_CHANGES_IN_AUTO_SCHEMA_EVOLUTION",
+ errorClass = "UNSUPPORTED_TABLE_CHANGES_IN_AUTO_SCHEMA_EVOLUTION.CATALOG_PARTIAL",
messageParameters = Map(
- "changes" -> changesDesc, "tableName" -> toSQLId(sanitizedTableName)))
+ "tableName" -> toSQLId(tableName), "changes" -> changesDesc))
+ }
+
+ def failedToEvolveSchemaDuringMergeError(
+ ident: Identifier,
+ catalogName: String,
+ cause: Throwable): Throwable = {
+ val tableName = (catalogName +: ident.asMultipartIdentifier).map(_.replaceAll("\"", ""))
Review Comment:
Same point.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
Review Comment:
Shall we pass CatalogPlugin and switch order of catalog and ident to match other errors?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveMergeIntoSchemaEvolution.scala:
##########
@@ -40,46 +43,58 @@ object ResolveMergeIntoSchemaEvolution extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// This rule should run only if all assignments are resolved, except those
// that will be satisfied by schema evolution
- case m@MergeIntoTable(_, _, _, _, _, _, _) if m.evaluateSchemaEvolution =>
- val changes = m.changesForSchemaEvolution
- if (changes.isEmpty) {
- m
- } else {
- val finalAttrMapping = ArrayBuffer.empty[(Attribute, Attribute)]
- val newTarget = m.targetTable.transform {
- case r: DataSourceV2Relation =>
- val referencedSourceSchema = MergeIntoTable.sourceSchemaForSchemaEvolution(m)
- val newTarget = performSchemaEvolution(r, referencedSourceSchema, changes)
- val oldTargetOutput = m.targetTable.output
- val newTargetOutput = newTarget.output
- val attributeMapping = oldTargetOutput.zip(newTargetOutput)
- finalAttrMapping ++= attributeMapping
- newTarget
- }
- val res = m.copy(targetTable = newTarget)
- res.rewriteAttrs(AttributeMap(finalAttrMapping.toSeq))
+ case m: MergeIntoTable if m.pendingSchemaChanges.nonEmpty =>
+ EliminateSubqueryAliases(m.targetTable) match {
+ case ExtractV2CatalogAndIdentifier(catalog, ident) =>
+ val newTable = evolveSchema(m, catalog, ident)
+ val mergeWithNewTarget = replaceMergeTarget(m, newTable)
+
+ val remainingChanges = mergeWithNewTarget.pendingSchemaChanges
+ if (remainingChanges.nonEmpty) {
+ throw QueryCompilationErrors.unsupportedTableChangesInAutoSchemaEvolutionError(
+ ident, catalog.name(), remainingChanges.toArray)
+ }
+
+ mergeWithNewTarget
+ case _ =>
+ m
}
}
- private def performSchemaEvolution(
- relation: DataSourceV2Relation,
- referencedSourceSchema: StructType,
- changes: Array[TableChange]): DataSourceV2Relation = {
- (relation.catalog, relation.identifier) match {
- case (Some(c: TableCatalog), Some(i)) =>
- c.alterTable(i, changes: _*)
- val newTable = c.loadTable(i)
- val newSchema = CatalogV2Util.v2ColumnsToStructType(newTable.columns())
- // Check if there are any remaining changes not applied.
- val remainingChanges = MergeIntoTable.schemaChanges(newSchema, referencedSourceSchema)
- if (remainingChanges.nonEmpty) {
- throw QueryCompilationErrors.unsupportedTableChangesInAutoSchemaEvolutionError(
- remainingChanges, i.toQualifiedNameParts(c))
- }
- relation.copy(table = newTable, output = DataTypeUtils.toAttributes(newSchema))
- case _ => logWarning(s"Schema Evolution enabled but data source $relation " +
- s"does not support it, skipping.")
- relation
+ private def evolveSchema(
Review Comment:
In light of needed this logic for inserts too, what about making it a bit more generic?
```
private def evolveSchema(
catalog: TableCatalog,
ident: Identifier,
changes: Seq[TableChange]): Unit = {
try {
catalog.alterTable(ident, changes: _*)
} catch {
...
}
}
```
And then using it like below:
```
evolveSchema(catalog, ident, m.pendingSchemaChanges)
val writePrivileges = MergeIntoTable.getWritePrivileges(m)
val newTable = catalog.loadTable(ident, writePrivileges.toSet.asJava)
val mergeWithNewTarget = replaceMergeTarget(m, newTable)
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1112,6 +1102,27 @@ object MergeIntoTable {
assignmenKeyExpr == assignmentValueExpr &&
assignmenKeyExpr == sourceFieldPath
Review Comment:
Typo in `assignmenKeyExpr`?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveMergeIntoSchemaEvolution.scala:
##########
@@ -40,46 +43,58 @@ object ResolveMergeIntoSchemaEvolution extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// This rule should run only if all assignments are resolved, except those
// that will be satisfied by schema evolution
- case m@MergeIntoTable(_, _, _, _, _, _, _) if m.evaluateSchemaEvolution =>
- val changes = m.changesForSchemaEvolution
- if (changes.isEmpty) {
- m
- } else {
- val finalAttrMapping = ArrayBuffer.empty[(Attribute, Attribute)]
- val newTarget = m.targetTable.transform {
- case r: DataSourceV2Relation =>
- val referencedSourceSchema = MergeIntoTable.sourceSchemaForSchemaEvolution(m)
- val newTarget = performSchemaEvolution(r, referencedSourceSchema, changes)
- val oldTargetOutput = m.targetTable.output
- val newTargetOutput = newTarget.output
- val attributeMapping = oldTargetOutput.zip(newTargetOutput)
- finalAttrMapping ++= attributeMapping
- newTarget
- }
- val res = m.copy(targetTable = newTarget)
- res.rewriteAttrs(AttributeMap(finalAttrMapping.toSeq))
+ case m: MergeIntoTable if m.pendingSchemaChanges.nonEmpty =>
+ EliminateSubqueryAliases(m.targetTable) match {
+ case ExtractV2CatalogAndIdentifier(catalog, ident) =>
+ val newTable = evolveSchema(m, catalog, ident)
+ val mergeWithNewTarget = replaceMergeTarget(m, newTable)
+
+ val remainingChanges = mergeWithNewTarget.pendingSchemaChanges
+ if (remainingChanges.nonEmpty) {
+ throw QueryCompilationErrors.unsupportedTableChangesInAutoSchemaEvolutionError(
+ ident, catalog.name(), remainingChanges.toArray)
+ }
+
+ mergeWithNewTarget
+ case _ =>
+ m
}
}
- private def performSchemaEvolution(
- relation: DataSourceV2Relation,
- referencedSourceSchema: StructType,
- changes: Array[TableChange]): DataSourceV2Relation = {
- (relation.catalog, relation.identifier) match {
- case (Some(c: TableCatalog), Some(i)) =>
- c.alterTable(i, changes: _*)
- val newTable = c.loadTable(i)
- val newSchema = CatalogV2Util.v2ColumnsToStructType(newTable.columns())
- // Check if there are any remaining changes not applied.
- val remainingChanges = MergeIntoTable.schemaChanges(newSchema, referencedSourceSchema)
- if (remainingChanges.nonEmpty) {
- throw QueryCompilationErrors.unsupportedTableChangesInAutoSchemaEvolutionError(
- remainingChanges, i.toQualifiedNameParts(c))
- }
- relation.copy(table = newTable, output = DataTypeUtils.toAttributes(newSchema))
- case _ => logWarning(s"Schema Evolution enabled but data source $relation " +
- s"does not support it, skipping.")
- relation
+ private def evolveSchema(
+ merge: MergeIntoTable,
+ catalog: TableCatalog,
+ ident: Identifier): Table = {
+ try {
+ catalog.alterTable(ident, merge.pendingSchemaChanges: _*)
+ val writePrivileges = MergeIntoTable.getWritePrivileges(merge)
+ catalog.loadTable(ident, writePrivileges.toSet.asJava)
+ } catch {
+ case e: IllegalArgumentException if !e.isInstanceOf[SparkThrowable] =>
Review Comment:
Good point.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
+ catalogName: String,
+ changes: Array[TableChange]): Throwable = {
+ val tableName = (catalogName +: ident.asMultipartIdentifier).map(_.replaceAll("\"", ""))
Review Comment:
What about below?
```
def unsupportedAutoSchemaEvolutionChangesError(
catalog: CatalogPlugin,
ident: Identifier,
changes: Array[TableChange]): Throwable = {
new AnalysisException(
errorClass = "UNSUPPORTED_AUTO_SCHEMA_EVOLUTION_CHANGES.PARTIAL_EVOLUTION",
messageParameters = Map(
"tableName" -> toSQLId(ident.toQualifiedNameParts(catalog)),
"changes" -> changes.mkString("; ")))
}
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
+ catalogName: String,
+ changes: Array[TableChange]): Throwable = {
+ val tableName = (catalogName +: ident.asMultipartIdentifier).map(_.replaceAll("\"", ""))
Review Comment:
Notice `ident.toQualifiedNameParts(catalog)` and new name.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1112,6 +1102,27 @@ object MergeIntoTable {
assignmenKeyExpr == assignmentValueExpr &&
assignmenKeyExpr == sourceFieldPath
}
+
+ private def areSchemaEvolutionReady(
+ assignments: Seq[Assignment],
+ source: LogicalPlan): Boolean = {
+ assignments.forall(assign => assign.resolved || isSchemaEvolutionCandidate(assign, source))
+ }
+
+ private def isSchemaEvolutionCandidate(assignment: Assignment, source: LogicalPlan): Boolean = {
+ assignment.value.resolved && isSameColumnAssignment(assignment, source)
+ }
+
+ // Helper method to check if an assignment key is equal to a source column
+ // and if the assignment value is that same source column.
+ // Example: UPDATE SET target.a = source.a
+ private def isSameColumnAssignment(assignment: Assignment, source: LogicalPlan): Boolean = {
+ // key must be a non-qualified field path that may be added to target schema via evolution
+ val keyPath = extractFieldPath(assignment.key, allowUnresolved = true)
+ // value should always be resolved (from source)
+ val valuePath = extractFieldPath(assignment.value, allowUnresolved = false)
+ keyPath == valuePath && assignment.value.references.subsetOf(source.outputSet)
Review Comment:
Just to double check: `assignment.value.references.subsetOf(source.outputSet)` works for nested?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
+ catalogName: String,
+ changes: Array[TableChange]): Throwable = {
+ val tableName = (catalogName +: ident.asMultipartIdentifier).map(_.replaceAll("\"", ""))
Review Comment:
You can also go for `unsupportedChanges` or `remainingChanges` to be explicit.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -7660,8 +7660,20 @@
},
"UNSUPPORTED_TABLE_CHANGES_IN_AUTO_SCHEMA_EVOLUTION" : {
Review Comment:
Like we can go for
```
QueryCompilationErrors$partialAutoSchemaEvolutionError()
QueryCompilationErrors$failedAutoSchemaEvolutionError()
```
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -7660,8 +7660,20 @@
},
"UNSUPPORTED_TABLE_CHANGES_IN_AUTO_SCHEMA_EVOLUTION" : {
Review Comment:
This will affect error method names too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #54704:
URL: https://github.com/apache/spark/pull/54704#discussion_r2913475147
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala:
##########
@@ -45,8 +45,8 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand with PredicateHelper
override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
case m @ MergeIntoTable(aliasedTable, source, cond, matchedActions, notMatchedActions,
- notMatchedBySourceActions, _) if m.resolved && m.rewritable && m.aligned &&
- !m.needSchemaEvolution && matchedActions.isEmpty && notMatchedActions.size == 1 &&
Review Comment:
Yes, i also checked with AI here:
Yes. It’s safe even without assuming rule order.
Reasoning:
- aligned is defined as: every action’s assignments align with the current targetTable.output (same length, matching attribute names, compatible types).
- Pending schema evolution means the catalog target hasn’t been evolved yet, so in the plan targetTable.output is still the pre‑evolution schema.
- Assignments that refer to new columns or evolved types therefore cannot align with that current target (e.g. different length or incompatible types), so aligned is false whenever there are pending schema changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #54704:
URL: https://github.com/apache/spark/pull/54704#discussion_r2916760770
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala:
##########
@@ -45,8 +45,8 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand with PredicateHelper
override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
case m @ MergeIntoTable(aliasedTable, source, cond, matchedActions, notMatchedActions,
- notMatchedBySourceActions, _) if m.resolved && m.rewritable && m.aligned &&
- !m.needSchemaEvolution && matchedActions.isEmpty && notMatchedActions.size == 1 &&
Review Comment:
Thanks for confirming!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #54704:
URL: https://github.com/apache/spark/pull/54704#discussion_r2913859517
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
+ catalogName: String,
+ changes: Array[TableChange]): Throwable = {
+ val tableName = (catalogName +: ident.asMultipartIdentifier).map(_.replaceAll("\"", ""))
val changesDesc = changes.map(_.toString).mkString("; ")
Review Comment:
changed
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -7660,8 +7660,20 @@
},
"UNSUPPORTED_TABLE_CHANGES_IN_AUTO_SCHEMA_EVOLUTION" : {
Review Comment:
yep, makes sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi closed pull request #54704: [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges
URL: https://github.com/apache/spark/pull/54704
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #54704:
URL: https://github.com/apache/spark/pull/54704#discussion_r2916736442
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -1624,8 +1621,7 @@ class Analyzer(
}
InsertAction(
resolvedInsertCondition,
- resolveAssignments(assignments, m, MergeResolvePolicy.SOURCE,
- throws = throws))
+ resolveAssignments(assignments, m, MergeResolvePolicy.SOURCE, throws))
Review Comment:
There is one more like this below at line 1639.
```
// The update value can access columns from the target table only.
resolveAssignments(assignments, m, MergeResolvePolicy.TARGET,
throws = throws),
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #54704:
URL: https://github.com/apache/spark/pull/54704#discussion_r2913861154
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveMergeIntoSchemaEvolution.scala:
##########
@@ -40,46 +43,58 @@ object ResolveMergeIntoSchemaEvolution extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// This rule should run only if all assignments are resolved, except those
// that will be satisfied by schema evolution
- case m@MergeIntoTable(_, _, _, _, _, _, _) if m.evaluateSchemaEvolution =>
- val changes = m.changesForSchemaEvolution
- if (changes.isEmpty) {
- m
- } else {
- val finalAttrMapping = ArrayBuffer.empty[(Attribute, Attribute)]
- val newTarget = m.targetTable.transform {
- case r: DataSourceV2Relation =>
- val referencedSourceSchema = MergeIntoTable.sourceSchemaForSchemaEvolution(m)
- val newTarget = performSchemaEvolution(r, referencedSourceSchema, changes)
- val oldTargetOutput = m.targetTable.output
- val newTargetOutput = newTarget.output
- val attributeMapping = oldTargetOutput.zip(newTargetOutput)
- finalAttrMapping ++= attributeMapping
- newTarget
- }
- val res = m.copy(targetTable = newTarget)
- res.rewriteAttrs(AttributeMap(finalAttrMapping.toSeq))
+ case m: MergeIntoTable if m.pendingSchemaChanges.nonEmpty =>
+ EliminateSubqueryAliases(m.targetTable) match {
+ case ExtractV2CatalogAndIdentifier(catalog, ident) =>
+ val newTable = evolveSchema(m, catalog, ident)
+ val mergeWithNewTarget = replaceMergeTarget(m, newTable)
+
+ val remainingChanges = mergeWithNewTarget.pendingSchemaChanges
+ if (remainingChanges.nonEmpty) {
+ throw QueryCompilationErrors.unsupportedTableChangesInAutoSchemaEvolutionError(
+ ident, catalog.name(), remainingChanges.toArray)
+ }
+
+ mergeWithNewTarget
+ case _ =>
+ m
}
}
- private def performSchemaEvolution(
- relation: DataSourceV2Relation,
- referencedSourceSchema: StructType,
- changes: Array[TableChange]): DataSourceV2Relation = {
- (relation.catalog, relation.identifier) match {
- case (Some(c: TableCatalog), Some(i)) =>
- c.alterTable(i, changes: _*)
- val newTable = c.loadTable(i)
- val newSchema = CatalogV2Util.v2ColumnsToStructType(newTable.columns())
- // Check if there are any remaining changes not applied.
- val remainingChanges = MergeIntoTable.schemaChanges(newSchema, referencedSourceSchema)
- if (remainingChanges.nonEmpty) {
- throw QueryCompilationErrors.unsupportedTableChangesInAutoSchemaEvolutionError(
- remainingChanges, i.toQualifiedNameParts(c))
- }
- relation.copy(table = newTable, output = DataTypeUtils.toAttributes(newSchema))
- case _ => logWarning(s"Schema Evolution enabled but data source $relation " +
- s"does not support it, skipping.")
- relation
+ private def evolveSchema(
Review Comment:
done, thanks for suggestion
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1112,6 +1102,27 @@ object MergeIntoTable {
assignmenKeyExpr == assignmentValueExpr &&
assignmenKeyExpr == sourceFieldPath
Review Comment:
fixed, thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #54704:
URL: https://github.com/apache/spark/pull/54704#discussion_r2913857885
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
Review Comment:
done
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -3515,13 +3515,27 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
}
def unsupportedTableChangesInAutoSchemaEvolutionError(
- changes: Array[TableChange], tableName: Seq[String]): Throwable = {
- val sanitizedTableName = tableName.map(_.replaceAll("\"", ""))
+ ident: Identifier,
+ catalogName: String,
+ changes: Array[TableChange]): Throwable = {
+ val tableName = (catalogName +: ident.asMultipartIdentifier).map(_.replaceAll("\"", ""))
Review Comment:
thanks, done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges [spark]
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on PR #54704:
URL: https://github.com/apache/spark/pull/54704#issuecomment-4039252974
Thanks, @szehon-ho! Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARk-55903][SQL] Simplify Merge Into Schema Evolution and Check Write Privileges [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #54704:
URL: https://github.com/apache/spark/pull/54704#discussion_r2908532428
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveMergeIntoSchemaEvolution.scala:
##########
@@ -40,46 +43,58 @@ object ResolveMergeIntoSchemaEvolution extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// This rule should run only if all assignments are resolved, except those
// that will be satisfied by schema evolution
- case m@MergeIntoTable(_, _, _, _, _, _, _) if m.evaluateSchemaEvolution =>
- val changes = m.changesForSchemaEvolution
- if (changes.isEmpty) {
- m
- } else {
- val finalAttrMapping = ArrayBuffer.empty[(Attribute, Attribute)]
- val newTarget = m.targetTable.transform {
- case r: DataSourceV2Relation =>
- val referencedSourceSchema = MergeIntoTable.sourceSchemaForSchemaEvolution(m)
- val newTarget = performSchemaEvolution(r, referencedSourceSchema, changes)
- val oldTargetOutput = m.targetTable.output
- val newTargetOutput = newTarget.output
- val attributeMapping = oldTargetOutput.zip(newTargetOutput)
- finalAttrMapping ++= attributeMapping
- newTarget
- }
- val res = m.copy(targetTable = newTarget)
- res.rewriteAttrs(AttributeMap(finalAttrMapping.toSeq))
+ case m: MergeIntoTable if m.pendingSchemaChanges.nonEmpty =>
+ EliminateSubqueryAliases(m.targetTable) match {
+ case ExtractV2CatalogAndIdentifier(catalog, ident) =>
+ val newTable = evolveSchema(m, catalog, ident)
+ val mergeWithNewTarget = replaceMergeTarget(m, newTable)
+
+ val remainingChanges = mergeWithNewTarget.pendingSchemaChanges
+ if (remainingChanges.nonEmpty) {
+ throw QueryCompilationErrors.unsupportedTableChangesInAutoSchemaEvolutionError(
+ ident, catalog.name(), remainingChanges.toArray)
+ }
+
+ mergeWithNewTarget
+ case _ =>
+ m
}
}
- private def performSchemaEvolution(
- relation: DataSourceV2Relation,
- referencedSourceSchema: StructType,
- changes: Array[TableChange]): DataSourceV2Relation = {
- (relation.catalog, relation.identifier) match {
- case (Some(c: TableCatalog), Some(i)) =>
- c.alterTable(i, changes: _*)
- val newTable = c.loadTable(i)
- val newSchema = CatalogV2Util.v2ColumnsToStructType(newTable.columns())
- // Check if there are any remaining changes not applied.
- val remainingChanges = MergeIntoTable.schemaChanges(newSchema, referencedSourceSchema)
- if (remainingChanges.nonEmpty) {
- throw QueryCompilationErrors.unsupportedTableChangesInAutoSchemaEvolutionError(
- remainingChanges, i.toQualifiedNameParts(c))
- }
- relation.copy(table = newTable, output = DataTypeUtils.toAttributes(newSchema))
- case _ => logWarning(s"Schema Evolution enabled but data source $relation " +
- s"does not support it, skipping.")
- relation
+ private def evolveSchema(
+ merge: MergeIntoTable,
+ catalog: TableCatalog,
+ ident: Identifier): Table = {
+ try {
+ catalog.alterTable(ident, merge.pendingSchemaChanges: _*)
+ val writePrivileges = MergeIntoTable.getWritePrivileges(merge)
+ catalog.loadTable(ident, writePrivileges.toSet.asJava)
+ } catch {
+ case e: IllegalArgumentException if !e.isInstanceOf[SparkThrowable] =>
Review Comment:
this is to keep in line with AlterTableExec exceptions: https://github.com/apache/spark/blob/adeb35d9373641bc6bf3f6ef74cfe339953002fc/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AlterTableExec.scala#L43 but let me know if it is unnecessary
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #54704:
URL: https://github.com/apache/spark/pull/54704#discussion_r2913719809
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1112,6 +1102,27 @@ object MergeIntoTable {
assignmenKeyExpr == assignmentValueExpr &&
assignmenKeyExpr == sourceFieldPath
}
+
+ private def areSchemaEvolutionReady(
+ assignments: Seq[Assignment],
+ source: LogicalPlan): Boolean = {
+ assignments.forall(assign => assign.resolved || isSchemaEvolutionCandidate(assign, source))
+ }
+
+ private def isSchemaEvolutionCandidate(assignment: Assignment, source: LogicalPlan): Boolean = {
+ assignment.value.resolved && isSameColumnAssignment(assignment, source)
+ }
+
+ // Helper method to check if an assignment key is equal to a source column
+ // and if the assignment value is that same source column.
+ // Example: UPDATE SET target.a = source.a
+ private def isSameColumnAssignment(assignment: Assignment, source: LogicalPlan): Boolean = {
+ // key must be a non-qualified field path that may be added to target schema via evolution
+ val keyPath = extractFieldPath(assignment.key, allowUnresolved = true)
+ // value should always be resolved (from source)
+ val valuePath = extractFieldPath(assignment.value, allowUnresolved = false)
+ keyPath == valuePath && assignment.value.references.subsetOf(source.outputSet)
Review Comment:
yes.. also from AI here is how it works:
MERGE ... UPDATE SET addr.city = source.addr.city
- Key: nested path addr.city, e.g. GetStructField(GetStructField(target.addr, ...), ...)
- Value: same path on source, e.g. GetStructField(GetStructField(source.addr, 0), 1)
this may be possible to improve, we'll have to explore in the next pr
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-55903][SQL] Simplify MERGE Schema Evolution and Check Write Privileges [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #54704:
URL: https://github.com/apache/spark/pull/54704#discussion_r2913861154
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveMergeIntoSchemaEvolution.scala:
##########
@@ -40,46 +43,58 @@ object ResolveMergeIntoSchemaEvolution extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
// This rule should run only if all assignments are resolved, except those
// that will be satisfied by schema evolution
- case m@MergeIntoTable(_, _, _, _, _, _, _) if m.evaluateSchemaEvolution =>
- val changes = m.changesForSchemaEvolution
- if (changes.isEmpty) {
- m
- } else {
- val finalAttrMapping = ArrayBuffer.empty[(Attribute, Attribute)]
- val newTarget = m.targetTable.transform {
- case r: DataSourceV2Relation =>
- val referencedSourceSchema = MergeIntoTable.sourceSchemaForSchemaEvolution(m)
- val newTarget = performSchemaEvolution(r, referencedSourceSchema, changes)
- val oldTargetOutput = m.targetTable.output
- val newTargetOutput = newTarget.output
- val attributeMapping = oldTargetOutput.zip(newTargetOutput)
- finalAttrMapping ++= attributeMapping
- newTarget
- }
- val res = m.copy(targetTable = newTarget)
- res.rewriteAttrs(AttributeMap(finalAttrMapping.toSeq))
+ case m: MergeIntoTable if m.pendingSchemaChanges.nonEmpty =>
+ EliminateSubqueryAliases(m.targetTable) match {
+ case ExtractV2CatalogAndIdentifier(catalog, ident) =>
+ val newTable = evolveSchema(m, catalog, ident)
+ val mergeWithNewTarget = replaceMergeTarget(m, newTable)
+
+ val remainingChanges = mergeWithNewTarget.pendingSchemaChanges
+ if (remainingChanges.nonEmpty) {
+ throw QueryCompilationErrors.unsupportedTableChangesInAutoSchemaEvolutionError(
+ ident, catalog.name(), remainingChanges.toArray)
+ }
+
+ mergeWithNewTarget
+ case _ =>
+ m
}
}
- private def performSchemaEvolution(
- relation: DataSourceV2Relation,
- referencedSourceSchema: StructType,
- changes: Array[TableChange]): DataSourceV2Relation = {
- (relation.catalog, relation.identifier) match {
- case (Some(c: TableCatalog), Some(i)) =>
- c.alterTable(i, changes: _*)
- val newTable = c.loadTable(i)
- val newSchema = CatalogV2Util.v2ColumnsToStructType(newTable.columns())
- // Check if there are any remaining changes not applied.
- val remainingChanges = MergeIntoTable.schemaChanges(newSchema, referencedSourceSchema)
- if (remainingChanges.nonEmpty) {
- throw QueryCompilationErrors.unsupportedTableChangesInAutoSchemaEvolutionError(
- remainingChanges, i.toQualifiedNameParts(c))
- }
- relation.copy(table = newTable, output = DataTypeUtils.toAttributes(newSchema))
- case _ => logWarning(s"Schema Evolution enabled but data source $relation " +
- s"does not support it, skipping.")
- relation
+ private def evolveSchema(
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org