fix(query): avoid serial distribution contagion in joins

[dqhl76]

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

Summary

Fix distributed join planning when the build side already has Serial distribution.

Previously, the heuristic distributed optimizer returned Serial requirements for both join inputs as soon as either input was Serial. A small build produced by a scalar aggregate or another single-produced source therefore forced a large distributed probe through Exchange(Merge). This concentrated probe processing and join memory on one node even when broadcasting the small build was substantially cheaper.

This PR allows an eligible small Serial build to be redistributed with Broadcast while keeping the probe requirement as Any. The existing fragment scheduler already handles both resulting physical shapes:

• A fragment consuming Exchange(Merge) is marked with has_merge_input; only the coordinator executes its real input, while other Broadcast destinations retain empty receiver actions.
• A single-row source fragment is partitioned across executors, so it is produced once before Broadcast without requiring coordinator-specific placement.

No new coordinator-placement flag or scheduler production-code change is needed.

Changes

Scalar Aggregate

Before:

Join
├── Serial aggregate build (1 row)
└── Exchange(Merge)
    └── Large probe

A small Serial aggregate forced the large probe to merge onto one node.

After:

Distributed Join
├── Exchange(Broadcast)
│   └── Serial aggregate build (1 row)
└── Distributed large probe

Only the one-row aggregate result is broadcast; the large probe remains distributed.

Sort / TopN

Before:

Join
├── Serial Sort / TopN build (few rows)
└── Exchange(Merge)
    └── Large probe

A small Serial sorted result forced the large probe to merge onto one node.

After:

Distributed Join
├── Exchange(Broadcast)
│   └── Serial Sort / TopN build (few rows)
└── Distributed large probe

Tests

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

Validation:

cargo test --package databend-common-sql --test it optimizer::distributed_join::test_serial_build_distribution -- --exact
cargo test --package databend-query --test it distributed::fragmenter -- --nocapture
cargo fmt --all -- --check
git diff --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):

Risk

The behavior change is limited to the heuristic distributed optimizer when the build is Serial, the probe is not Serial, the join supports a broadcast build, and the existing broadcast cost or enforcement settings select Broadcast. Large builds and unsupported joins retain the previous Serial fallback.

Fragment scheduling behavior is unchanged. Regression tests cover the two relevant existing paths: direct Merge inputs use has_merge_input for coordinator placement, while single-row source fragments use normal source partition redistribution.

AI assistance

• AI usage: An AI coding agent assisted with root-cause analysis, drafted the optimizer change, added optimizer and fragment scheduler regression tests, and prepared this PR description
• Responsible human: @dqhl76
• The responsible human has read every line of this diff and can explain each change

---

This change is 

[dqhl76]

follow-up idea:

The existing fallback explicitly notes that redistribution could be enforced here:

databend/src/query/sql/src/planner/plans/join.rs

Lines 904 to 912 in ff30e51

	// if join/probe side is Serial or this is a non-equi join, we use Serial distribution
	if probe_physical_prop.distribution == Distribution::Serial
	|| build_physical_prop.distribution == Distribution::Serial
	|| (self.equi_conditions.is_empty() && !self.non_equi_conditions.is_empty())
	{
	// TODO(leiysky): we can enforce redistribution here
	required.distribution = Distribution::Serial;
	return Ok(required);
	}

This PR addresses the safe and common case where a small Serial build can be broadcast while keeping the probe distributed. However, when broadcast is not suitable, the remaining fallback still requires both inputs to be Serial.