peter-toth commented on code in PR #59269:
URL: https://github.com/apache/spark/pull/59269#discussion_r4211300571
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -2010,6 +2182,188 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
}
}
+ test("FileSourceStrategy skips an offered conjunct on the rows the reader
marks as checked") {
+ // The scan gets a column for the reader to mark the rows it checked in,
and the post-scan
+ // Filter evaluates the offered conjunct as `checked OR conjunct`. A
conjunct that is not
+ // offered stays as it is, and the column goes no further than the Filter.
The bloom names `K`
+ // where the relation says `k`, as the query can spell it, so the Filter's
conjunct and the one
+ // offered to the scan differ in name.
+ withTempDir { dir =>
+ val path = writeSingleParquetFile(dir,
+ spark.range(0, 50).selectExpr("id AS k", "CAST(id AS STRING) AS v"),
+ rowGroupSize = 64 * 1024L)
+ withSQLConf(SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key ->
"true") {
+ val relation = spark.read.parquet(path).queryExecution.optimizedPlan
+ def column(name: String): Attribute =
+ relation.output.find(_.name == name).getOrElse(fail(s"no $name in
the relation output"))
+ val bloom = BloomFilterMightContain(bloomLiteralOf(42L),
+ new XxHash64(Seq(column("k").withName("K"))))
+ val notOffered = EqualTo(column("v"), Literal("42"))
+ val logical = LogicalFilter(And(bloom, notOffered), relation)
+ val physical = FileSourceStrategy(logical).headOption
+ .getOrElse(fail(s"FileSourceStrategy did not plan $logical"))
+ val scan = scanOf(physical)
+ val checkedName = FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME
+ val checked = scan.output.find(_.name == checkedName)
+ .getOrElse(fail(s"the scan must output the checked column; got
${scan.output}"))
+ assert(scan.requiredSchema.fieldNames.toSeq == Seq("k", "v",
checkedName),
+ s"the reader must be asked for the column too; got
${scan.requiredSchema}")
+ val conjuncts =
+ physical.collect { case f: FilterExec =>
splitConjunctivePredicates(f.condition) }.flatten
+ assert(conjuncts.exists(_.semanticEquals(Or(checked, bloom))),
Review Comment:
Thanks, done in
[9a71742c400](https://github.com/apache/spark/pull/59269/commits/9a71742c400a9104bb4595ac582a449a6e6fede4).
- A new test plans a real `InjectRuntimeFilter` join. It finds the `IF`
around the bloom that holds the subquery, and reads the marks off the scan.
- The reader tests now read row groups of 20 rows in batches of 4.
- Row mode is moot now, since a row-mode scan gets no column (your comment
on L373).
- `withStorageFilters` stays without the column. Those tests check the rows
the reader emits, and the marks have their own tests.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -158,24 +158,30 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
* filters do, rather than the spelling the query used.
*
* They stay in the post-scan `Filter` as well, the way a pushed data filter
does. The reader is
- * offered them, not obliged to honor them, so the plan keeps the exact
check. What it costs is
- * evaluating the conjunct a second time for the rows the scan emits. What
it buys is a reader
- * free to give up on a file it cannot read in part, or on a stretch of rows
whose survivors cost
- * too much to keep track of, without the answer depending on it.
+ * offered them, not obliged to honor them, so the plan keeps the exact
check. What it buys is a
+ * reader free to give up on a file it cannot read in part, or on a stretch
of rows whose
+ * survivors cost too much to keep track of, without the answer depending on
it. The `Filter`
+ * skips the check on the rows the reader marks as checked, in a column the
scan adds for that
+ * (see [[FileFormat.buildReaderWithStorageFilters]]). So the `Filter`
evaluates a conjunct only
Review Comment:
Done in
[9a71742c400](https://github.com/apache/spark/pull/59269/commits/9a71742c400a9104bb4595ac582a449a6e6fede4).
The contract lives in `FileFormat.buildReaderWithStorageFilters`, and the
other places point to it. `FileSourceStrategy` no longer says the `Filter`
evaluates a conjunct only on unchecked rows. `ParquetStorageFilter` no longer
carries the column (your comment on L213). The conf doc describes the columnar
case without naming `spark.sql.codegen.maxFields`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -1924,9 +1943,162 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
}
}
+ // The `(k, v)` schema with the column the planner adds for the reader to
mark the rows it checked
+ // in. A filter created over it fills that column.
+ private def checkedSchema(valueType: DataType = StringType): StructType =
+ kvSchema(valueType =
valueType).add(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME, BooleanType)
+
+ // Reads `path` with `filter`, which was created over `schema`, and returns
each row's key, as
+ // `keyOf` reads it, with what the checked column says for the row. It goes
through the Hadoop
+ // initialize, since the test-only one projects only columns the file has.
+ private def readChecked(
Review Comment:
Done in
[9a71742c400](https://github.com/apache/spark/pull/59269/commits/9a71742c400a9104bb4595ac582a449a6e6fede4).
`readChecked` builds the reader with `formatReader` and `drain`, under the
session conf. The cap comes from the conf, and so does off heap, with `asTask`.
`hadoopSplit` is gone, and the 2-arg `initialize` test is back to master's form.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -350,23 +360,36 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
}
}
- val outputDataSchema = (readDataColumns ++
generatedMetadataColumns).toStructType
-
// Offered conjuncts become `storageFilters` on the scan and stay in the
post-scan Filter too.
- // This runs here because eligibility depends on `outputDataSchema`.
- val storageFilters =
- storageFiltersFor(normalizedFilters, fsRelation, readDataColumns,
outputDataSchema)
+ // This runs here because eligibility depends on the data columns the
reader reads.
+ val storageFilters = storageFiltersFor(normalizedFilters, fsRelation,
readDataColumns,
+ (readDataColumns ++ generatedMetadataColumns).toStructType)
+ // The column the reader marks the rows it checked in, which the
post-scan Filter reads. See
+ // `FileFormat.buildReaderWithStorageFilters`. It is left out where it
alone would push the
+ // scan past the whole-stage codegen field limit, which would cost the
scan its columnar
+ // output. The post-scan Filter then evaluates the offered conjuncts on
every row.
+ val storageFilterCheckedColumn = Option.when(storageFilters.nonEmpty) {
+ AttributeReference(FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME,
BooleanType)()
+ }.filterNot { checked =>
+ val conf = fsRelation.sparkSession.sessionState.conf
+ val output =
Review Comment:
Done in
[9a71742c400](https://github.com/apache/spark/pull/59269/commits/9a71742c400a9104bb4595ac582a449a6e6fede4).
`readerDataColumns` is the shared base, and the condition is positive now, in
a `filter` on the option. The `_metadata` `Project` is built from the scan's
output without the flattened metadata attributes. The `storageFiltersFor`
parameter is `readerDataSchema` now.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilterSuite.scala:
##########
@@ -2010,6 +2182,188 @@ class ParquetStorageFilterSuite extends QueryTest with
SharedSparkSession
}
}
+ test("FileSourceStrategy skips an offered conjunct on the rows the reader
marks as checked") {
+ // The scan gets a column for the reader to mark the rows it checked in,
and the post-scan
+ // Filter evaluates the offered conjunct as `checked OR conjunct`. A
conjunct that is not
+ // offered stays as it is, and the column goes no further than the Filter.
The bloom names `K`
+ // where the relation says `k`, as the query can spell it, so the Filter's
conjunct and the one
+ // offered to the scan differ in name.
+ withTempDir { dir =>
+ val path = writeSingleParquetFile(dir,
+ spark.range(0, 50).selectExpr("id AS k", "CAST(id AS STRING) AS v"),
+ rowGroupSize = 64 * 1024L)
+ withSQLConf(SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key ->
"true") {
+ val relation = spark.read.parquet(path).queryExecution.optimizedPlan
+ def column(name: String): Attribute =
+ relation.output.find(_.name == name).getOrElse(fail(s"no $name in
the relation output"))
+ val bloom = BloomFilterMightContain(bloomLiteralOf(42L),
+ new XxHash64(Seq(column("k").withName("K"))))
+ val notOffered = EqualTo(column("v"), Literal("42"))
+ val logical = LogicalFilter(And(bloom, notOffered), relation)
+ val physical = FileSourceStrategy(logical).headOption
+ .getOrElse(fail(s"FileSourceStrategy did not plan $logical"))
+ val scan = scanOf(physical)
+ val checkedName = FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME
+ val checked = scan.output.find(_.name == checkedName)
+ .getOrElse(fail(s"the scan must output the checked column; got
${scan.output}"))
+ assert(scan.requiredSchema.fieldNames.toSeq == Seq("k", "v",
checkedName),
+ s"the reader must be asked for the column too; got
${scan.requiredSchema}")
+ val conjuncts =
+ physical.collect { case f: FilterExec =>
splitConjunctivePredicates(f.condition) }.flatten
+ assert(conjuncts.exists(_.semanticEquals(Or(checked, bloom))),
+ s"the offered conjunct must be skipped on checked rows; got
$conjuncts")
+ assert(conjuncts.exists(_.semanticEquals(notOffered)),
+ s"a conjunct not offered must stay as it is; got $conjuncts")
+ assert(!physical.output.exists(_.name == checkedName),
+ s"the column must not leave the scan's Filter; got
${physical.output}")
+ }
+ }
+ }
+
+ test("FileSourceStrategy offers nothing to a relation with a column named
like the checked one") {
+ // The scan adds a column of that name for the reader to fill, so a column
of the relation's own
+ // must not be taken for it, in any case. With nothing offered, the scan
gets no such column.
+ withTempDir { dir =>
+ val name =
FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME.toUpperCase(Locale.ROOT)
+ val path = writeSingleParquetFile(dir,
+ spark.range(0, 50).selectExpr("id AS k", "CAST(id AS STRING) AS v",
s"true AS $name"),
+ rowGroupSize = 64 * 1024L)
+ withSQLConf(SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key ->
"true") {
+ val relation = spark.read.parquet(path).queryExecution.optimizedPlan
+ val k = relation.output.find(_.name == "k").getOrElse(fail("no k in
the relation output"))
+ val logical = LogicalFilter(
+ BloomFilterMightContain(bloomLiteralOf(42L), new XxHash64(Seq(k))),
relation)
+ val physical = FileSourceStrategy(logical).headOption
+ .getOrElse(fail(s"FileSourceStrategy did not plan $logical"))
+ val scan = scanOf(physical)
+ assert(scan.storageFilters.isEmpty, s"nothing must be offered; got
${scan.storageFilters}")
+ assert(!scan.requiredSchema.fieldNames.contains(
+ FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME),
+ s"nor a checked column added; got ${scan.requiredSchema}")
+ }
+ }
+ }
+
+ test("FileSourceStrategy leaves the checked column out where it alone
crosses the field limit") {
+ // One more output column past `spark.sql.codegen.maxFields` would turn
the scan's columnar
+ // output off. So the scan still gets the filter but no checked column,
and the Filter evaluates
+ // the bloom on every row. A scan already past the limit reads rows
anyway, and gets the column.
+ // The limit counts the whole output, so a partition column counts too.
+ withTempDir { dir =>
+ val path = new File(dir, "parted").getAbsolutePath
+ spark.range(0, 50).selectExpr("id AS k", "CAST(id AS STRING) AS v", "id
% 3 AS p")
+ .repartition(1).write.partitionBy("p").parquet(path)
+ withSQLConf(SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key ->
"true") {
+ val relation = spark.read.parquet(path).queryExecution.optimizedPlan
+ val k = relation.output.find(_.name == "k").getOrElse(fail("no k in
the relation output"))
+ val logical = LogicalFilter(
+ BloomFilterMightContain(bloomLiteralOf(42L), new XxHash64(Seq(k))),
relation)
+ // `k`, `v` and the partition column `p` are 3 fields, and 4 with the
checked column.
+ Seq(4 -> true, 3 -> false, 2 -> true).foreach { case (maxFields,
withColumn) =>
+ withSQLConf(SQLConf.WHOLESTAGE_MAX_NUM_FIELDS.key ->
maxFields.toString) {
+ val scan = scanOf(FileSourceStrategy(logical).headOption
+ .getOrElse(fail(s"FileSourceStrategy did not plan $logical")))
+ assert(scan.storageFilters.size == 1,
+ s"a limit of $maxFields must still offer the bloom; got
${scan.storageFilters}")
+ assert(scan.requiredSchema.fieldNames.contains(
+ FileFormat.STORAGE_FILTER_CHECKED_COLUMN_NAME) == withColumn,
+ s"a limit of $maxFields must ${if (withColumn) "add" else "leave
out"} the checked " +
+ s"column; got ${scan.requiredSchema}")
+ assert(scan.supportsColumnar == (maxFields >= 3),
+ s"the checked column must not cost a limit of $maxFields the
scan's columnar " +
+ s"output; got ${scan.output}")
+ }
+ }
+ }
+ }
+ }
+
+ Seq(false, true).foreach { withMetadata =>
+ test("the post-scan Filter still filters the rows the reader gave up on " +
+ s"(_metadata selected = $withMetadata)") {
+ // A runtime bloom is redundant with its join, which would drop such
rows anyway, so this
+ // filters on a bloom alone, against a plain read. The scan's metrics
show whether the filter
+ // was applied. A cap of 39 bytes gives the filter up in every row group
with a survivor, and
+ // the post-scan Filter must then filter those rows itself. So must it
when parquet-mr reads,
+ // which does not know the checked column and reads it as null. A
partition column puts the
+ // checked column between columns of the scan's output, and with
`_metadata` selected, the
+ // Project that builds it sits below the Filter and must keep the
checked column.
+ withTempDir { dir =>
+ val path = new File(dir, "parted").getAbsolutePath
+ rows400.toDF("k", "v").withColumn("p", col("k") % 3)
+ .repartition(1).write.partitionBy("p")
+ .option(ParquetOutputFormat.BLOCK_SIZE, 256L).parquet(path)
+ val columns = Seq(col("k"), col("v"), col("p")) ++ (if (withMetadata) {
+ Seq(col("_metadata.row_index").as("ri"),
col("_metadata.file_name").as("file"))
+ } else {
+ Nil
+ })
+ val relation = spark.read.parquet(path).select(columns:
_*).queryExecution.analyzed
+ val k = relation.output.find(_.name == "k").getOrElse(fail("no k in
the relation output"))
+ val bloom = BloomFilterMightContain(bloomLiteralOf(42L), new
XxHash64(Seq(k)))
+ // Plans under `planning` and reads under `reading`, and returns the
plan with its rows.
+ def run(
+ enabled: Boolean,
+ planning: Map[String, String] = Map.empty,
+ reading: Map[String, String] = Map.empty): (SparkPlan, Seq[Row]) =
{
+ val pushdown = SQLConf.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED.key
-> enabled.toString
+ val df = withSQLConf((planning + pushdown).toSeq: _*) {
+ val df = classic.Dataset.ofRows(spark, LogicalFilter(bloom,
relation))
+ df.queryExecution.executedPlan
+ df
+ }
+ val rows = withSQLConf(reading.toSeq:
_*)(df.collect().toSeq.sortBy(_.getLong(0)))
+ (df.queryExecution.executedPlan, rows)
+ }
+ def excludedWithinRowGroups(plan: SparkPlan): Long = collect(plan) {
Review Comment:
Done in
[9a71742c400](https://github.com/apache/spark/pull/59269/commits/9a71742c400a9104bb4595ac582a449a6e6fede4).
`rowsExcludedByStorageFilters` takes the metric keys. `columnOf`, `bloomOn`
and `planned` cover the lookup, the bloom and the planning, and the fixture
uses `writeKeyParquetFile`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -203,12 +208,24 @@ object ParquetStorageFilter {
s"${ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME}, because the
reader writes row " +
"indexes over that column, so what it reads back depends on how its
row group was read")
+ // The column the reader marks the rows it checked in, which the planner
adds to every scan it
+ // offers storage filters to. A schema without it, as a test can build,
gets no marks.
+ val checkedColumnIndex =
Review Comment:
Done in
[9a71742c400](https://github.com/apache/spark/pull/59269/commits/9a71742c400a9104bb4595ac582a449a6e6fede4).
The reader finds the slot by name in `sparkRequestedSchema`, through
`ParquetStorageFilter.checkedColumnIndex`, which also checks the type. The
constructor field and both `require`s are gone.
--
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]
