[core][format][spark] Support nested field predicate pushdown

[zhuxiangyi]

Purpose

Predicates on a struct's sub-field are not pushed down today. SparkExpressionConverter rejects any NamedReference with more than one part, so WHERE user.addr.city = 'Beijing' is only evaluated by the engine after every row has been read.

This PR pushes such predicates down to the parquet row group / page level.

Design. Introduce NestedFieldTransform, a Transform holding the enclosing top-level FieldRef plus the ordered field names to descend into it. It is deliberately not a FieldTransform, so LeafPredicate.fieldRefOptional() stays empty for these predicates, and every consumer that equates a leaf with a top-level column — manifest stats evaluation, file index lookup, ORC pushdown, schema evolution rewriting, partition-only predicate detection — falls into its existing give-up path unchanged. That is why the diff contains no defensive guards at those call sites.

The path is stored as component names rather than positions and is re-resolved by name whenever the transform is remapped onto a different row type (as column pruning and row-level auth do): a leaf that was pruned away fails closed instead of silently resolving onto whatever now sits at that position, and a reordered row type still finds the original field.

The parquet side resolves the dotted name against the file schema and re-dispatches through the normal function visitor via the existing visitNonFieldLeaf hook, so every pushable function works on a nested field exactly as it does on a flat one, without per-function code. Decimal and timestamp predicates carry the file's own resolved path rather than the leaf's bare name, so they address the same column a flat predicate would.

Supported. IS NULL, IS NOT NULL, =, <>, <, <=, >, >=, BETWEEN, IN, NOT IN, and AND/OR mixing a nested field with a top-level one. Any nesting depth.

Refused, falling back to engine evaluation:

• any path component under a repeated group — parquet-mr cannot filter under repetition;
• a path descending into a non-row type;
• the enclosing column's name or any nested component containing a dot — parquet-mr addresses a column by a dot-joined path that cannot express such a name, so resolving it would either miss the real column or, on an unlucky schema, address a different one;
• everything the flat path already refuses (startsWith / endsWith / contains / like).

Deliberately out of scope, each independent of this change:

• Manifest-level min/max and file index pruning. SimpleStats is a positional row over top-level columns, so a nested leaf has no slot to read from — pushdown here is parquet-only. Making those layers work on nested fields requires reorganising statistics by field id, which is a separate and much larger change.
• Schema evolution. A data file whose schema version predates the table's current schema does not get this pushdown: SchemaEvolutionUtil.devolveFilters translates a predicate by its top-level FieldRef, which a nested predicate deliberately does not expose, so such predicates are dropped for those files. Results stay correct — the engine still evaluates the filter — but the parquet-level pruning is gone until a later write or compaction rewrites the files under the current schema. Supporting it means resolving the enclosing column by field id and re-resolving the components against the file's schema; that is a separate change.
• ORC. OrcPredicateFunctionVisitor.visitNonFieldLeaf returns empty and is untouched.
• Flink. PredicateConverter does not produce nested predicates today (a nested access arrives as a GET call, not a FieldReferenceExpression), so the Flink path never constructs a NestedFieldTransform and its behaviour is unchanged.

Existing tables are unaffected. No format change, no new option, read path only. Pruning uses row group and page statistics that are already present in existing files, so no rewrite or compaction is needed. Pushdown remains an optimisation: a non-partition data filter is also kept in postScan, so Spark still evaluates it row by row and the result set cannot change.

Measured on 400k rows with a wide struct, counting real bytes read:

data layout	point =	BETWEEN, 1%	BETWEEN, 10%	absent value
clustered on the nested field	5.1%	5.1%	15.1%	0.03% (footer only)
zone-ordered	5.1%	10.1%	15.1%	0.03%
randomly distributed	100%	100%	100%	0.03%

A flat control column matched the nested column in every case. As with any min/max based pruning, the gain depends entirely on data locality.

This also adds the missing PredicateBuilder.notIn(Transform, List) overload — notIn was the only builder method without a Transform variant.

Tests

Tests reproduce every guarded scenario end to end against real row groups (ParquetFormatReadWriteTest) as well as at the filter-construction and transform level: nested predicates surviving remapping onto a pruned or reordered row type, decimal predicates across every physical type parquet can hold them in, both timestamp variants, and a nested or top-level component whose name contains a dot.

