aokolnychyi commented on a change in pull request #35395: URL: https://github.com/apache/spark/pull/35395#discussion_r824997052
########## File path: sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala ########## @@ -0,0 +1,96 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.connector.catalog + +import java.util + +import org.apache.spark.sql.connector.distributions.{Distribution, Distributions} +import org.apache.spark.sql.connector.expressions.{FieldReference, LogicalExpressions, NamedReference, SortDirection, SortOrder, Transform} +import org.apache.spark.sql.connector.read.{Scan, ScanBuilder} +import org.apache.spark.sql.connector.write.{BatchWrite, LogicalWriteInfo, RequiresDistributionAndOrdering, RowLevelOperation, RowLevelOperationBuilder, RowLevelOperationInfo, Write, WriteBuilder, WriterCommitMessage} +import org.apache.spark.sql.connector.write.RowLevelOperation.Command +import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +class InMemoryRowLevelOperationTable( + name: String, + schema: StructType, + partitioning: Array[Transform], + properties: util.Map[String, String]) + extends InMemoryTable(name, schema, partitioning, properties) with SupportsRowLevelOperations { + + override def newRowLevelOperationBuilder( + info: RowLevelOperationInfo): RowLevelOperationBuilder = { + () => PartitionBasedOperation(info.command) + } + + case class PartitionBasedOperation(command: Command) extends RowLevelOperation { + private final val PARTITION_COLUMN_REF = FieldReference(PartitionKeyColumn.name) + + var configuredScan: InMemoryBatchScan = _ + + override def requiredMetadataAttributes(): Array[NamedReference] = { + Array(PARTITION_COLUMN_REF) + } + + override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = { + new InMemoryScanBuilder(schema) { + override def build: Scan = { + val scan = super.build() + configuredScan = scan.asInstanceOf[InMemoryBatchScan] + scan + } + } + } + + override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = new WriteBuilder { + + override def build(): Write = new Write with RequiresDistributionAndOrdering { + override def requiredDistribution(): Distribution = { + Distributions.clustered(Array(PARTITION_COLUMN_REF)) + } + + override def requiredOrdering(): Array[SortOrder] = { + Array[SortOrder]( + LogicalExpressions.sort( + PARTITION_COLUMN_REF, + SortDirection.ASCENDING, + SortDirection.ASCENDING.defaultNullOrdering()) + ) + } + + override def toBatch: BatchWrite = PartitionBasedReplaceData(configuredScan) + + override def description(): String = "InMemoryWrite" + } + } + + override def description(): String = "InMemoryPartitionReplaceOperation" + } + + private case class PartitionBasedReplaceData(scan: InMemoryBatchScan) extends TestBatchWrite { + + override def commit(messages: Array[WriterCommitMessage]): Unit = dataMap.synchronized { + val newData = messages.map(_.asInstanceOf[BufferedRows]) + val readRows = scan.data.flatMap(_.asInstanceOf[BufferedRows].rows) Review comment: Yeah, the API would definitely work with subqueries. We won't even have to do much to support them. The example above contained a subquery and was rewritten correctly (I executed my prototype). ``` DELETE FROM t WHERE id IN (SELECT * FROM deleted_id) ``` ``` == Optimized Logical Plan == ReplaceData RelationV2[id#66, dep#67] t +- Project [id#66, dep#67] +- Filter NOT (exists#86 <=> true) +- Join ExistenceJoin(exists#86), (id#66 = value#19) :- Project [id#66, dep#67] : +- Filter dynamicpruning#85 [_file_name#70] : : +- Project [_file_name#83] : : +- Join LeftSemi, (id#81 = value#72) : : :- Project [id#81, _file_name#83] : : : +- RelationV2[id#81, dep#82, _file_name#83] t : : +- LocalRelation [value#72] : +- RelationV2[id#66, dep#67, _file_name#70] t +- LocalRelation [value#19] ``` > How does Spark know how to get the file name column? There is no DS v2 API for Delta to tell Spark: hey you can select _file_name column to get the "group id". Are we going to add an implicit assumption that, the Scan will add an extra column silently like my proposal? If yes, I think this solves the problem and is actually very similar to my proposal with the difference that we use the existing hidden column and runtime filter API to report "group id" from the source and use it to determine the final affected "groups". Delta's `Scan` for row-level operations can inherit `SupportsRuntimeFiltering` and report `_file_name` as a filtering attribute via `SupportsRuntimeFiltering$filterAttributes`. The `_file_name` should be a metadata column that can be projected via `RowLevelOperation$requiredMetadataAttributes`. That's what I propose as a way to report which column to use to filter out group IDs. Both metadata columns and runtime filtering is already supported for DS V2 so using runtime filtering with subqueries is a matter of adding an optimizer rule. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
