fresh-borzoni commented on code in PR #3397:
URL: https://github.com/apache/fluss/pull/3397#discussion_r3329601900
##########
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkPrimaryKeyTableReadTest.scala:
##########
@@ -432,6 +434,28 @@ class SparkPrimaryKeyTableReadTest extends
FlussSparkTestBase {
}
}
+ test("Spark Read: mixed partition and non-partition filter (PK table)") {
Review Comment:
Could we add the opposite of the mixed-filter test above? i.e. a
partition-only query (WHERE dt = '...') that asserts there's no FilterExec left
in the plan. That's really the core thing this PR changes, and right now
nothing checks it - easy to accidentally regress later.
##########
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/utils/SparkPartitionPredicate.scala:
##########
@@ -28,25 +28,40 @@ import scala.jdk.CollectionConverters._
/** Extracts a partition-key predicate and prunes the partition list at
planning time. */
object SparkPartitionPredicate {
- def extract(tableInfo: TableInfo, predicates: Seq[Predicate]):
Option[FlussPredicate] = {
+ def extract(
+ tableInfo: TableInfo,
+ predicates: Seq[Predicate]): (Seq[Predicate], Option[FlussPredicate]) = {
val partitionKeys = tableInfo.getPartitionKeys
- if (partitionKeys.isEmpty) return None
+ if (partitionKeys.isEmpty) {
+ return (predicates, None)
+ }
val rowType = PartitionUtils.partitionRowType(tableInfo)
val onlyPartitionKeys = new PartitionPredicateVisitor(partitionKeys)
- val converted = predicates.flatMap {
+ // Separate predicates: those that can be converted and only touch
partition keys
+ val (partitionPredicates, nonPartitionPredicates) = predicates.partition {
+ sparkPredicate =>
+ SparkPredicateConverter
+ .convert(rowType, sparkPredicate)
Review Comment:
Small thing: we call convert(...) here to classify the predicate, then call
it again right below to actually use it. Could we convert once and reuse the
result?
Something like mapping each predicate to its converted form, then
partition-ing on isDefined. Only runs at planning time so not a big deal, just
avoids doing the same work twice.
##########
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScanBuilder.scala:
##########
@@ -96,8 +98,11 @@ trait FlussLakeSupportsPushDownV2Filters extends
FlussSupportsPushDownV2Filters
def flussConfig: FlussConfiguration
override def pushPredicates(predicates: Array[Predicate]): Array[Predicate]
= {
+ val nonPartitionPredicates = super.pushPredicates(predicates)
Review Comment:
Here the lake path calls super.pushPredicates, which for ARROW tables runs
convertAndStorePredicates and sets pushedPredicate/acceptedPredicates, then we
overwrite both a few lines down.
So that work gets thrown away every time.
Could we pull the partition-extraction bit into a small helper and call that
instead of super, so we skip the ARROW step we don't use?
##########
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/utils/SparkPartitionPredicate.scala:
##########
@@ -63,4 +78,19 @@ object SparkPartitionPredicate {
PartitionUtils.toPartitionRow(p.getResolvedPartitionSpec.getPartitionValues,
rowType))
}
}
+
+ /**
+ * Tests whether a partition (described by its ordered partition values)
matches the given
+ * predicate. Returns true when no predicate is provided.
+ */
+ def matchesPartition(
Review Comment:
If partitionValues ever comes in empty, toPartitionRow gets the wrong number
of values and could blow up.
Should we just guard it (case Some(_) if partitionValues.isEmpty => true)?
--
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]