• NestedFieldTransformTest (12) — reads, null propagation, no FieldRef exposed, stats never prune, JSON round trip, and identity remapping onto a pruned or reordered row type.
• ParquetFiltersTest (46) — every pushable function on a nested field; decimal across every physical type and both timestamp variants keep the resolved path; a repeated group, a missing column, and a nested or top-level name containing a dot are all refused.
• ParquetFormatReadWriteTest (19) — end to end against real row groups: nested BIGINT/DECIMAL/TIMESTAMP/LOCAL-ZONED-TIMESTAMP predicates and dotted names keep the matching row.
• TableQueryAuthResultTest (+2) — a row filter on a nested field fails closed when its leaf is pruned away, and still finds the field when the row type is reordered.

Run against Spark 3.3, 3.4 and 3.5. PaimonPushDownTest is regression-clean.

API and Format

No change to any on-disk format, no new table option, no new configuration. NestedFieldTransform is registered as a Transform subtype so predicates serialise and deserialise like the existing ones.

Documentation

None required — no user-facing option is added; the behaviour change is that an existing query plan gains a pushed filter.

[JingsongLi]

[P1] Preserve the full nested path for DECIMAL and TIMESTAMP predicates

This re-dispatches the nested transform as a FieldRef named "payload.amount", but the decimal, timestamp, and local-zoned-timestamp visitors later build FilterApi columns from primitiveType.getName(), which is only "amount". For a predicate such as payload.amount = 12.34, parquet-mr therefore receives a missing top-level column and its statistics filter can drop every row group as all-null, producing an empty result. Please keep the resolved FileColumn.path when validating these physical types and use that full path to construct the predicate column; regression tests should cover nested DECIMAL and both timestamp variants against actual row groups.

[zhuxiangyi]

Thanks, this was exactly right and I could reproduce it.

The path resolution itself was fine — findFileColumn walks the components and returns a FileColumn carrying both the resolved path and the physical type. What I got wrong was one level up: primitiveType() discarded the path and returned only the type, so the decimal and timestamp visitors had nothing to rebuild the column from except PrimitiveType.getName(), which can only be the leaf's own name. A nested BIGINT happened to be fine because it goes through pushdownTarget, which does keep the path — which is also why my original tests missed this.

primitiveType is now fileColumn and returns the whole FileColumn; decimalColumn and timestampColumn likewise, and the three visitors build their column from column.path.

Tests against actual row groups as you asked — each writes two rows and asserts the matching row survives the filter:

• ParquetFormatReadWriteTest.testNestedDecimalPredicateKeepsMatchingRows
• ParquetFormatReadWriteTest.testNestedTimestampPredicateKeepsMatchingRows
• ParquetFormatReadWriteTest.testNestedLocalZonedTimestampPredicateKeepsMatchingRows
• ParquetFormatReadWriteTest.testNestedBigIntPredicateKeepsMatchingRows (control)

One note on how they assert, in case it looks loose: parquet filtering is row-group granular, so with both rows in one row group a matching predicate legitimately returns both. The tests assert the matching row is present rather than that exactly one row comes back, since what we need to catch is the match disappearing. Happy to tighten this if you would rather see row groups separated explicitly.

Filter-level coverage: testNestedDecimalKeepsTheFullPath, testNestedTimestampKeepsTheFullPath, testNestedLocalZonedTimestampKeepsTheFullPath, and testNestedTimestampMicrosKeepsTheFullPath for the micros literal path.

Writing those turned up two more holes in my own coverage, now closed:

• the decimal visitor builds a column per physical type and I had only exercised INT64 — testNestedDecimalKeepsTheFullPathForEveryPhysicalType covers INT32, INT64, FIXED_LEN_BYTE_ARRAY and BINARY;
• IN / NOT IN build the column through the same visitor — testNestedDecimalInAndNotInKeepTheFullPath.

[JingsongLi]

[P1] Re-resolve nested identity when inputs are remapped

This preserves an ordinal path even when the replacement FieldRef has a different nested RowType. Nested transforms are now JSON-serializable and can be used by REST row filters, so a policy on info.secret with path [0] against ROW<secret, region> can be remapped against a Spark-pruned ROW and silently evaluate info.region instead. With same-typed fields this does not fail closed and can admit unauthorized rows. Please persist stable nested names or field IDs and re-resolve them during remapping, while ensuring auth reads the full nested dependencies; alternatively, reject nested transforms in row filters until their identity can be preserved.

[zhuxiangyi]

Good catch, and I reproduced it: remapping a transform on info.secret onto a pruned ROW<region> silently produced info.region. Storing a bare position was my mistake — I had thought about the index moving but not about the row type itself changing shape.

