[FLINK-39421] Metadata filter push-down for table sources

[jnh5y]

Add dedicated metadata filter push-down path through SupportsReadingMetadata. Metadata predicates are classified separately from physical predicates and pushed via applyMetadataFilters() with a dedicated MetadataFilterResult type.

MetadataFilterPushDownSpec stores a pre-translated predicateRowType using metadata key names (not SQL aliases), making the serialized spec self-contained without needing a column-to-key mapping.

Generated-by: Claude Code (claude-opus-4-6)

What is the purpose of the change

Predicates on metadata columns (e.g., Kafka offset, timestamp, partition) cannot be pushed through the existing SupportsFilterPushDown path because FilterPushDownSpec's serialized RexInputRef indices break during compiled plan restoration when ProjectPushDownSpec narrows the row type. This PR adds a dedicated metadata filter push-down path that solves the serialization problem and enables metadata-aware source optimizations. See FLIP-574 for the full design.

Brief change log

• Extended SupportsReadingMetadata with supportsMetadataFilterPushDown() and applyMetadataFilters() default methods and a MetadataFilterResult type
• Added MetadataFilterPushDownSpec that snapshots the predicate row type at optimization time, using metadata key names (not SQL aliases)
• Extended PushFilterIntoTableSourceScanRule to classify predicates as physical, metadata, or mixed, with a two-path push-down flow
• Added alias-to-key translation in the planner so sources receive predicates in metadata key space

Verifying this change

This change added tests and can be verified as follows:

• Added MetadataFilterInReadingMetadataTest with 6 planner rule tests: basic push-down, opt-out when unsupported, aliased metadata keys, mixed physical+metadata separation, partial acceptance, and interaction with projection push-down
• Added serde roundtrip test in DynamicTableSourceSpecSerdeTest for MetadataFilterPushDownSpec
• Existing physical filter push-down tests continue to pass unchanged

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): yes — new default methods on @PublicEvolving interface SupportsReadingMetadata
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? JavaDocs

[flinkbot]

CI report:

• 2b6f73d Azure: FAILURE
• f96af04 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build