JingsongLi commented on code in PR #10457: URL: https://github.com/apache/paimon/pull/10457#discussion_r4236717717
########## paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/RepartitionLargePaimonScan.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.paimon.spark.catalyst.optimizer + +import org.apache.paimon.spark.{PaimonScan, SparkConnectorOptions} +import org.apache.paimon.spark.read.BinPackingSplits +import org.apache.paimon.spark.util.{OptionUtils, SplitUtils} +import org.apache.paimon.types.BlobType + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Repartition, RepartitionOperation} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation +import org.apache.spark.sql.internal.SQLConf + +import scala.collection.JavaConverters._ + +/** Redistributes oversized scan partitions for downstream operators, preserving the scan itself. */ +object RepartitionLargePaimonScan extends Rule[LogicalPlan] { + + // injectOptimizerRule runs before V2 scan pushdown. The final user-provided optimizer batch + // runs after it, so register there as well when Spark initializes its optimizer rules. + def register(spark: SparkSession): Unit = { + val experimental = spark.experimental + if (!experimental.extraOptimizations.contains(this)) { + experimental.extraOptimizations = experimental.extraOptimizations :+ this + } + } + + override def apply(plan: LogicalPlan): LogicalPlan = { + if ( + !OptionUtils + .getOptionString(SparkConnectorOptions.READ_REPARTITION_LARGE_SCAN_ENABLED) + .toBoolean + ) { + return plan + } + + def rewrite(node: LogicalPlan): LogicalPlan = node match { + // Preserve an explicit shuffle and make repeated applications of this rule idempotent. + case repartition: RepartitionOperation + if repartition.shuffle && repartition.child.isInstanceOf[DataSourceV2ScanRelation] => + repartition + case relation: DataSourceV2ScanRelation if !relation.isStreaming => + relation.scan match { + case scan: PaimonScan + if scan.table + .rowType() + .getFields + .asScala + .exists(field => BlobType.isBlobFileField(field.`type`())) => + val partitions = scan.inputPartitions + val targetSize = BinPackingSplits.filesMaxPartitionBytes(scan.coreOptions, SQLConf.get) + val threshold = BigInt(targetSize) * 2 + val sizes = partitions.map { + partition => partition.splits.map(split => BigInt(SplitUtils.splitSize(split))).sum + } + if (targetSize > 0 && sizes.exists(_ > threshold)) { + val numPartitions = (sizes.sum + targetSize - 1) / targetSize + require( + numPartitions <= Int.MaxValue, + s"Scan repartition count $numPartitions exceeds ${Int.MaxValue}; " + + "increase the scan split target size" + ) + if (numPartitions <= partitions.size) { + relation + } else { + Repartition(numPartitions.toInt, shuffle = true, relation) Review Comment: [P1] Evaluate input-file expressions before the inserted shuffle When this option is enabled and a scan exceeds the threshold, inserting `Repartition` directly around the relation leaves an existing `Project` or `Filter` containing `input_file_name()` above the shuffle. The expression then runs in a downstream task without the source reader's `InputFileBlockHolder` context, so it returns an empty string and can change query results. I reproduced this on Spark 3.5.8 with AQE disabled: create a data-evolution/row-tracking table containing `id INT`, set `source.split.target-size=4kb`, write 5,000 rows into one normal Parquet file, then run `ALTER TABLE t ADD COLUMN payload BINARY COMMENT '__BLOB_FIELD'`. The full schema makes this scan eligible even though `payload` is pruned. With the option disabled, `SELECT id, input_file_name() FROM t` returns the actual file path and `SELECT id FROM t WHERE length(input_file_name()) > 0` returns all 5,000 rows. With the option enabled, the filename becomes empty and the same filtering query returns zero rows. Please evaluate/materialize the file-dependent expressions below the added shuffle, or skip affected scans. Add a regression test for both the projection and filtering cases. ########## paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/RepartitionLargePaimonScan.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.paimon.spark.catalyst.optimizer + +import org.apache.paimon.spark.{PaimonScan, SparkConnectorOptions} +import org.apache.paimon.spark.read.BinPackingSplits +import org.apache.paimon.spark.util.{OptionUtils, SplitUtils} +import org.apache.paimon.types.BlobType + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Repartition, RepartitionOperation} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation +import org.apache.spark.sql.internal.SQLConf + +import scala.collection.JavaConverters._ + +/** Redistributes oversized scan partitions for downstream operators, preserving the scan itself. */ +object RepartitionLargePaimonScan extends Rule[LogicalPlan] { + + // injectOptimizerRule runs before V2 scan pushdown. The final user-provided optimizer batch + // runs after it, so register there as well when Spark initializes its optimizer rules. + def register(spark: SparkSession): Unit = { + val experimental = spark.experimental + if (!experimental.extraOptimizations.contains(this)) { + experimental.extraOptimizations = experimental.extraOptimizations :+ this + } + } + + override def apply(plan: LogicalPlan): LogicalPlan = { + if ( + !OptionUtils + .getOptionString(SparkConnectorOptions.READ_REPARTITION_LARGE_SCAN_ENABLED) + .toBoolean + ) { + return plan + } + + def rewrite(node: LogicalPlan): LogicalPlan = node match { + // Preserve an explicit shuffle and make repeated applications of this rule idempotent. + case repartition: RepartitionOperation + if repartition.shuffle && repartition.child.isInstanceOf[DataSourceV2ScanRelation] => + repartition + case relation: DataSourceV2ScanRelation if !relation.isStreaming => + relation.scan match { + case scan: PaimonScan + if scan.table + .rowType() + .getFields + .asScala + .exists(field => BlobType.isBlobFileField(field.`type`())) => + val partitions = scan.inputPartitions + val targetSize = BinPackingSplits.filesMaxPartitionBytes(scan.coreOptions, SQLConf.get) + val threshold = BigInt(targetSize) * 2 + val sizes = partitions.map { + partition => partition.splits.map(split => BigInt(SplitUtils.splitSize(split))).sum + } + if (targetSize > 0 && sizes.exists(_ > threshold)) { + val numPartitions = (sizes.sum + targetSize - 1) / targetSize + require( + numPartitions <= Int.MaxValue, + s"Scan repartition count $numPartitions exceeds ${Int.MaxValue}; " + + "increase the scan split target size" + ) + if (numPartitions <= partitions.size) { + relation + } else { + Repartition(numPartitions.toInt, shuffle = true, relation) Review Comment: [P2] Keep dynamic partition pruning adjacent to the scan The late optimizer batch runs after Spark inserts dynamic partition pruning. Rewriting an existing `Filter(dynamicpruning, scan)` as `Filter(dynamicpruning, Repartition(scan))` prevents `DataSourceV2Strategy` from extracting that condition into `BatchScanExec.runtimeFilters`: its `PhysicalOperation` extractor cannot traverse `Repartition`. The condition instead becomes a regular `FilterExec` above the exchange, so Paimon's runtime partition filtering is never invoked and all source partitions are read and shuffled first. I reproduced this on Spark 3.5.8 with AQE disabled and DPP enabled: a Blob fact table has three `pt` partitions, each with 100 rows and 8 KiB payloads, and `source.split.target-size=64kb`. Join it on `pt` to a broadcast dimension containing `(pt, keep) = (0,1), (1,0), (2,0)`, with `WHERE d.keep = 1`, and select `f.id, length(f.payload)`. With this option disabled, the fact scan has a runtime partition filter, reads 4 files and emits 100 rows. Enabled, its runtime filters are empty, it reads 12 files and emits 300 rows before the ordinary filter reduces the result. Final rows are identical, but source pruning is lost. Please place the repartition above the existing dynamic-pruning filter while preserving Spark's scan extraction, and add an integration assertion that the fact scan retains its runtime 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
