[core] Cache projected schemas for data evolution stats

[leaves12138]

Purpose

DataEvolutionFileStoreScan.evolutionStats runs once per Row-ID group and repeatedly resolves the same projected file schemas and field positions. This becomes expensive for wide data-evolution tables with many groups, even when the snapshot contains only a few distinct physical column layouts.

This change adds a scan-local cache keyed by (schemaId, writeCols, valueStatsCols). Each cache entry reuses:

• the projected data-file schema;
• a physical field-ID to stats-offset map, where -1 represents a physically present field without stats;
• the projected stats field types.

The lookup map is built only on a projection-cache miss, rather than once per file or Row-ID group. evolutionStats also precomputes target field types and stops scanning older files after all fields have been resolved.

evolutionStats accepts the cache explicitly. Normal scan planning passes one cache across all Row-ID groups, while standalone tests can pass a fresh cache.

Benchmark

Replayed a real-world metadata snapshot with:

• 1,027 fields;
• 19,856 active files;
• 17,670 Row-ID groups;
• the original query predicate and a limit of 1,000.

Hot-run averages:

• master: 45.71 seconds;
• projected-schema cache before cached field lookups: 10.70 seconds;
• final implementation: 4.65 seconds.

The final implementation is 9.83x faster than master and 2.30x faster than the initial projected-schema-cache implementation. All variants produced zero candidate files.

Tests

• mvn -pl paimon-core -am -Pfast-build -DfailIfNoTests=false -DwildcardSuites=none -Dtest=DataEvolutionFileStoreScanTest test
• mvn -pl paimon-core -am -Pfast-build -DfailIfNoTests=false -DwildcardSuites=none -Dtest='*DataEvolution*Test' test (208 tests)
• Replayed the wide-table metadata snapshot twice; all runs returned the same candidate files.

[sundapeng]

Thanks for the follow-ups — the cached field-lookup map removes the O(F²) matching loop, and the projection-isolation test is a nice addition.

With those in I re-ran the numbers (1027 cols × 2000 row-id groups, single layout):

Variant	Time	vs master
master	6718 ms	1×
this PR (cache + field lookup)	120 ms	~56×
+ resolve only predicate-referenced fields	71 ms	~95×

Two remaining suggestions:

1. As a possible follow-up: resolve only the fields referenced by inputFilter instead of iterating all fields per group. With the cache and the O(1) lookups in place, what's left is the per-group full-field iteration; stats for fields the predicate doesn't reference are never read by inputFilter.test(...), so skipping their resolution is safe (unresolved fields read as null → conservatively kept). In the benchmark above that's another ~1.7×. If you add this, extract the referenced fields once per scan rather than per group — per-group extraction rebuilds the RowType name→field map and shows up in the profile.

2. A one-line comment on the HashMap cache's thread safety would help: it's only touched on the single-threaded plan() post-filter path, so it's safe, but the class calls out thread-safety requirements elsewhere, so making this explicit avoids reviewers wondering about it.

[JingsongLi]

Introduce a class for the key, so we can know what is what.

[JingsongLi]

Can be a separate class.

[JingsongLi]

Could we avoid encoding three domain states (null = field absent, -1 = field present without stats, and >= 0 = stats position) in a Map<Integer, Integer>? A typed value such as FileFieldStats.withoutStats() / withStats(index, type) would preserve the single O(1) lookup, keep the stats index and type together, and confine integer sentinels to the DataEvolutionRow / DataEvolutionArray boundary.

[JingsongLi]

+1