feat: reshuffle block metadata after pruning

[SkyFan2002]

I hereby agree to the terms of the CLA available at: https://docs.databend.com/dev/policies/cla/

Summary

This PR improves read parallelism for distributed Fuse scans by reshuffling the surviving block metadata after pruning.

Previously, lazy Fuse segments were distributed before pruning. Although the segments could be evenly assigned across executors, pruning might leave very different numbers of blocks on each node. The surviving block reads could therefore remain concentrated on one or a few executors, causing scan skew and underutilizing the rest of the cluster.

This PR separates distributed Fuse scanning into the following stages:

FusePrune
    -> block metadata exchange
    -> FuseBlockRead

1. FusePrune runs the existing pruning pipeline and emits empty DataBlocks carrying BlockPartitionMeta.
2. A dedicated Flight exchange hashes each surviving FuseBlockPartInfo and redistributes the metadata across executors.
3. FuseBlockRead expands the exchanged metadata back to the configured I/O parallelism and reads the actual block data on the destination nodes.

The exchange uses a stable bucket order based on each node's persistent cache_id, rather than the transient exchange destination order. This keeps block ownership stable when exchange destination ordering changes and avoids unnecessarily reducing persistent-cache locality.

Activation conditions

The metadata exchange is enabled only when all of the following conditions are met:

• enable_distributed_pruning = 1
• enable_prune_pipeline = 1
• The query runs on a multi-node cluster
• The source table uses the Fuse engine
• The scan contains lazy-level partitions

The physical-plan rewrite is applied only when the scan belongs to a fragment that can host readers on all executors. A scan in the root fragment may be coordinator-only, so it keeps the original TableScan path.

Scans that do not satisfy these conditions also continue to use the existing path. Disabling either distributed pruning or the prune pipeline therefore provides a fallback without the new metadata exchange.

Implementation details

• Records eligible Fuse scans while building TableScan, then injects the metadata exchange after the row-data exchanges have been finalized.
• Introduces two physical operators:
  • FusePrune, a source operator that emits pruned block partitions without reading block data.
  • FuseBlockRead, which consumes exchanged block partitions and performs the existing Fuse block-read and scan-output processing.
• Treats FusePrune as a source fragment and propagates its DataSourcePlan through fragment serialization.
• Adds a dedicated FusePartExchangeInjector that:
  • Accepts only node-to-node exchanges.
  • Validates that exchanged blocks contain BlockPartitionMeta.
  • Validates that every exchanged partition is a FuseBlockPartInfo.
  • Routes every partition to exactly one destination.
  • Uses stable cache_id ordering when mapping hash buckets to exchange outputs.
• Makes BlockPartitionMeta serializable across Flight while retaining the complete FuseBlockPartInfo, including block indexes, statistics, bloom-filter locations, ranges, and virtual-block metadata.
• Splits the Fuse read pipeline so pruning output and block reading can run on opposite sides of the metadata exchange.
• Preserves scan projection, internal-column handling, runtime scan filters, limit optimizations, pruning statistics, and scan profile metrics.
• Adapts index refresh and MERGE target-table optimization to recognize the new physical operators.

Reviewer focus

Please pay particular attention to the following areas:

1. Physical-plan placement

   • src/query/service/src/physical_plans/physical_distributed_pruning.rs
   • Verify that the rewrite is applied only where the reader fragment runs on all executors and every exchange destination has a receiver.
   • In particular, please check the handling of ordinary exchange inputs, materialized CTE inputs, broadcast sink inputs, and coordinator-only root scans.

2. Partition routing and cache stability

   • src/query/service/src/servers/flight/v1/exchange/fuse_part_exchange.rs
   • Verify that every surviving block partition is delivered exactly once.
   • Please also check whether sorting destinations by persistent cache_id gives the expected stable ownership and cache-locality behavior when exchange destination order changes.

3. Exchange and serialization correctness

   • Verify that BlockPartitionMeta and the full FuseBlockPartInfo survive the Flight serialization round trip.
   • Check the behavior for empty pruning results, empty destinations, missing metadata, unexpected partition types, receiver errors, and closed channels.

4. Pipeline construction and parallelism

   • src/query/storages/fuse/src/operations/read_data.rs
   • src/query/storages/fuse/src/operations/read/fuse_source.rs
   • Verify that metadata is emitted with bounded batching and that block reads are expanded back to the configured I/O parallelism after the exchange.
   • Please also check that replacing and restoring the exchange injector cannot affect surrounding exchanges.

5. Compatibility with existing scan behavior

   • Check that index refresh, MERGE target-table optimizations, runtime scan filters, limit handling, internal columns, projection, EXPLAIN output, pruning statistics, and scan profile metrics retain their previous semantics.

6. Performance trade-off

   • The change adds a metadata exchange before block reads. Please assess whether the additional scheduling, serialization, and network overhead is appropriate for the expected reduction in post-pruning scan skew.
   • This PR currently has unit and cluster SQL coverage but does not include benchmark results.

Test coverage

The added coverage includes:

• Physical-plan rewrite placement for local and distributed scans.
• Stable cache_id bucket mapping independent of exchange destination order.
• Exactly-once partition routing across exchange outputs.
• Rejection of missing metadata and non-block partitions.
• Full BlockPartitionMeta serialization round trip.
• Receiver batching, error forwarding, and closed-channel handling.
• Cluster SQL coverage for:
  • Filtered scans.
  • Empty pruning results.
  • Aggregation and limit queries.
  • Distributed-pruning fallback.
  • Prune-pipeline fallback.
  • MERGE behavior with intentionally skewed segment layouts.
• EXPLAIN ANALYZE and scan-profile compatibility.

Tests

• Unit Test
• Logic Test
• Benchmark Test
• No Test - Explain why

Validation:

• cargo check -p databend-query --lib
• cargo test -p databend-query --lib create_builds_prune_exchange_read_shape -- --nocapture
• cargo test -p databend-query --lib fuse_part_exchange -- --nocapture
• cargo clippy -p databend-query --lib -- -D warnings
• cargo fmt --all -- --check

Type of change

• Bug Fix (non-breaking change which fixes an issue)
• New Feature (non-breaking change which adds functionality)
• Breaking Change (fix or feature that could cause existing functionality not to work as expected)
• Documentation Update
• Refactoring
• Performance Improvement
• Other (please describe):

AI assistance

• AI usage: An AI coding agent assisted with the distributed Fuse pruning, block metadata exchange implementation, and tests; I reviewed the final diff
• Responsible human: @SkyFan2002
• The responsible human has read every line of this diff and can explain each change

---

This change is