You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "szehon-ho (via GitHub)" <gi...@apache.org> on 2026/04/02 23:59:04 UTC
[PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
szehon-ho opened a new pull request, #55179:
URL: https://github.com/apache/spark/pull/55179
## Summary
When `OptimizeMetadataOnlyDeleteFromTable` fails to push standard V2 predicates (e.g. `IN`, `STARTS_WITH`) for a metadata-only delete, it now falls back to a second pass that:
1. Converts partition-column filters to `PartitionPredicate`s (reusing SPARK-55596 infrastructure)
2. Translates remaining data-column filters to standard V2 predicates
3. Combines them (partition predicates first) and calls `table.canDeleteWhere`
This mirrors the two-pass approach already used for scan filter pushdown in `PushDownUtils.pushPartitionPredicates`.
### Changes
- **`OptimizeMetadataOnlyDeleteFromTable`**: Added `tryDeleteWithPartitionPredicates` fallback method and `tryTranslateToV2` helper
- **`PushDownUtils`**: Extracted `createPartitionPredicates` and made `flattenNestedPartitionFilters` package-private for reuse; `getPartitionPredicateSchema` now returns `None` for empty partition fields
- **`InMemoryTableWithV2Filter`**: Extracted `evalPredicate` to companion object for reuse by test tables
- **`InMemoryPartitionPredicateDeleteTable`** (new): Test table supporting `PartitionPredicate`s and configurable data predicate acceptance
- **`DataSourceV2EnhancedDeleteFilterSuite`** (new): 9 test cases covering first-pass accept, second-pass accept/reject, mixed partition+data filters, UDF on non-contiguous partition columns, multiple PartitionPredicates, and row-level fallback
## Test plan
- [x] `DataSourceV2EnhancedDeleteFilterSuite` — 9/9 pass
- [x] `DataSourceV2EnhancedPartitionFilterSuite` — 19/19 pass (no regressions)
- [x] `GroupBasedDeleteFromTableSuite` — 32/32 pass (no regressions)
- [x] Scalastyle — 0 errors
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #55179:
URL: https://github.com/apache/spark/pull/55179#issuecomment-4196851375
LGTM if CI is green
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55179:
URL: https://github.com/apache/spark/pull/55179#discussion_r3043815173
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/OptimizeMetadataOnlyDeleteFromTable.scala:
##########
@@ -70,6 +74,42 @@ object OptimizeMetadataOnlyDeleteFromTable extends Rule[LogicalPlan] with Predic
}.toArray
}
+ /**
+ * Attempts to convert partition-column filters to [[PartitionPredicate]]s and
+ * combine them with translated V2 data filters for a metadata-only delete. (See SPARK-55596)
+ *
+ * Returns [[Some]] with the plan if the table accepts the combined predicates,
+ * or [[None]] if partition predicates cannot be created or the table rejects them.
+ */
+ private def tryDeleteWithPartitionPredicates(
+ table: SupportsDeleteV2,
+ relation: DataSourceV2Relation,
+ normalizedPredicates: Seq[Expression]): Option[LogicalPlan] = {
+ for {
+ partitionFields <- PushDownUtils.getPartitionPredicateSchema(relation)
+ flattenedFilters = PushDownUtils.flattenNestedPartitionFilters(
+ normalizedPredicates, partitionFields).keys.toSeq
+ (candidatePredicates, remainingFilters) =
+ PushDownUtils.createPartitionPredicates(flattenedFilters, partitionFields)
+ // None if no partition predicates created
+ partPredicates <- Option.when(candidatePredicates.nonEmpty)(candidatePredicates)
+ // None if any remaining filter cannot be translated to V2
+ dataV2Filters <- tryTranslateToV2(remainingFilters)
+ combined = partPredicates.toArray ++ dataV2Filters
+ if table.canDeleteWhere(combined)
+ } yield {
+ logDebug(s"Switching to delete with PartitionPredicate filters: " +
+ s"${combined.mkString("[", ", ", "]")}")
+ DeleteFromTableWithFilters(relation, combined.toImmutableArraySeq)
+ }
+ }
+
+ /** Translates all expressions to V2 filters, or returns [[None]] if any fail. */
+ private def tryTranslateToV2(predicates: Seq[Expression]): Option[Array[Predicate]] = {
+ val filters = toDataSourceV2Filters(predicates)
Review Comment:
Nit: this seems to be only callsite of `toDataSourceV2Filters()` so you can probably combine them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #55179:
URL: https://github.com/apache/spark/pull/55179#discussion_r3030884431
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala:
##########
@@ -140,31 +140,36 @@ object InMemoryTableWithV2Filter {
partitionNames: Seq[String],
filters: Array[Predicate]): Iterable[Seq[Any]] = {
keys.filter { partValues =>
- filters.flatMap(splitAnd).forall {
- case p: Predicate if p.name().equals("=") =>
- p.children()(1).asInstanceOf[LiteralValue[_]].value ==
- InMemoryBaseTable.extractValue(p.children()(0).toString, partitionNames, partValues)
- case p: Predicate if p.name().equals("<=>") =>
- val attrVal = InMemoryBaseTable
- .extractValue(p.children()(0).toString, partitionNames, partValues)
- val value = p.children()(1).asInstanceOf[LiteralValue[_]].value
- if (attrVal == null && value == null) {
- true
- } else if (attrVal == null || value == null) {
- false
- } else {
- value == attrVal
- }
- case p: Predicate if p.name().equals("IS_NULL") =>
- val attr = p.children()(0).toString
- null == InMemoryBaseTable.extractValue(attr, partitionNames, partValues)
- case p: Predicate if p.name().equals("IS_NOT_NULL") =>
- val attr = p.children()(0).toString
- null != InMemoryBaseTable.extractValue(attr, partitionNames, partValues)
- case p: Predicate if p.name().equals("ALWAYS_TRUE") => true
- case f =>
- throw new IllegalArgumentException(s"Unsupported filter type: $f")
- }
+ val resolve: String => Any = attr =>
+ InMemoryBaseTable.extractValue(attr, partitionNames, partValues)
+ filters.flatMap(splitAnd).forall(evalPredicate(_, resolve))
+ }
+ }
+
+ /**
+ * Evaluates a single V2 predicate by resolving column values through the
+ * given function. Supports =, <=>, IS_NULL, IS_NOT_NULL, and ALWAYS_TRUE.
+ */
+ def evalPredicate(
Review Comment:
just refactor for re-use in new test InMemoryTable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #55179:
URL: https://github.com/apache/spark/pull/55179#discussion_r3047274708
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/OptimizeMetadataOnlyDeleteFromTable.scala:
##########
@@ -70,6 +74,42 @@ object OptimizeMetadataOnlyDeleteFromTable extends Rule[LogicalPlan] with Predic
}.toArray
}
+ /**
+ * Attempts to convert partition-column filters to [[PartitionPredicate]]s and
+ * combine them with translated V2 data filters for a metadata-only delete. (See SPARK-55596)
+ *
+ * Returns [[Some]] with the plan if the table accepts the combined predicates,
+ * or [[None]] if partition predicates cannot be created or the table rejects them.
+ */
+ private def tryDeleteWithPartitionPredicates(
+ table: SupportsDeleteV2,
+ relation: DataSourceV2Relation,
+ normalizedPredicates: Seq[Expression]): Option[LogicalPlan] = {
+ for {
+ partitionFields <- PushDownUtils.getPartitionPredicateSchema(relation)
+ flattenedFilters = PushDownUtils.flattenNestedPartitionFilters(
+ normalizedPredicates, partitionFields).keys.toSeq
+ (candidatePredicates, remainingFilters) =
+ PushDownUtils.createPartitionPredicates(flattenedFilters, partitionFields)
+ // None if no partition predicates created
+ partPredicates <- Option.when(candidatePredicates.nonEmpty)(candidatePredicates)
+ // None if any remaining filter cannot be translated to V2
+ dataV2Filters <- tryTranslateToV2(remainingFilters)
+ combined = partPredicates.toArray ++ dataV2Filters
+ if table.canDeleteWhere(combined)
+ } yield {
+ logDebug(s"Switching to delete with PartitionPredicate filters: " +
+ s"${combined.mkString("[", ", ", "]")}")
+ DeleteFromTableWithFilters(relation, combined.toImmutableArraySeq)
+ }
+ }
+
+ /** Translates all expressions to V2 filters, or returns [[None]] if any fail. */
+ private def tryTranslateToV2(predicates: Seq[Expression]): Option[Array[Predicate]] = {
+ val filters = toDataSourceV2Filters(predicates)
Review Comment:
Good catch, thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #55179:
URL: https://github.com/apache/spark/pull/55179#issuecomment-4204147902
thanks, merging to master!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #55179: [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete
URL: https://github.com/apache/spark/pull/55179
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #55179:
URL: https://github.com/apache/spark/pull/55179#issuecomment-4197398632
@szehon-ho , can you please update the PR description to follow the template?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #55179:
URL: https://github.com/apache/spark/pull/55179#discussion_r3030888631
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -218,9 +237,9 @@ object PushDownUtils extends Logging {
* (identity transform on a nested field), the analyzer produces
* `GetStructField(attr("s"), "tz")`. This method replaces that chain with `attr("s.tz")`.
*
- * Returns a map from normalized expression to original.
+ * Returns a map from flattened expression to original.
*/
- private def normalizeNestedPartitionFilters(
+ private[v2] def flattenNestedPartitionFilters(
Review Comment:
rename, because 'normalize' is already used in OptimizeMetadataOnlyDelete
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #55179:
URL: https://github.com/apache/spark/pull/55179#issuecomment-4201529928
Done, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55179:
URL: https://github.com/apache/spark/pull/55179#discussion_r3031199666
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -149,7 +149,7 @@ object PushDownUtils extends Logging {
case _ => None
}
if (fields.length == transforms.length) {
- Some(fields.toSeq)
+ Some(fields.toSeq).filter(_.nonEmpty)
Review Comment:
This `.filter(_.nonEmpty)` guard is redundant: the outer check at line 139 guarantees `transforms.nonEmpty`, and `fields.length == transforms.length` at line 151 ensures `fields` is non-empty.
```suggestion
Some(fields.toSeq)
```
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryPartitionPredicateDeleteTable.scala:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.expressions.filter.{PartitionPredicate, Predicate}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.util.ArrayImplicits._
+
+/**
+ * In-memory table that supports row-level operations and accepts [[PartitionPredicate]]s
+ * in V2 [[canDeleteWhere]]/[[deleteWhere]] for metadata-only deletes.
+ *
+ * Contains some knobs to control acceptance of various partition and data predicates.
+ */
+class InMemoryPartitionPredicateDeleteTable(
+ name: String,
+ schema: StructType,
+ partitioning: Array[Transform],
+ properties: util.Map[String, String])
+ extends InMemoryRowLevelOperationTable(name, schema, partitioning, properties) {
+
+ private val acceptPartitionPredicates: Boolean =
+ properties.getOrDefault(
+ InMemoryPartitionPredicateDeleteTable.AcceptPartitionPredicatesKey, "true").toBoolean
+
+ private val acceptDataPredicates: Boolean =
+ properties.getOrDefault(
+ InMemoryPartitionPredicateDeleteTable.AcceptDataPredicatesKey, "false").toBoolean
+
+ private val partPaths = partCols.map(_.mkString(".")).toSet
+
+ private def refsOnlyPartCols(p: Predicate): Boolean =
+ p.references().forall(ref => partPaths.contains(ref.fieldNames().mkString(".")))
+
+ override def canDeleteWhere(predicates: Array[Predicate]): Boolean = {
+ predicates.forall {
+ case _: PartitionPredicate => acceptPartitionPredicates
+ case p =>
+ InMemoryTableWithV2Filter.supportsPredicates(Array(p)) &&
+ (acceptDataPredicates || refsOnlyPartCols(p))
+ }
+ }
+
+ override def deleteWhere(predicates: Array[Predicate]): Unit = dataMap.synchronized {
+ val (partPreds, standardPreds) = predicates.partition(_.isInstanceOf[PartitionPredicate])
+ val (partStdPreds, dataStdPreds) = standardPreds.partition(refsOnlyPartCols)
+
+ val candidateKeys = if (partStdPreds.nonEmpty) {
+ InMemoryTableWithV2Filter.filtersToKeys(
+ dataMap.keys, partCols.map(_.toSeq.quoted).toImmutableArraySeq, partStdPreds)
+ } else {
+ dataMap.keys
+ }
+
+ // Handle partition predicates.
+ val keysToProcess = if (partPreds.nonEmpty) {
+ val pArr = partPreds.map(_.asInstanceOf[PartitionPredicate])
+ candidateKeys.filter { key =>
+ val partRow = PartitionInternalRow(key.toArray)
+ pArr.forall(_.eval(partRow))
+ }
+ } else {
+ candidateKeys
+ }
+
+ // Handle data predicates (simulate data source with data column statistics)
Review Comment:
The comment says "data column statistics" but the code evaluates predicates row-by-row, not via statistics.
```suggestion
// Handle data predicates (simulate a data source applying row-level data filters)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #55179:
URL: https://github.com/apache/spark/pull/55179#discussion_r3042131592
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryPartitionPredicateDeleteTable.scala:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.expressions.filter.{PartitionPredicate, Predicate}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.util.ArrayImplicits._
+
+/**
+ * In-memory table that supports row-level operations and accepts [[PartitionPredicate]]s
+ * in V2 [[canDeleteWhere]]/[[deleteWhere]] for metadata-only deletes.
+ *
+ * Contains some knobs to control acceptance of various partition and data predicates.
+ */
+class InMemoryPartitionPredicateDeleteTable(
+ name: String,
+ schema: StructType,
+ partitioning: Array[Transform],
+ properties: util.Map[String, String])
+ extends InMemoryRowLevelOperationTable(name, schema, partitioning, properties) {
+
+ private val acceptPartitionPredicates: Boolean =
+ properties.getOrDefault(
+ InMemoryPartitionPredicateDeleteTable.AcceptPartitionPredicatesKey, "true").toBoolean
+
+ private val acceptDataPredicates: Boolean =
+ properties.getOrDefault(
+ InMemoryPartitionPredicateDeleteTable.AcceptDataPredicatesKey, "false").toBoolean
+
+ private val partPaths = partCols.map(_.mkString(".")).toSet
+
+ private def refsOnlyPartCols(p: Predicate): Boolean =
+ p.references().forall(ref => partPaths.contains(ref.fieldNames().mkString(".")))
+
+ override def canDeleteWhere(predicates: Array[Predicate]): Boolean = {
+ predicates.forall {
+ case _: PartitionPredicate => acceptPartitionPredicates
+ case p =>
+ InMemoryTableWithV2Filter.supportsPredicates(Array(p)) &&
+ (acceptDataPredicates || refsOnlyPartCols(p))
+ }
+ }
+
+ override def deleteWhere(predicates: Array[Predicate]): Unit = dataMap.synchronized {
+ val (partPreds, standardPreds) = predicates.partition(_.isInstanceOf[PartitionPredicate])
+ val (partStdPreds, dataStdPreds) = standardPreds.partition(refsOnlyPartCols)
+
+ val candidateKeys = if (partStdPreds.nonEmpty) {
+ InMemoryTableWithV2Filter.filtersToKeys(
+ dataMap.keys, partCols.map(_.toSeq.quoted).toImmutableArraySeq, partStdPreds)
+ } else {
+ dataMap.keys
+ }
+
+ // Handle partition predicates.
+ val keysToProcess = if (partPreds.nonEmpty) {
+ val pArr = partPreds.map(_.asInstanceOf[PartitionPredicate])
+ candidateKeys.filter { key =>
+ val partRow = PartitionInternalRow(key.toArray)
+ pArr.forall(_.eval(partRow))
+ }
+ } else {
+ candidateKeys
+ }
+
+ // Handle data predicates (simulate data source with data column statistics)
Review Comment:
Hm yea I thought about this too. I think using 'simulate a data source with data column statistics' is more accurate. The test table goes row-by-row, but its a simulation.
This is for the use-case where a data column predicate can still be completely handled by the data source. For example, iceberg min-max statistics.
```delete from table t where x < 10```, and all Iceberg data files have min value less than 10.
Of course the InMemoryPartitionPredicateDeleteTable is not structured that way, so its a 'simulation'. It'd be a bit overkill to implement min/max stats to make it a more accurate simulation? Maybe in a follow up.
By the way, this pushdown is not about PartitionPredicate, but the existing case, its just I wanted to add a test that mixes the two.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56346][SQL] Use PartitionPredicate in DSV2 Metadata Only Delete [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #55179:
URL: https://github.com/apache/spark/pull/55179#discussion_r3030886655
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -177,6 +177,32 @@ object PushDownUtils extends Logging {
}
}
+ /**
+ * Separates partition filters from data filters and converts pushable partition
Review Comment:
again, refactor for re-use in OptimizeMetadataOnlyDeleteQuery
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org