peter-toth commented on code in PR #58522:
URL: https://github.com/apache/spark/pull/58522#discussion_r3947350296
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -324,6 +324,39 @@ trait KeyGroupedPartitioningRuntimeFilterTests extends
KeyGroupedPartitioningSui
}
}
}
+
+ test("SPARK-59248: runtime filtering with a pruned partition key keeps the
full key rows") {
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "false",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_FALLBACK_FILTER_RATIO.key -> "10",
+ SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true") {
+ createTable(items, itemsColumns, Array(identity("id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ s"(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ s"(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ s"(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val pCols = Array(
+ Column.create("store_id", IntegerType),
+ Column.create("item_id", IntegerType),
Review Comment:
**Finding 6.** This test does not reach the keyed branch of
`replanWithRuntimeFilters` with a pruned key, in either of the two suites that
run it. I instrumented `replanWithRuntimeFilters` to print `output`,
`runtimeFilters`, `filtered` and the partitioning it is handed.
`KeyGroupedPartitioningRuntimeFilterSuite` (V2 filtering) — the key is
pruned, but no filter is pushed:
output=List(item_id#144, price#145)
runtimeFilters=List(dynamicpruningexpression(cast(item_id#144 as bigint)
IN dynamicpruning#193))
filtered=false kp=Some(ArraySeq(store_id#143, item_id#144))
`item_id` is `IntegerType` here while `items.id` is `LongType`, so the join
wraps it and the DPP filter becomes `cast(item_id as bigint) IN ...`.
`DataSourceV2Strategy.translateRuntimeFilterV2` cannot translate that, and
`InMemoryTable` does not do iterative pushdown, so `pushRuntimeFilters` returns
`false`. The `filtered` branch never runs.
`KeyGroupedPartitioningCatalystRuntimeFilterSuite` (Catalyst filtering) —
the filter is pushed, but nothing is pruned:
output=List(store_id#25, item_id#26, price#27)
runtimeFilters=List(dynamicpruningexpression(cast(item_id#26 as bigint)
IN dynamicpruning#91))
filtered=true kp=Some(ArraySeq(store_id#25, item_id#26))
`InMemoryCatalystRuntimeFilterTable`'s scan keeps the full table schema, so
`store_id` stays in the scan output and `outputPartitioning` projects nothing
away.
So the test is green with `filteredPartitions` pointed back at
`outputPartitioning`. "Pointing the call site back at `outputPartitioning`
fails exactly those two and nothing else" in
[issuecomment-5564248263](https://github.com/apache/spark/pull/58522#issuecomment-5564248263)
is that measurement: this test should have been a third.
One line fixes it — the suite's own `purchasesColumns` already declares
`item_id` as `LongType`:
```suggestion
Column.create("item_id", LongType),
```
with `checkAnswer(df, Seq(Row(1L, 42.0f), Row(1L, 44.0f)))` on line 357.
Measured with that applied. On this branch `filtered=true` and the scan is
handed the full-width `[store_id, item_id]`, and the test passes. With
`filteredPartitions` fed `outputPartitioning` it is handed
`Some(Vector(item_id#26L))` and fails:
java.lang.ArrayIndexOutOfBoundsException: Index 1 out of bounds for
length 1
at
org.apache.spark.sql.catalyst.expressions.InterpretedHashFunction.hash(hash.scala:817)
at
org.apache.spark.sql.catalyst.util.InternalRowComparableWrapper.hashCode(...:74)
at
org.apache.spark.sql.execution.datasources.v2.PushDownUtils$.replanWithRuntimeFilters(...:327)
That is the runtime-filtering consequence from the bullet above, and nothing
else in the suite catches it.
Two smaller points on the same test. The Catalyst mixin can never prune a
column, so it runs a no-pruning variant whatever you do here — worth a word in
the comment, or move the test out of the shared trait. And the description's
test list does not mention this test at all, which is the one a reader looking
for the runtime-filtering case would search for.
--
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]
