dongjoon-hyun commented on code in PR #59269:
URL: https://github.com/apache/spark/pull/59269#discussion_r4210240482
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -595,6 +634,8 @@ protected boolean loadNextRowGroup() throws IOException {
m.rowsExcludedWithinRowGroup().add(baselineRows - finalRowCount);
}
+ // The rows phase 2 read are exactly the survivors unless the filter was
given up.
+ rowsChecked = !filterGivenUp;
Review Comment:
**A row group that gave up only splicing is marked checked, but its key
values then come from a second read through the offset index, not from the
values the filter evaluated.** Splicing alone is given up when the survivor
buffer passes `maxSplicedRowGroupBytes` while the ranges still fit (L857-864),
or when buffering a survivor throws (L843-855). `abandonSplicing()` clears
`spliceCurrentRowGroup`, so phase 2 reads `requestedColumns`, key columns
included, over the survivor ranges (L592, L604, L641), and this line still
marks every row true. If the offset index of a key column is wrong, or a
row-range bug shifts the rows, phase 2 emits rows whose re-read keys the bloom
rejects, and `Or(checked, bloom)` keeps them. Before this PR the post-scan
Filter dropped them, and only a key/value mispairing could get through. That is
the new caveat in the conf doc and at L189-192. It is also why
`FileFormat.scala:184` ("evaluated on the row's own values") and
`FileSourceStrategy.scala:161` ("t
he plan keeps the exact check") no longer hold for this case.
Could this be `rowsChecked = spliceCurrentRowGroup;` instead?
- A spliced row group emits exactly the key values the predicate evaluated.
L811 evaluates the scratch row, L842 copies the same values into the
accumulators, and `nextBatch` swaps them into the key slots (L401, L414). So
skipping the bloom there cannot change any row's outcome, even with a wrong
index.
- `spliceCurrentRowGroup` is false after every `abandonSplicing()` (L579,
L616, L853, L860) and whenever the row group cannot be narrowed (L563), and
true implies that the filter was kept.
- The cost is one bloom evaluation per survivor, only in row groups that
already re-read and re-decode their key columns.
The "or return rows the filter rejects" caveat could then go instead of
being documented, and the expectation at `ParquetStorageFilterSuite.scala:2006`
would flip to false. @sunchao listed this as an accepted behavior change; this
would remove it instead.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -194,6 +200,10 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
outputDataSchema: StructType): Seq[Expression] = {
val sparkSession = fsRelation.sparkSession
if (!fsRelation.fileFormat.supportsStorageFilterPushdown(sparkSession))
return Nil
+ if (fsRelation.schema.fieldNames.exists(
Review Comment:
**A file that physically holds a column with this name is not covered by the
guard, and the late-materialization reader then reads that column.** The check
looks at the relation schema only, so the column can still be in the files,
e.g. under `spark.read.schema("k BIGINT, v STRING")`, a catalog schema, or a
schema inferred from another file with `mergeSchema` off. Then:
- `ParquetReadSupport.clipParquetGroupFields` matches the planner's field to
the file's column by name, case-insensitively by default, and keeps the file's
physical type (`ParquetReadSupport.scala:460-475` and `:262-265`). So
`checkColumn` does not treat the column as missing.
- `suppliesOwnVector`
(`LateMaterializationParquetRecordReader.java:652-659`) answers true only for
key slots while splicing, so `installRowGroup` builds a column reader for the
checked slot. The column is also in `nonKey` (L479-483), which phase 2 reads
(L592). The plain path for a declined file reads it the same way.
- If the column is not BOOLEAN, e.g. INT64,
`ParquetVectorUpdaterFactory.getUpdater` throws
`SchemaColumnConvertNotSupportedException`, and the task fails with
`FAILED_READ_FILE.PARQUET_COLUMN_DATA_TYPE_MISMATCH`. The same query succeeds
today, since the base never requests this column.
- If it is BOOLEAN, the chunk is read and decoded into super's vector for
nothing. A file that has none of the projected non-key columns is also no
longer declined, because this column now counts as one.
So the reader javadoc's "the file's values never reach it"
(`LateMaterializationParquetRecordReader.java:106`) holds for the emitted
batch, but not for the read. Could the reader answer `suppliesOwnVector` true
for the checked slot (a primitive boolean, as that hook requires) and leave its
path out of `nonKey`, so the file's column is never decoded? @sunchao's comment
on the conf doc covers the parquet-mr side of the same name collision.
##########
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 =>
Review Comment:
**Scans that read rows anyway keep the column, and the top `ProjectExec` it
forces copies every surviving row.** This rule mirrors only the
`isTooManyFields` term of `FileSourceScanExec.supportsColumnar`
(`DataSourceScanExec.scala:757-764`), not `wholeStageEnabled`. So a scan
already past `spark.sql.codegen.maxFields`, or one planned with whole-stage
codegen off, keeps the column. The new tests assert both (the `2 -> true` case,
and the e2e row arm). Then:
- `FilterExec` outputs one more attribute, so `projects ==
withFilter.output` at L460 never holds. `RemoveRedundantProjects` cannot drop
the `ProjectExec` either, since it compares exprIds
(`RemoveRedundantProjects.scala:111` and `:116`). Before this PR that Project
was never built, or it was removed, because `checkNullability` (L98 there)
tolerates the `isnotnull(k)` tightening.
- Neither node is whole-stage codegen'd there
(`WholeStageCodegenExec.scala:929-933`, or `:989` with codegen off), so
`ProjectEvaluatorFactory` runs an `UnsafeProjection` per surviving row (L36-38
there). That is a second full-row copy after the scan's own, to save one
`xxhash64` plus the bloom probes on checked rows, and nothing on unchecked ones.
Example: `SELECT f.*, d.name FROM fact f JOIN dim d ON f.k = d.k WHERE
d.region = 'X'` with a runtime bloom on an unpartitioned 150-column `fact`. In
the other direction, with whole-stage codegen off and the output at exactly
`maxFields`, the column is dropped although no columnar output is at stake.
Could the column be left out of scans that read rows anyway, at least the
ones already past `maxFields`, which are wide by definition? For example, the
decision could be `conf.wholeStageEnabled && !isTooManyFields(conf, (output :+
checked).toStructType)`.
##########
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) {
Review Comment:
**Where no row can be marked true, the column and the `Or` only add work.**
`rowsChecked` stays false or null for every row when:
- a row group has no offset index. `canNarrowRowGroup` is then false
(`LateMaterializationParquetRecordReader.java:561-563`), so phase 1 gives up at
the first survivor (L830-835). That covers files written by pyarrow's
`write_table` with its defaults, or by parquet-mr before 1.11, as the comment
at L555-557 there notes;
- `parquet.filter.columnindex.enabled` is false;
- the reader declines the file, or parquet-mr reads the scan.
Each row then pays the nullable-boolean load and the `Or` branches before
the same bloom evaluation as before, plus a fill per batch. That is a couple of
ns next to tens for the bloom, so it is small, but it buys nothing there.
@sunchao counted the per-conjunct branch as a non-blocking cost; the point here
is the scans where it can never pay off. The `columnindex` case is known at
planning time from the scan's Hadoop conf, so the planner could skip the column
there. Could the conf doc also say that files without a page index get no
benefit from the skip?
##########
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:
**The contract is now restated in about nine places, and some copies
contradict the field-limit case.**
- Here: "So the `Filter` evaluates a conjunct only on the rows the reader
did not check." At exactly `spark.sql.codegen.maxFields`, the scan still gets
the filter but no column (L371-379), so the `Filter` evaluates it on every row.
The new test asserts that (`ParquetStorageFilterSuite.scala:2249-2272`), and
L367-370 and `FileFormat.scala:191-193` say so.
- `ParquetStorageFilter.scala:211-212` says the planner adds the column "to
every scan it offers storage filters to", and that a schema without it is one
"a test can build". The same case gives `create` a production schema without it
(`checkedColumnIndex == -1`), so tightening that into a `require` would fail
those queries. `create`'s doc (L172-174 there) also credits `storageFiltersFor`
with pre-checking every `require`, but the new `BooleanType` one holds only
because `apply` builds the attribute as `BooleanType`.
- The doc of the public `spark.sql.parquet.storageFilterPushdown.enabled`
(`SQLConf.scala:1985-1987`) now names `spark.sql.codegen.maxFields`, which is
`.internal()` and so absent from the published configuration page.
Could the contract live in `FileFormat.buildReaderWithStorageFilters` only,
with the other places pointing to it, and the conf doc describe the field-limit
case without naming an internal conf?
##########
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:
**The rewrite is asserted only for a literal bloom, and the marks only
through `readChecked` and one columnar read, so the paths the feature runs on
in practice go unchecked.**
- This test uses `bloomLiteralOf(42L)`. A real `InjectRuntimeFilter` bloom
carries a `ScalarSubquery`, and the tests that plan one (around L1031, L1936
and L3235-3364) assert only `countBloomFiltersInPostScanFilters(plan) >= 1` and
join results. The former also counts a bloom inside the `Or`, and a wrong mark
cannot change the latter. If the `ExpressionSet` match stopped matching such a
conjunct, the feature would quietly do nothing and the suite would still pass.
- The row-mode late-materialization reader (a scan past `maxFields`, or
whole-stage codegen off, with the vectorized reader on) is never checked for
marks. The e2e row arm turns the vectorized reader off, so parquet-mr reads
that scan.
- `withStorageFilters` (L876-882) adds storage filters but no checked
column, so the 18 end-to-end reader tests that use it run without marks.
- `readChecked` reads row groups of 20 or 100 rows with capacity 4096, so
marks across the batches of one row group are not covered.
Could the e2e test also run a real runtime-bloom join, assert the rewritten
conjunct, and read the marks off the scan, both columnar and in row mode with
the vectorized reader on?
##########
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:
**The scan's output groups are now listed by hand in three places, and this
check reads as a double negative.** This list repeats `outputAttributes`
(L389-392) without the column, and the `_metadata` Project (L443-444) lists the
groups once more. If a later scan-internal column is added to
`outputDataColumns` only, this check undercounts by one. The checked column can
then push the scan past `maxFields` and cost it its columnar output, which is
the case this check exists to prevent. Leaving it out of the `_metadata`
Project breaks the `Filter`'s binding. The `storageFiltersFor` parameter
`outputDataSchema` (L200, without the column) also now means something
different from the local of the same name (L391).
Could this be one shared base list and a named boolean, e.g.
`Option.when(storageFilters.nonEmpty && !checkedAloneCrossesLimit)(...)`, with
the `_metadata` Project built from `scan.output` minus the flattened metadata
attributes?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -418,11 +441,19 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
Alias(KnownNotNull(CreateStruct(structColumns.toImmutableArraySeq)),
FileFormat.METADATA_NAME)(exprId = metadataStruct.exprId)
execution.ProjectExec(
- readDataColumns ++ partitionColumns :+ metadataAlias, scan)
+ readDataColumns ++ storageFilterCheckedColumn ++ partitionColumns :+
metadataAlias, scan)
}.getOrElse(scan)
+ // The offered conjuncts are skipped on the rows the reader marks as
checked. They are
+ // matched by `ExpressionSet`, since `storageFilters` carry the
relation's column names.
+ val postScanFilters = storageFilterCheckedColumn match {
+ case Some(checked) =>
+ val offered = ExpressionSet(storageFilters)
+ afterScanFilters.toSeq.map(f => if (offered.contains(f)) Or(checked,
f) else f)
Review Comment:
**With two offered blooms on one key, whole-stage `FilterExec` CSE computes
the shared `xxhash64` before the first `Or`, so checked rows still pay the
hash.** With `spark.sql.subexpressionElimination.skipForShortcutExpr` at its
default, false, `EquivalentExpressions` recurses into both children of an `Or`
(`EquivalentExpressions.scala:160-173`). So an `xxhash64(k)` under two
`Or(checked, might_contain(...))` conjuncts is a common subexpression, and
`FilterExec.doConsume` emits its state before the first conjunct that
references it (`basicPhysicalOperators.scala:347-371`), ahead of the
short-circuit. Two blooms on one key are possible because
`InjectRuntimeFilter.hasBloomFilter` (L399-409 there) compares keys with
`fastEquals`, which includes the qualifier, e.g. `SELECT * FROM (SELECT f.*
FROM fact f JOIN d1 ON f.k = d1.k WHERE d1.x = 1) t JOIN d2 ON t.k = d2.k WHERE
d2.y = 2`. Results are not affected; only the probes are skipped.
This is narrow, but `If(checked, Literal(true), f)` would avoid it. `If` is
a `ConditionalExpression`, so CSE only looks at its predicate and at what both
branches share (`conditionalExpressions.scala:66-72`). It also filters a null
mark exactly as false, which is what @sunchao's P2 on this line asks for.
##########
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:
**This is `rowsExcludedByStorageFilters` (L969-974) restricted to one
metric, and the new planner tests repeat the same setup.** The local copy is
needed only because the shared helper also sums `ROWS_EXCLUDED_BY_ROW_GROUP`,
and this test now sums the metrics both ways (L2331 and L2349, against L2360).
In the new tests:
- the attribute lookup `relation.output.find(_.name ==
...).getOrElse(fail(...))` appears at L2198, L2233, L2258 and L2302;
- the bloom `BloomFilterMightContain(bloomLiteralOf(42L), new
XxHash64(Seq(...)))` appears at L2199, L2235, L2260 and L2303;
- `FileSourceStrategy(logical).headOption.getOrElse(fail(...))` appears at
L2203, L2236 and L2264, on top of the existing copies;
- the fixture at L2192-2194 is what `writeKeyParquetFile(dir, 0L until 50L,
rowGroupSize = 64 * 1024L)` writes.
Could `rowsExcludedByStorageFilters` take the metric keys, and small helpers
cover the lookup, the bloom and the planning?
##########
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:
**`readChecked` and `hadoopSplit` add a third way to build the reader, while
`formatReader` with `drain` already covers what these tests need.** They go
through the Hadoop `initialize` and the test-only constructor, with a conf
holding 7 of the 13 keys that `ParquetFileFormat.setupHadoopConf` sets, and
`readAllWith`'s doc now carves out an exception for them (L2408-2410).
`formatReader` (L173) builds the production reader through
`buildReaderWithStorageFilters` with any required schema, so it can project the
missing checked column too. This test already reads the same file and schema
through it, via `readUnderCap` (L2009-2010). The cap comes from the session
conf, and off-heap from `COLUMN_VECTOR_OFFHEAP_ENABLED` plus `asTask` (as at
L805 and L1795). `OnKey` and the damaged-index retry with `withLogged` already
run through it too (L791, L1749).
As it is, the marks are read under a conf that differs from production (e.g.
`TIMESTAMP_NANOS_TYPES_ENABLED`, the session time zone, the rebase modes), and
the assertion at L2006 is backed by bytes from a different reader setup
(L2008-2012). Could `readChecked` be built on `formatReader` and `drain`?
##########
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:
**Could the reader look this slot up itself, as it does the row-index
column?** `checkedColumnIndex` is computed here and carried on the filter only
for `takeOverBatch` (`LateMaterializationParquetRecordReader.java:344`), which
already holds the same schema: `sparkRequestedSchema` comes from
`SPARK_ROW_REQUESTED_SCHEMA`, which is this same `requiredSchema`.
`VectorizedParquetRecordReader` finds the row-index column with
`ParquetRowIndexUtil.createGeneratorIfNeeded(sparkSchema)`, whose
`findRowIndexColumnIndexInSchema` checks the type too. A lookup in the reader
would drop the constructor field, the two `require`s here, and the test block
at `ParquetStorageFilterSuite.scala:545-557`. Neither `require` can fail on a
planner-built plan, since the attribute is always `BooleanType` and storage
filters reference only read data columns. It would also remove a second source
of truth. Today a filter created over `checkedSchema()` and handed to a reader
opened with the test-only `initialize
(path, columns)` throws `ArrayIndexOutOfBoundsException` at
`LateMaterializationParquetRecordReader.java:349`, or overwrites a partition
slot.
--
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]
