uros-b commented on code in PR #58296:
URL: https://github.com/apache/spark/pull/58296#discussion_r3865256229
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2CatalystRuntimeFilterSuite.scala:
##########
@@ -398,3 +579,73 @@ private class BothRuntimeFilteringInterfacesScan
override def filter(expressions: Array[Expression]): Unit = {}
}
+
+/** A scan declaring a filter attribute the read schema does not carry. */
+private class MissingFilterAttributeScan extends Scan with
SupportsRuntimeCatalystFiltering {
+
+ override def readSchema(): StructType = new StructType().add("part",
IntegerType)
+
+ override def filterAttributes(): Array[NamedReference] =
Array(FieldReference("missing"))
+
+ override def filter(expressions: Array[Expression]): Unit = {}
+}
+
+/**
+ * A scan breaking the rule that a filter attribute must be a top level read
schema column: it
+ * reports `part.nested` over the int column `part`, so resolving it fails on
the extract base.
+ */
+private class NestedFilterAttributeScan extends Scan with
SupportsRuntimeCatalystFiltering {
+
+ override def readSchema(): StructType = new StructType().add("part",
IntegerType)
+
+ override def filterAttributes(): Array[NamedReference] =
+ Array(FieldReference(Seq("part", "nested")))
+
+ override def filter(expressions: Array[Expression]): Unit = {}
+}
+
+/**
+ * A scan breaking the same rule over a struct column: it reports `s.tz` where
`s` is a struct, so
+ * resolving it succeeds and widens to `s` rather than failing.
+ */
+private class StructNestedFilterAttributeScan extends Scan with
SupportsRuntimeCatalystFiltering {
+
+ override def readSchema(): StructType =
+ new StructType().add("s", new StructType().add("tz", StringType))
+
+ override def filterAttributes(): Array[NamedReference] =
Array(FieldReference(Seq("s", "tz")))
+
+ override def filter(expressions: Array[Expression]): Unit = {}
+}
+
+private case class KeyedInputPartition(key: Int) extends InputPartition with
HasPartitionKey {
+ override def partitionKey(): InternalRow = InternalRow(key)
+}
+
+/**
+ * A scan reporting one set of partitions before filtering and another after,
so it can break the
+ * requirement to preserve the partitioning it originally reported.
+ */
+private class PartitioningBreakingScan(
+ initialPartitions: Seq[InputPartition],
+ afterFilter: Seq[InputPartition])
+ extends Scan with Batch with SupportsRuntimeCatalystFiltering {
+
+ private var filtered = false
+
+ override def readSchema(): StructType = new StructType().add("part",
IntegerType)
+
+ override def toBatch: Batch = this
+
+ override def planInputPartitions(): Array[InputPartition] =
+ if (filtered) afterFilter.toArray else initialPartitions.toArray
+
+ override def createReaderFactory(): PartitionReaderFactory =
+ throw new UnsupportedOperationException()
+
+ override def filterAttributes(): Array[NamedReference] =
Array(FieldReference("part"))
+
+ override def filter(expressions: Array[Expression]): Unit = {
+ filtered = true
+ }
+}
Review Comment:
(file should end with a newline)
--
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]