Took the first option you offered. The path is now the ordered component names rather than positions, and copyWithNewInputs re-resolves them against the replacement row type, so a pruned-away leaf fails closed and a reordered row type still addresses the same field. Positions are derived once in the constructor and used only for evaluation.

On the "while ensuring auth reads the full nested dependencies" part — I have not done that. Keeping info.secret in the read schema when a row filter references it means touching the projection layer, and I was not sure that belonged in this PR. What is guaranteed now is that the case fails loudly instead of resolving elsewhere: the exception propagates out of TableQueryAuthResult.remapPredicate and no caller catches it (AbstractDataTableScan:134). If you would rather have the dependency actually pulled into the projection, please say so — I am glad to do it here or in a follow-up, whichever you prefer.

Tests at the auth entry point, since that is the path you were pointing at:

• TableQueryAuthResultTest.testNestedRowFilterDoesNotDriftWhenTheLeafIsPruned
• TableQueryAuthResultTest.testNestedRowFilterFollowsTheFieldWhenPositionsShift

and at the transform level, NestedFieldTransformTest.testRemapOntoAPrunedRowTypeDoesNotDrift / testRemapFollowsTheFieldWhenPositionsShift. The second one is there to keep me honest: a validation that simply throws would pass the first test but fail this one, since a reordered row type has to resolve to the original field.

In case it is useful for judging the blast radius, I also checked the other two copyWithNewInputs callers: PredicateProjectionConverter and PartitionValuePredicateVisitor both pass fieldRef.type() through unchanged and only remap the top-level index, so neither could drift. TableQueryAuthResult is the one caller that re-derives the type from a different row type.

[JingsongLi]

[P2] Preserve multipart field-name boundaries

Joining the resolved components with dots loses identifier boundaries. For a valid schema such as ROW<s ROW<"a.b" STRING>>, Spark supplies the parts [s, a.b], but this transform emits s.a.b and ParquetFilters later splits it into [s, a, b]. parquet-mr then treats the real [s, a.b] column as missing and may prune matching row groups. Please retain the ordered components and construct the Parquet ColumnPath from that array; at minimum, decline Parquet pushdown whenever a nested component contains a dot.

[zhuxiangyi]

Confirmed. In the end-to-end test the matching row is lost every time.

The ordered components are retained now, as part of the change above. For the parquet side this PR takes the "at minimum" option you offered: pushdown is declined when any component contains a dot.

Building the ColumnPath from the array does look feasible — Operators.*Column(ColumnPath) is package-private and ParquetFilters lives in that package — but it means routing components instead of a joined name through findFileColumn and the column-construction sites, which also changes the flat path. My instinct was to keep this PR to the correctness fixes and send that separately, especially since the full version would also change behaviour for flat columns whose names contain a dot. But I do not feel strongly — if you would rather have it here, I will do it.

While fixing this I ran into a related case: the dot can also sit in the top-level column's own name. ROW<"a.b" ROW<city STRING>> joins to a.b.city, which splits into [a, b, city] and misses the real ["a.b", "city"] — the same silent loss, and with an unlucky schema it could resolve to a genuinely different column rather than none. The guard rejects both spellings.

Tests:

• ParquetFiltersTest.testNestedComponentContainingADotIsNotPushedDown
• ParquetFiltersTest.testNestedFieldUnderATopLevelNameContainingADotIsNotPushedDown
• ParquetFormatReadWriteTest.testNestedComponentContainingADotKeepsMatchingRows
• ParquetFormatReadWriteTest.testTopLevelNameContainingADotKeepsMatchingRows

SparkV2FilterConverterTestBase keeps the control that a flat column named `a.b` still resolves to a FieldTransform and is unaffected by the guard.

[JingsongLi]

[P1] Reject collisions between a nested path and a dotted top-level name

Joining the components here is still ambiguous even when none of the individual components contains a dot. A valid schema can contain both s ROW<a INT> and a top-level column named s.a (the existing flat dotted-name coverage confirms the latter is supported). A predicate on the nested s.a is re-dispatched as FieldRef("s.a"); findFileColumn checks an exact top-level name before walking the split components, so it binds this predicate to the top-level s.a column instead of s -> a.

If the top-level column's row-group stats do not match while the nested column does, parquet-mr prunes the row group and Spark's residual filter never sees the matching row. Please retain the component path through column resolution/build a ColumnPath, or at minimum reject nested pushdown when the file schema has a top-level field equal to the dot-joined path. An end-to-end row-group test with both columns and opposing values should expose the false negative.