feat(promql): run a bound tree over raw series in one fused node

`FusedSeriesExec` evaluates a compiled `NativeTree` directly over the raw
series its child produces, instead of materialising a row batch through the
per-operator chain. It binds the time column, the fields and the tags once,
checks the input against that binding when the plan is built, and converts the
chunks the tree settles back into the statement's output columns.

The node's partition count is the child's, one tree per partition. The child
must be a raw-series scan that assigns every series whole to one partition:
`PerSeries` distribution is what makes a per-partition evaluation equivalent to
evaluating the whole series. A zero-partition input is refused; anything else
is the child's business.

A label set can recur across partitions once `rate` and the map kernels drop
`__name__`, so the occupancy map is plan state rather than per-stream. Two
series reaching one label set is an error only when their occupied grid points
overlap, which is Prometheus' range-query rule; disjoint points are legal and
are merged into one occupancy record. The reservation for that map is attached
to the first execution's pool and released when the plan is dropped, because a
later partition still has to collide with what an earlier one emitted.

`occupied_runs` reads a chunk's absolute positions rather than assuming one
contiguous run: a sparse evaluated chunk may hold several, and treating the
first as the whole range would count the holes as rows.

A poll does a bounded amount of work and wakes itself when the budget runs out,
so a long run of chunks that settle nothing cannot monopolise the executor
thread.

Signed-off-by: discord9 <discord9@163.com>

src/promql/src/extension_plan.rs

@@ -14,6 +14,7 @@
 
 mod absent;
 mod empty_metric;
+mod fused_series;
 mod histogram_fold;
 mod instant_manipulate;
 mod normalize;
@@ -37,6 +38,7 @@ use datafusion::error::{DataFusionError, Result as DataFusionResult};
 use datafusion::logical_expr::{Expr, Extension, LogicalPlan};
 use datatypes::data_type::DataType as _;
 pub use empty_metric::{EmptyMetric, EmptyMetricExec, EmptyMetricStream, build_special_time_expr};
+pub use fused_series::{FusedSeries, FusedSeriesBinding, FusedSeriesExec, FusedSeriesStream};
 pub use histogram_fold::{
     HistogramFold, HistogramFoldExec, HistogramFoldOperation, HistogramFoldStream,
 };