[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 positions 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 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.

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;
• 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.
• 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

• NestedFieldTransformTest (9 new) — one and two level reads; a null anywhere on the path yields null; a predicate on null evaluates false; no FieldRef is exposed; stats never prune; projection keeps the path; JSON round trip; a path through a non-row type is rejected.
• ParquetFiltersTest (4 new) — nested field pushdown; every pushable function on a nested field, plus AND-mixing with a top-level column and startsWith rejection; a nested field under a repeated group is not pushed down; a nested field missing from the file.
• SparkV2FilterConverterTestBase (2 new) — end to end on info.uid and info.addr.city, asserting the predicate reaches pushedDataFilters, that fieldRefOptional() is empty, and that fieldNames() reports the enclosing top-level column; plus a top-level column whose name contains a dot, asserting it resolves to a FieldTransform rather than being split into a path.

Run against Spark 3.3, 3.4 and 3.5 (32/32 each). PaimonPushDownTest is regression-clean at 21/21.

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.

[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.

[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.