feat: push scalar subquery filters into scans

[discord9]

Which issue does this PR close?

• N/A.

Rationale for this change

An uncorrelated scalar subquery is executed and published before DataFusion starts the main input. A physical predicate such as:

ts >= (SELECT ...) - INTERVAL '1 minute'

can therefore be resolved before scan execution. Snapshotting ScalarSubqueryExpr enables local pruning, but runtime pushdown through execution-plan boundaries requires a DynamicFilterPhysicalExpr producer/consumer relationship.

Parquet also retains supported predicates for statistics pruning while reporting PushedDown::No when row-filter pushdown is disabled. Scalar-subquery bindings must therefore be retained based on the rewritten input actually containing the dynamic-filter expression ID, rather than only the pushdown discriminant.

What changes are included in this PR?

• Implement PhysicalExpr::snapshot() for ScalarSubqueryExpr.
  • Completed non-null and typed-null results snapshot to typed literals.
  • Pending and reset results report that the scalar value is unavailable.
• During Post physical filter pushdown, make ScalarSubqueryExec:
  • discover matching Column >= ScalarSubqueryExpr - Literal predicates scoped to its own ScalarSubqueryResults;
  • create initially-true dynamic filters and push them only into the main input;
  • retain bindings when the rewritten input contains the dynamic-filter expression ID, including Parquet statistics-only pushdown;
  • after all scalar results are published, snapshot the original predicate and update/complete each independent dynamic-filter runtime state before executing the main input.
• Preserve the original predicate as the authoritative query condition.
• Add built-in protobuf support for scalar-subquery dynamic-filter bindings.
  • The wire change is additive: ScalarSubqueryExecNode gains repeated binding records containing the producer predicate and consumer expression ID.
  • Default-codec decoding restores all consumer instances with that ID. Independent decoded states are all updated, while remapped consumers sharing runtime state are updated once.
• Add regression coverage for snapshotting, timestamp/interval pruning, pushdown discovery and retention, execution/error/reset semantics, default protobuf round trips with multiple consumers, and real Parquet statistics pruning with row-filter pushdown disabled.

This keeps the existing scalar-subquery execution order: all scalar subqueries finish before the main input starts. It does not rewrite the query to a nested-loop join, execute streams concurrently, or add a new datasource protocol.

Are these changes tested?

Yes. The following focused checks pass:

cargo fmt --all -- --check
git diff --check
cargo check -p datafusion-physical-plan -p datafusion-proto --features proto
cargo test -p datafusion-physical-expr --lib scalar_subquery
cargo test -p datafusion-physical-expr --lib dynamic_filters
cargo test -p datafusion-pruning scalar_subquery_timestamp_snapshot_builds_pruning_predicate
cargo test -p datafusion-physical-plan --lib scalar_subquery --features proto
cargo test -p datafusion-proto --test proto_integration --features parquet -- scalar_subquery
cargo test -p datafusion --test core_integration \
  physical_optimizer::filter_pushdown::scalar_subquery_dynamic_filter_parquet_pruning_with_pushdown_disabled \
  --features parquet -- --nocapture

Are there any user-facing changes?

Queries with eligible uncorrelated scalar-subquery predicates can now resolve the scalar value before scan execution and deliver the resulting dynamic predicate to scan pruning. This enables runtime file/row-group pruning and downstream consumers such as GreptimeDB MergeScan without changing SQL semantics or scalar-subquery execution order.