You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2021/12/17 17:39:17 UTC
[GitHub] [iceberg] aokolnychyi opened a new pull request #3764: Spark: Implement copy-on-write UPDATE
aokolnychyi opened a new pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764
This PR implements copy-on-write UPDATE in Spark.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r771996866
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentAlignmentSupport.scala
##########
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.AnsiCast
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.catalyst.expressions.Cast
+import org.apache.spark.sql.catalyst.expressions.CreateNamedStruct
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.ExtractValue
+import org.apache.spark.sql.catalyst.expressions.GetStructField
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.NamedExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
+import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types.StructField
+import org.apache.spark.sql.types.StructType
+import scala.collection.mutable
+
+trait AssignmentAlignmentSupport extends CastSupport {
+
+ self: SQLConfHelper =>
+
+ private case class ColumnUpdate(ref: Seq[String], expr: Expression)
+
+ /**
+ * Checks whether assignments are aligned and match table columns.
+ *
+ * @param table a target table
+ * @param assignments assignments to check
+ * @return true if the assignments are aligned
+ */
+ protected def aligned(table: LogicalPlan, assignments: Seq[Assignment]): Boolean = {
+ val sameSize = table.output.size == assignments.size
+ sameSize && table.output.zip(assignments).forall { case (attr, assignment) =>
+ val key = assignment.key
+ val value = assignment.value
+ asAssignmentRef(attr) == asAssignmentRef(key) && attr.dataType == value.dataType
+ }
+ }
+
+ /**
+ * Aligns assignments to match table columns.
+ * <p>
+ * This method processes and reorders given assignments so that each target column gets
+ * an expression it should be set to. If a column does not have a matching assignment,
+ * it will be set to its current value. For example, if one passes a table with columns c1, c2
+ * and an assignment c2 = 1, this method will return c1 = c1, c2 = 1.
+ * <p>
+ * This method also handles updates to nested columns. If there is an assignment to a particular
+ * nested field, this method will construct a new struct with one field updated
+ * preserving other fields that have not been modified. For example, if one passes a table with
+ * columns c1, c2 where c2 is a struct with fields n1 and n2 and an assignment c2.n2 = 1,
+ * this method will return c1 = c1, c2 = struct(c2.n1, 1).
+ *
+ * @param table a target table
+ * @param assignments assignments to align
+ * @return aligned assignments that match table columns
+ */
+ protected def alignAssignments(
+ table: LogicalPlan,
+ assignments: Seq[Assignment]): Seq[Assignment] = {
+
+ val columnUpdates = assignments.map(a => ColumnUpdate(asAssignmentRef(a.key), a.value))
+ val outputExprs = applyUpdates(table.output, columnUpdates)
+ outputExprs.zip(table.output).map {
+ case (expr, attr) => toAssignment(attr, expr)
+ }
+ }
+
+ protected def toAssignment(key: Expression, value: Expression): Assignment = {
Review comment:
Some comments to explain what is going on here would be really helpful. Maybe also rename this to something that makes it more clear, like `handleCharTypeLimit`.
If you end up renaming this, also consider accepting an Assignment and optionally transforming it, like this:
```scala
protected def handleCharLimits(assignment: Assignment): Assignment = {
val rawKeyType = ...
if (CharVarcharUtils.hasCharVarchar(rawKeyType)) {
// build the transformed assignment
} else {
assignment
}
}
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772553485
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelCommandDynamicPruning.scala
##########
@@ -86,6 +89,16 @@ case class RowLevelCommandDynamicPruning(spark: SparkSession) extends Rule[Logic
val matchingRowsPlan = command match {
case d: DeleteFromIcebergTable =>
Filter(d.condition.get, relation)
+ case u: UpdateIcebergTable =>
+ // UPDATEs with subqueries may be rewritten using a UNION with two identical scan relations
+ // each scan relation will get its own dynamic filter that will be shared during execution
+ // the analyzer will assign different expr IDs for each scan relation output attributes
+ // that's why the condition may refer to invalid attr expr IDs and must be transformed
Review comment:
I've tried to improve the comment to make it clear and also made a note about the rewrite rule. Let me know if its any better.
> For example, UPDATE t SET id = 1 WHERE true doesn't need to project id from the table. If id is not projected, then the attrMap is incorrect.
It was working correctly as we don't support schema pruning in copy-on-write operations. The only case where it is applicable if we have `true` update condition and then we can discard only columns assigned to literals. Seemed like a very narrow use case. I think we can handle it later as we did not support it in the old implementation. If the condition is not trivial, the assigned expression will be `if ... else oldValue` so we won't be able to prune the column.
That being said, I did modify the logic for building the attribute map. It should work even if we add schema pruning.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r771996866
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentAlignmentSupport.scala
##########
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.AnsiCast
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.catalyst.expressions.Cast
+import org.apache.spark.sql.catalyst.expressions.CreateNamedStruct
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.ExtractValue
+import org.apache.spark.sql.catalyst.expressions.GetStructField
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.NamedExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
+import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types.StructField
+import org.apache.spark.sql.types.StructType
+import scala.collection.mutable
+
+trait AssignmentAlignmentSupport extends CastSupport {
+
+ self: SQLConfHelper =>
+
+ private case class ColumnUpdate(ref: Seq[String], expr: Expression)
+
+ /**
+ * Checks whether assignments are aligned and match table columns.
+ *
+ * @param table a target table
+ * @param assignments assignments to check
+ * @return true if the assignments are aligned
+ */
+ protected def aligned(table: LogicalPlan, assignments: Seq[Assignment]): Boolean = {
+ val sameSize = table.output.size == assignments.size
+ sameSize && table.output.zip(assignments).forall { case (attr, assignment) =>
+ val key = assignment.key
+ val value = assignment.value
+ asAssignmentRef(attr) == asAssignmentRef(key) && attr.dataType == value.dataType
+ }
+ }
+
+ /**
+ * Aligns assignments to match table columns.
+ * <p>
+ * This method processes and reorders given assignments so that each target column gets
+ * an expression it should be set to. If a column does not have a matching assignment,
+ * it will be set to its current value. For example, if one passes a table with columns c1, c2
+ * and an assignment c2 = 1, this method will return c1 = c1, c2 = 1.
+ * <p>
+ * This method also handles updates to nested columns. If there is an assignment to a particular
+ * nested field, this method will construct a new struct with one field updated
+ * preserving other fields that have not been modified. For example, if one passes a table with
+ * columns c1, c2 where c2 is a struct with fields n1 and n2 and an assignment c2.n2 = 1,
+ * this method will return c1 = c1, c2 = struct(c2.n1, 1).
+ *
+ * @param table a target table
+ * @param assignments assignments to align
+ * @return aligned assignments that match table columns
+ */
+ protected def alignAssignments(
+ table: LogicalPlan,
+ assignments: Seq[Assignment]): Seq[Assignment] = {
+
+ val columnUpdates = assignments.map(a => ColumnUpdate(asAssignmentRef(a.key), a.value))
+ val outputExprs = applyUpdates(table.output, columnUpdates)
+ outputExprs.zip(table.output).map {
+ case (expr, attr) => toAssignment(attr, expr)
+ }
+ }
+
+ protected def toAssignment(key: Expression, value: Expression): Assignment = {
Review comment:
Some comments to explain what is going on here would be really helpful. Maybe also rename this to something that makes it more clear, like `handleCharTypeLimit`.
If you end up renaming this, also consider accepting an Assignment and optionally transforming it, like this:
```scala
protected def handleCharLimits(assignment: Assignment): Assignment = {
val rawKeyType = ...
if (CharVarcharUtils.hasCharVarchar(rawKeyType)) {
// build the transformed assignment
} else {
assignment
}
}
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772556960
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelCommandDynamicPruning.scala
##########
@@ -109,6 +123,18 @@ case class RowLevelCommandDynamicPruning(spark: SparkSession) extends Rule[Logic
dynamicPruningSubqueries.reduce(And)
}
+ private def buildAttrMap(
+ tableAttrs: Seq[Attribute],
+ scanAttrs: Seq[Attribute]): AttributeMap[Attribute] = {
+
+ val attrMapping = tableAttrs.flatMap { tableAttr =>
+ scanAttrs
+ .find(scanAttr => scanAttr.name == tableAttr.name)
Review comment:
Should this use the resolver or are we confident that the case will always match?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772000648
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
Review comment:
Where is `u.resolved` implemented? Does that check alignment since `buildUpdateProjection` requires aligned assignments?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772557440
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan with updated rows
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val allRowsPlan = buildUpdateProjection(readRelation, assignments, updateCond)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition contains a subquery
+ private def buildReplaceDataWithUnionPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ // the same read relation will be used to read records that must be updated and be copied over
+ // DeduplicateRelations will take care of duplicated attr IDs
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan for records that match the cond and should be updated
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
Review comment:
I did move this to another place. Let me know if that's what you meant, @rdblue.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772557747
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan with updated rows
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val allRowsPlan = buildUpdateProjection(readRelation, assignments, updateCond)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition contains a subquery
+ private def buildReplaceDataWithUnionPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ // the same read relation will be used to read records that must be updated and be copied over
+ // DeduplicateRelations will take care of duplicated attr IDs
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan for records that match the cond and should be updated
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val matchedRowsPlan = Filter(updateCond, readRelation)
+ val updatedRowsPlan = buildUpdateProjection(matchedRowsPlan, assignments)
+
+ // build a plan for records that did not match the cond but had to be copied over
+ val remainingRowFilter = Not(EqualNullSafe(updateCond, Literal.TrueLiteral))
+ val remainingRowsPlan = Filter(remainingRowFilter, readRelation)
+
+ // new state is a union of updated and copied over records
+ val allRowsPlan = Union(updatedRowsPlan, remainingRowsPlan)
Review comment:
Renamed it to `updatedAndRemainingRowsPlan`. Still not entirely happy.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#issuecomment-998141914
@rdblue, it is ready for another round. I'll create a table of proposed distributions and orderings and will think about the default one. That's still an open item.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772548945
##########
File path: spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestUpdate.java
##########
@@ -603,11 +603,14 @@ public void testUpdateWithSelfSubquery() {
ImmutableList.of(row(1, "hr"), row(2, "x")),
sql("SELECT * FROM %s ORDER BY id", tableName));
- sql("UPDATE %s SET dep = 'y' WHERE " +
- "id = (SELECT count(*) FROM (SELECT DISTINCT id FROM %s) AS t)", tableName, tableName);
- assertEquals("Should have expected rows",
- ImmutableList.of(row(1, "hr"), row(2, "y")),
- sql("SELECT * FROM %s ORDER BY id", tableName));
+ // TODO: Spark does not support AQE and DPP with aggregates at the moment
Review comment:
Unfortunately, yes. It contains a distinct aggregation.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772576410
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan with updated rows
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val allRowsPlan = buildUpdateProjection(readRelation, assignments, updateCond)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition contains a subquery
+ private def buildReplaceDataWithUnionPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ // the same read relation will be used to read records that must be updated and be copied over
+ // DeduplicateRelations will take care of duplicated attr IDs
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan for records that match the cond and should be updated
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val matchedRowsPlan = Filter(updateCond, readRelation)
+ val updatedRowsPlan = buildUpdateProjection(matchedRowsPlan, assignments)
+
+ // build a plan for records that did not match the cond but had to be copied over
+ val remainingRowFilter = Not(EqualNullSafe(updateCond, Literal.TrueLiteral))
+ val remainingRowsPlan = Filter(remainingRowFilter, readRelation)
+
+ // new state is a union of updated and copied over records
+ val allRowsPlan = Union(updatedRowsPlan, remainingRowsPlan)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // this method assumes the assignments have been already aligned before
+ // the condition passed to this method may be different from the UPDATE condition
+ private def buildUpdateProjection(
+ plan: LogicalPlan,
+ assignments: Seq[Assignment],
+ cond: Expression = Literal.TrueLiteral): LogicalPlan = {
+
+ // TODO: avoid executing the condition for each column
Review comment:
If this is what we'd end up with anyway, then let's hold off. I see that this is what we do in 3.1 anyway so no need to change it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772001711
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan with updated rows
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val allRowsPlan = buildUpdateProjection(readRelation, assignments, updateCond)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition contains a subquery
+ private def buildReplaceDataWithUnionPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ // the same read relation will be used to read records that must be updated and be copied over
+ // DeduplicateRelations will take care of duplicated attr IDs
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan for records that match the cond and should be updated
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
Review comment:
If the condition is empty or `true`, then this doesn't need the remaining rows plan or the union. Should we handle that case explicitly?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r771997288
##########
File path: spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestUpdate.java
##########
@@ -603,11 +603,14 @@ public void testUpdateWithSelfSubquery() {
ImmutableList.of(row(1, "hr"), row(2, "x")),
sql("SELECT * FROM %s ORDER BY id", tableName));
- sql("UPDATE %s SET dep = 'y' WHERE " +
- "id = (SELECT count(*) FROM (SELECT DISTINCT id FROM %s) AS t)", tableName, tableName);
- assertEquals("Should have expected rows",
- ImmutableList.of(row(1, "hr"), row(2, "y")),
- sql("SELECT * FROM %s ORDER BY id", tableName));
+ // TODO: Spark does not support AQE and DPP with aggregates at the moment
Review comment:
This fails if AQE is enabled?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r771997525
##########
File path: spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestUpdate.java
##########
@@ -662,17 +665,6 @@ public void testUpdateWithNotInSubquery() {
sql("SELECT * FROM %s ORDER BY id ASC NULLS LAST, dep", tableName));
}
- @Test
- public void testUpdateWithNotInSubqueryNotSupported() {
Review comment:
Looks like this is no longer needed because `testUpdateWithNotInSubquery` is not ignored anymore.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772549165
##########
File path: spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestUpdate.java
##########
@@ -662,17 +665,6 @@ public void testUpdateWithNotInSubquery() {
sql("SELECT * FROM %s ORDER BY id ASC NULLS LAST, dep", tableName));
}
- @Test
- public void testUpdateWithNotInSubqueryNotSupported() {
Review comment:
Exactly. It is a valid test now.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#issuecomment-998376239
Thanks, @aokolnychyi! I've merged this to unblock MERGE INTO and we can follow up on the requested sort orders.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772571252
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
Review comment:
It's minor, but these lines of code are duplicated in a few places. You could have a method that returns the new relation with the original output, metadata attrs, and optional id attrs to simplify these. Not a big deal.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772549742
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
Review comment:
Nevermind, I saw that Iceberg just doesn't return a `SupportsDelta` class.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772555323
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
Review comment:
We are checking the constructed `RowLevelOperation` we got from `RowLevelOperationBuilder`. At this point, we passed the command and options to the data source and it reported what strategy should be used. In the future, we will just add another method call here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772555864
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
Review comment:
It previously relied on the order of rules but I made that explicit. That's a good call, I agree.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#issuecomment-998154259
Oops, looks like I broke something. Let me see.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772557047
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
Review comment:
Not sure I got. Could you elaborate, @rdblue?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772557855
##########
File path: spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
##########
@@ -199,6 +203,20 @@ public DistributionMode copyOnWriteDeleteDistributionMode() {
}
}
+ public DistributionMode copyOnWriteUpdateDistributionMode() {
+ String updateModeName = confParser.stringConf()
+ .option(SparkWriteOptions.DISTRIBUTION_MODE)
+ .tableProperty(TableProperties.UPDATE_DISTRIBUTION_MODE)
+ .parseOptional();
+
+ if (updateModeName != null) {
+ DistributionMode updateMode = DistributionMode.fromName(updateModeName);
+ return adjustWriteDistributionMode(updateMode);
Review comment:
Let me think.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772002405
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelCommandDynamicPruning.scala
##########
@@ -86,6 +89,16 @@ case class RowLevelCommandDynamicPruning(spark: SparkSession) extends Rule[Logic
val matchingRowsPlan = command match {
case d: DeleteFromIcebergTable =>
Filter(d.condition.get, relation)
+ case u: UpdateIcebergTable =>
+ // UPDATEs with subqueries may be rewritten using a UNION with two identical scan relations
+ // each scan relation will get its own dynamic filter that will be shared during execution
+ // the analyzer will assign different expr IDs for each scan relation output attributes
+ // that's why the condition may refer to invalid attr expr IDs and must be transformed
Review comment:
Okay, after reading through all the cases for rewriting update, this makes a lot more sense. I think it would have helped me reading through this to either direct to the union case in `RewriteUpdateTable` or to explain the rewrite here and give a little more context on when there will be a union.
As for my other comments, I think it would still be good to rename `attrMap` and I'm not sure whether there could be a case where `u.table.output` doesn't match `relation.output`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772629132
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
Review comment:
Sounds good. Added a method.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772000064
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
Review comment:
In the future, will we also check whether the delta plan is selected? Iceberg can support delta, but choose not to use it because of table settings.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r771592118
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan with updated rows
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val allRowsPlan = buildUpdateProjection(readRelation, assignments, updateCond)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition contains a subquery
+ private def buildReplaceDataWithUnionPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ // the same read relation will be used to read records that must be updated and be copied over
+ // DeduplicateRelations will take care of duplicated attr IDs
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
Review comment:
This part requires a bit of time to understand. The comment above is also important.
In short, we are using the same relation on both sides of the union. Then `DeduplicateRelations` clones one of them and assigns fresh expression IDs so that attributes on both sides are different. During job planning, we plan the job only once and then replace both `DataSourceV2Relation` with two `DataSourceV2ScanRelation`. The scan relations are different objects but they share the SAME `Scan` instance. During dynamic filtering, we assign each scan relation a dynamic filter. The dynamic filter subquery will be executed only once and then shared by two dynamic filters at runtime. Luckily, Spark reuses subqueries with and without AQE. The reason why we have two dynamic filters is because we don't know which side of the union plan will be executed first.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue merged pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue merged pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772001356
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan with updated rows
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val allRowsPlan = buildUpdateProjection(readRelation, assignments, updateCond)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition contains a subquery
+ private def buildReplaceDataWithUnionPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ // the same read relation will be used to read records that must be updated and be copied over
+ // DeduplicateRelations will take care of duplicated attr IDs
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan for records that match the cond and should be updated
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val matchedRowsPlan = Filter(updateCond, readRelation)
+ val updatedRowsPlan = buildUpdateProjection(matchedRowsPlan, assignments)
+
+ // build a plan for records that did not match the cond but had to be copied over
+ val remainingRowFilter = Not(EqualNullSafe(updateCond, Literal.TrueLiteral))
+ val remainingRowsPlan = Filter(remainingRowFilter, readRelation)
+
+ // new state is a union of updated and copied over records
+ val allRowsPlan = Union(updatedRowsPlan, remainingRowsPlan)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // this method assumes the assignments have been already aligned before
+ // the condition passed to this method may be different from the UPDATE condition
+ private def buildUpdateProjection(
+ plan: LogicalPlan,
+ assignments: Seq[Assignment],
+ cond: Expression = Literal.TrueLiteral): LogicalPlan = {
+
+ // TODO: avoid executing the condition for each column
Review comment:
It wouldn't be too hard to add a projection with one more column and then use the column value to determine whether to replace values.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772001898
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan with updated rows
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val allRowsPlan = buildUpdateProjection(readRelation, assignments, updateCond)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition contains a subquery
+ private def buildReplaceDataWithUnionPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ // the same read relation will be used to read records that must be updated and be copied over
+ // DeduplicateRelations will take care of duplicated attr IDs
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan for records that match the cond and should be updated
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val matchedRowsPlan = Filter(updateCond, readRelation)
+ val updatedRowsPlan = buildUpdateProjection(matchedRowsPlan, assignments)
+
+ // build a plan for records that did not match the cond but had to be copied over
+ val remainingRowFilter = Not(EqualNullSafe(updateCond, Literal.TrueLiteral))
+ val remainingRowsPlan = Filter(remainingRowFilter, readRelation)
+
+ // new state is a union of updated and copied over records
+ val allRowsPlan = Union(updatedRowsPlan, remainingRowsPlan)
Review comment:
The name `allRowsPlan` is a little misleading since it isn't all rows in the table. That said, I'm not sure what I would rename it to.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772001566
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
Review comment:
Nit: this could be constructed in a private method or passed in instead of `relation`. I think that would help readability.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772548715
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentAlignmentSupport.scala
##########
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.AnsiCast
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.catalyst.expressions.Cast
+import org.apache.spark.sql.catalyst.expressions.CreateNamedStruct
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.ExtractValue
+import org.apache.spark.sql.catalyst.expressions.GetStructField
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.NamedExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
+import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types.StructField
+import org.apache.spark.sql.types.StructType
+import scala.collection.mutable
+
+trait AssignmentAlignmentSupport extends CastSupport {
+
+ self: SQLConfHelper =>
+
+ private case class ColumnUpdate(ref: Seq[String], expr: Expression)
+
+ /**
+ * Checks whether assignments are aligned and match table columns.
+ *
+ * @param table a target table
+ * @param assignments assignments to check
+ * @return true if the assignments are aligned
+ */
+ protected def aligned(table: LogicalPlan, assignments: Seq[Assignment]): Boolean = {
+ val sameSize = table.output.size == assignments.size
+ sameSize && table.output.zip(assignments).forall { case (attr, assignment) =>
+ val key = assignment.key
+ val value = assignment.value
+ asAssignmentRef(attr) == asAssignmentRef(key) && attr.dataType == value.dataType
+ }
+ }
+
+ /**
+ * Aligns assignments to match table columns.
+ * <p>
+ * This method processes and reorders given assignments so that each target column gets
+ * an expression it should be set to. If a column does not have a matching assignment,
+ * it will be set to its current value. For example, if one passes a table with columns c1, c2
+ * and an assignment c2 = 1, this method will return c1 = c1, c2 = 1.
+ * <p>
+ * This method also handles updates to nested columns. If there is an assignment to a particular
+ * nested field, this method will construct a new struct with one field updated
+ * preserving other fields that have not been modified. For example, if one passes a table with
+ * columns c1, c2 where c2 is a struct with fields n1 and n2 and an assignment c2.n2 = 1,
+ * this method will return c1 = c1, c2 = struct(c2.n1, 1).
+ *
+ * @param table a target table
+ * @param assignments assignments to align
+ * @return aligned assignments that match table columns
+ */
+ protected def alignAssignments(
+ table: LogicalPlan,
+ assignments: Seq[Assignment]): Seq[Assignment] = {
+
+ val columnUpdates = assignments.map(a => ColumnUpdate(asAssignmentRef(a.key), a.value))
+ val outputExprs = applyUpdates(table.output, columnUpdates)
+ outputExprs.zip(table.output).map {
+ case (expr, attr) => toAssignment(attr, expr)
+ }
+ }
+
+ protected def toAssignment(key: Expression, value: Expression): Assignment = {
Review comment:
I added `AssignmentUtils` with a separate method dedicated to this functionality.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772629019
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelCommandDynamicPruning.scala
##########
@@ -109,6 +123,18 @@ case class RowLevelCommandDynamicPruning(spark: SparkSession) extends Rule[Logic
dynamicPruningSubqueries.reduce(And)
}
+ private def buildAttrMap(
+ tableAttrs: Seq[Attribute],
+ scanAttrs: Seq[Attribute]): AttributeMap[Attribute] = {
+
+ val attrMapping = tableAttrs.flatMap { tableAttr =>
+ scanAttrs
+ .find(scanAttr => scanAttr.name == tableAttr.name)
Review comment:
Switched to using a resolver.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772000648
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
Review comment:
Where is `u.resolved` implemented? Does that check alignment since `buildUpdateProjection` requires aligned assignments? I think we should make sure that the alignment has been 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772003153
##########
File path: spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
##########
@@ -199,6 +203,20 @@ public DistributionMode copyOnWriteDeleteDistributionMode() {
}
}
+ public DistributionMode copyOnWriteUpdateDistributionMode() {
+ String updateModeName = confParser.stringConf()
+ .option(SparkWriteOptions.DISTRIBUTION_MODE)
+ .tableProperty(TableProperties.UPDATE_DISTRIBUTION_MODE)
+ .parseOptional();
+
+ if (updateModeName != null) {
+ DistributionMode updateMode = DistributionMode.fromName(updateModeName);
+ return adjustWriteDistributionMode(updateMode);
Review comment:
As with the copy-on-write delete case, I think I'd prefer to preserve the existing file boundaries and row order if the table is unpartitioned and unsorted. I would also do that as the default, if neither the write order or the update order is specified for unpartitioned tables.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r771997001
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentAlignmentSupport.scala
##########
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.AnsiCast
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.catalyst.expressions.Cast
+import org.apache.spark.sql.catalyst.expressions.CreateNamedStruct
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.ExtractValue
+import org.apache.spark.sql.catalyst.expressions.GetStructField
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.NamedExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
+import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types.StructField
+import org.apache.spark.sql.types.StructType
+import scala.collection.mutable
+
+trait AssignmentAlignmentSupport extends CastSupport {
+
+ self: SQLConfHelper =>
+
+ private case class ColumnUpdate(ref: Seq[String], expr: Expression)
+
+ /**
+ * Checks whether assignments are aligned and match table columns.
+ *
+ * @param table a target table
+ * @param assignments assignments to check
+ * @return true if the assignments are aligned
+ */
+ protected def aligned(table: LogicalPlan, assignments: Seq[Assignment]): Boolean = {
+ val sameSize = table.output.size == assignments.size
+ sameSize && table.output.zip(assignments).forall { case (attr, assignment) =>
+ val key = assignment.key
+ val value = assignment.value
+ asAssignmentRef(attr) == asAssignmentRef(key) && attr.dataType == value.dataType
+ }
+ }
+
+ /**
+ * Aligns assignments to match table columns.
+ * <p>
+ * This method processes and reorders given assignments so that each target column gets
+ * an expression it should be set to. If a column does not have a matching assignment,
+ * it will be set to its current value. For example, if one passes a table with columns c1, c2
+ * and an assignment c2 = 1, this method will return c1 = c1, c2 = 1.
+ * <p>
+ * This method also handles updates to nested columns. If there is an assignment to a particular
+ * nested field, this method will construct a new struct with one field updated
+ * preserving other fields that have not been modified. For example, if one passes a table with
+ * columns c1, c2 where c2 is a struct with fields n1 and n2 and an assignment c2.n2 = 1,
+ * this method will return c1 = c1, c2 = struct(c2.n1, 1).
+ *
+ * @param table a target table
+ * @param assignments assignments to align
+ * @return aligned assignments that match table columns
+ */
+ protected def alignAssignments(
+ table: LogicalPlan,
+ assignments: Seq[Assignment]): Seq[Assignment] = {
+
+ val columnUpdates = assignments.map(a => ColumnUpdate(asAssignmentRef(a.key), a.value))
+ val outputExprs = applyUpdates(table.output, columnUpdates)
+ outputExprs.zip(table.output).map {
+ case (expr, attr) => toAssignment(attr, expr)
+ }
+ }
+
+ protected def toAssignment(key: Expression, value: Expression): Assignment = {
+ val rawKeyType = key.transform {
+ case attr: AttributeReference =>
+ CharVarcharUtils.getRawType(attr.metadata)
+ .map(attr.withDataType)
+ .getOrElse(attr)
+ }.dataType
+
+ val transformedValue = if (CharVarcharUtils.hasCharVarchar(rawKeyType)) {
+ CharVarcharUtils.stringLengthCheck(value, rawKeyType)
+ } else {
+ value
+ }
+
+ val transformedKey = key.transform {
+ case attr: AttributeReference => CharVarcharUtils.cleanAttrMetadata(attr)
+ }
+
+ Assignment(transformedKey, transformedValue)
+ }
+
+ private def applyUpdates(
+ cols: Seq[NamedExpression],
+ updates: Seq[ColumnUpdate],
+ resolver: Resolver = conf.resolver,
+ namePrefix: Seq[String] = Nil): Seq[Expression] = {
+
+ // iterate through columns at the current level and find which column updates match
+ cols.map { col =>
+ // find matches for this column or any of its children
+ val prefixMatchedUpdates = updates.filter(a => resolver(a.ref.head, col.name))
+ prefixMatchedUpdates match {
+ // if there is no exact match and no match for children, return the column as is
+ case updates if updates.isEmpty =>
+ col
+
+ // if there is an exact match, return the assigned expression
+ case Seq(update) if isExactMatch(update, col, resolver) =>
+ castIfNeeded(col, update.expr, resolver)
+
+ // if there are matches only for children
+ case updates if !hasExactMatch(updates, col, resolver) =>
+ col.dataType match {
+ case StructType(fields) =>
+ // build field expressions
+ val fieldExprs = fields.zipWithIndex.map { case (field, ordinal) =>
+ Alias(GetStructField(col, ordinal, Some(field.name)), field.name)()
+ }
+
+ // recursively apply this method on nested fields
+ val newUpdates = updates.map(u => u.copy(ref = u.ref.tail))
+ val updatedFieldExprs = applyUpdates(
+ fieldExprs,
+ newUpdates,
+ resolver,
+ namePrefix :+ col.name)
+
+ // construct a new struct with updated field expressions
+ toNamedStruct(fields, updatedFieldExprs)
+
+ case otherType =>
+ val colName = (namePrefix :+ col.name).mkString(".")
+ throw new AnalysisException(
+ "Updating nested fields is only supported for StructType " +
+ s"but $colName is of type $otherType"
+ )
+ }
+
+ // if there are conflicting updates, throw an exception
+ // there are two illegal scenarios:
+ // - multiple updates to the same column
+ // - updates to a top-level struct and its nested fields (e.g., a.b and a.b.c)
+ case updates if hasExactMatch(updates, col, resolver) =>
+ val conflictingCols = updates.map(u => (namePrefix ++ u.ref).mkString("."))
+ throw new AnalysisException(
+ "Updates are in conflict for these columns: " +
+ conflictingCols.distinct.mkString(", "))
+ }
+ }
+ }
+
+ private def toNamedStruct(fields: Seq[StructField], fieldExprs: Seq[Expression]): Expression = {
+ val namedStructExprs = fields.zip(fieldExprs).flatMap { case (field, expr) =>
+ Seq(Literal(field.name), expr)
+ }
+ CreateNamedStruct(namedStructExprs)
+ }
+
+ private def hasExactMatch(
+ updates: Seq[ColumnUpdate],
+ col: NamedExpression,
+ resolver: Resolver): Boolean = {
+
+ updates.exists(assignment => isExactMatch(assignment, col, resolver))
+ }
+
+ private def isExactMatch(
+ update: ColumnUpdate,
+ col: NamedExpression,
+ resolver: Resolver): Boolean = {
+
+ update.ref match {
+ case Seq(namePart) if resolver(namePart, col.name) => true
+ case _ => false
+ }
+ }
+
+ protected def castIfNeeded(
+ tableAttr: NamedExpression,
+ expr: Expression,
+ resolver: Resolver): Expression = {
+
+ val storeAssignmentPolicy = conf.storeAssignmentPolicy
+
+ // run the type check and catch type errors
+ storeAssignmentPolicy match {
+ case StoreAssignmentPolicy.STRICT | StoreAssignmentPolicy.ANSI =>
+ if (expr.nullable && !tableAttr.nullable) {
+ throw new AnalysisException(
+ s"Cannot write nullable values to non-null column '${tableAttr.name}'")
+ }
+
+ // use byName = true to catch cases when struct field names don't match
+ // e.g. a struct with fields (a, b) is assigned as a struct with fields (a, c) or (b, a)
+ val errors = new mutable.ArrayBuffer[String]()
+ val canWrite = DataType.canWrite(
+ expr.dataType, tableAttr.dataType, byName = true, resolver, tableAttr.name,
+ storeAssignmentPolicy, err => errors += err)
+
+ if (!canWrite) {
+ throw new AnalysisException(
+ s"Cannot write incompatible data:\n- ${errors.mkString("\n- ")}")
+ }
+
+ case _ => // OK
+ }
+
+ storeAssignmentPolicy match {
+ case _ if tableAttr.dataType.sameType(expr.dataType) =>
+ expr
+ case StoreAssignmentPolicy.ANSI =>
+ AnsiCast(expr, tableAttr.dataType, Option(conf.sessionLocalTimeZone))
+ case _ =>
+ Cast(expr, tableAttr.dataType, Option(conf.sessionLocalTimeZone))
+ }
+ }
+
+ protected def asAssignmentRef(expr: Expression): Seq[String] = expr match {
Review comment:
+1 for removing `implicit`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r771999598
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelCommandDynamicPruning.scala
##########
@@ -86,6 +89,16 @@ case class RowLevelCommandDynamicPruning(spark: SparkSession) extends Rule[Logic
val matchingRowsPlan = command match {
case d: DeleteFromIcebergTable =>
Filter(d.condition.get, relation)
+ case u: UpdateIcebergTable =>
+ // UPDATEs with subqueries may be rewritten using a UNION with two identical scan relations
+ // each scan relation will get its own dynamic filter that will be shared during execution
+ // the analyzer will assign different expr IDs for each scan relation output attributes
+ // that's why the condition may refer to invalid attr expr IDs and must be transformed
Review comment:
If I understand correctly, sometimes a plan gets transformed, from this:
```
ReplaceData
Update(id#1 IN (1, 5), data#2 = 'foo')
V2Relation(db.table, [id#1, data#2])
```
To this:
```
ReplaceData
Update(id#1 IN (1, 5), data#2 = 'foo')
Union([id#1, data#2])
V2Relation(db.table, [id#1, data#2])
V2Relation(db.table, [id#4, data#5])
```
So this is basically creating the dynamic filter for each scan separately and fixing up the attrs, from the update's IDs to the relation's IDs.
Is there a test case for this, or is it something that you ran into in practice?
It took me a while to understand this (assuming that I do) and I think it would be nice if it were more clear what is being replaced. Renaming `attrMap` to `scanRelationAttrs` would probably help so it is clear we're looking up the right attr for the scan relation.
It would also be good to to add a comment that explains why `u.table.output` is always going to match `relation.output`. In fact, I'm not sure that will. What happens when a column in the data is not used to produce the output? Is it pruned?
For example, `UPDATE t SET id = 1 WHERE true` doesn't need to project `id` from the table. If `id` is not projected, then the `attrMap` is incorrect. Is there a reason why that can't happen?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772558446
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelCommandDynamicPruning.scala
##########
@@ -109,6 +123,18 @@ case class RowLevelCommandDynamicPruning(spark: SparkSession) extends Rule[Logic
dynamicPruningSubqueries.reduce(And)
}
+ private def buildAttrMap(
+ tableAttrs: Seq[Attribute],
+ scanAttrs: Seq[Attribute]): AttributeMap[Attribute] = {
+
+ val attrMapping = tableAttrs.flatMap { tableAttr =>
+ scanAttrs
+ .find(scanAttr => scanAttr.name == tableAttr.name)
Review comment:
We could use a resolver to be safe but we are sure the names will match as the other attribute is constructed by cloning the original one. I'll update anyway.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#issuecomment-998222081
@rdblue, had to slightly modify `AssignmentUtils`. It turns out our alignment check was too aggressive. We did not catch this before as we did not invoke that code in `RewriteUpdateTable`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772619595
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan with updated rows
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val allRowsPlan = buildUpdateProjection(readRelation, assignments, updateCond)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition contains a subquery
+ private def buildReplaceDataWithUnionPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ // the same read relation will be used to read records that must be updated and be copied over
+ // DeduplicateRelations will take care of duplicated attr IDs
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan for records that match the cond and should be updated
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val matchedRowsPlan = Filter(updateCond, readRelation)
+ val updatedRowsPlan = buildUpdateProjection(matchedRowsPlan, assignments)
+
+ // build a plan for records that did not match the cond but had to be copied over
+ val remainingRowFilter = Not(EqualNullSafe(updateCond, Literal.TrueLiteral))
+ val remainingRowsPlan = Filter(remainingRowFilter, readRelation)
+
+ // new state is a union of updated and copied over records
+ val allRowsPlan = Union(updatedRowsPlan, remainingRowsPlan)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // this method assumes the assignments have been already aligned before
+ // the condition passed to this method may be different from the UPDATE condition
+ private def buildUpdateProjection(
+ plan: LogicalPlan,
+ assignments: Seq[Assignment],
+ cond: Expression = Literal.TrueLiteral): LogicalPlan = {
+
+ // TODO: avoid executing the condition for each column
Review comment:
I think we would need a custom node to handle it. I'll look into it after the release.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #3764: Spark: Implement copy-on-write UPDATE
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #3764:
URL: https://github.com/apache/iceberg/pull/3764#discussion_r772556787
##########
File path: spark/v3.2/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
##########
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.expressions.Alias
+import org.apache.spark.sql.catalyst.expressions.EqualNullSafe
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.If
+import org.apache.spark.sql.catalyst.expressions.Literal
+import org.apache.spark.sql.catalyst.expressions.Not
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.Assignment
+import org.apache.spark.sql.catalyst.plans.logical.Filter
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.catalyst.plans.logical.ReplaceData
+import org.apache.spark.sql.catalyst.plans.logical.Union
+import org.apache.spark.sql.catalyst.plans.logical.UpdateIcebergTable
+import org.apache.spark.sql.connector.iceberg.catalog.SupportsRowLevelOperations
+import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.UPDATE
+import org.apache.spark.sql.connector.iceberg.write.SupportsDelta
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+/**
+ * Assigns a rewrite plan for v2 tables that support rewriting data to handle UPDATE statements.
+ *
+ * This rule assumes the commands have been fully resolved and all assignments have been aligned.
+ * That's why it must be run after AlignRowLevelCommandAssignments.
+ *
+ * This rule also must be run in the same batch with DeduplicateRelations in Spark.
+ */
+object RewriteUpdateTable extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case u @ UpdateIcebergTable(aliasedTable, assignments, cond, None) if u.resolved =>
+ EliminateSubqueryAliases(aliasedTable) match {
+ case r @ DataSourceV2Relation(tbl: SupportsRowLevelOperations, _, _, _, _) =>
+ val operation = buildRowLevelOperation(tbl, UPDATE)
+ val table = RowLevelOperationTable(tbl, operation)
+ val rewritePlan = operation match {
+ case _: SupportsDelta =>
+ throw new AnalysisException("Delta updates are currently not supported")
+ case _ if cond.exists(SubqueryExpression.hasSubquery) =>
+ buildReplaceDataWithUnionPlan(r, table, assignments, cond)
+ case _ =>
+ buildReplaceDataPlan(r, table, assignments, cond)
+ }
+ UpdateIcebergTable(r, assignments, cond, Some(rewritePlan))
+
+ case p =>
+ throw new AnalysisException(s"$p is not an Iceberg table")
+ }
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition does NOT contain a subquery
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan with updated rows
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val allRowsPlan = buildUpdateProjection(readRelation, assignments, updateCond)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // build a rewrite plan for sources that support replacing groups of data (e.g. files, partitions)
+ // if the condition contains a subquery
+ private def buildReplaceDataWithUnionPlan(
+ relation: DataSourceV2Relation,
+ table: RowLevelOperationTable,
+ assignments: Seq[Assignment],
+ cond: Option[Expression]): ReplaceData = {
+
+ // resolve all needed attrs (e.g. metadata attrs for grouping data on write)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, table.operation)
+
+ // construct a read relation and include all required metadata columns
+ // the same read relation will be used to read records that must be updated and be copied over
+ // DeduplicateRelations will take care of duplicated attr IDs
+ val readAttrs = dedupAttrs(relation.output ++ metadataAttrs)
+ val readRelation = relation.copy(table = table, output = readAttrs)
+
+ // build a plan for records that match the cond and should be updated
+ val updateCond = cond.getOrElse(Literal.TrueLiteral)
+ val matchedRowsPlan = Filter(updateCond, readRelation)
+ val updatedRowsPlan = buildUpdateProjection(matchedRowsPlan, assignments)
+
+ // build a plan for records that did not match the cond but had to be copied over
+ val remainingRowFilter = Not(EqualNullSafe(updateCond, Literal.TrueLiteral))
+ val remainingRowsPlan = Filter(remainingRowFilter, readRelation)
+
+ // new state is a union of updated and copied over records
+ val allRowsPlan = Union(updatedRowsPlan, remainingRowsPlan)
+
+ // build a plan to replace read groups in the table
+ val writeRelation = relation.copy(table = table)
+ ReplaceData(writeRelation, allRowsPlan, relation)
+ }
+
+ // this method assumes the assignments have been already aligned before
+ // the condition passed to this method may be different from the UPDATE condition
+ private def buildUpdateProjection(
+ plan: LogicalPlan,
+ assignments: Seq[Assignment],
+ cond: Expression = Literal.TrueLiteral): LogicalPlan = {
+
+ // TODO: avoid executing the condition for each column
Review comment:
Unfortunately, the optimizer would squash the projections and replace the attribute with its expression. We had this limitation in the old implementation 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org