21663: fix: Remove nested async block causing Stacked Borrows violation in PushDecoderStreamState - #318

## Which issue does this PR close?

- Part of #318.
- Umbrella PR: apache#23738.
- Follow-up for floating-point equality keys: apache#24375.

## Rationale for this change

This is the first layer of the ASOF JOIN stack. It establishes a
broadcast-based physical execution contract independently so later
floating-point equality, logical-plan, SQL, DataFrame, and serialization
changes
can be reviewed as smaller follow-up PRs.

The initial implementation deliberately favors the simpler broadcast
design:
the right input must fit in memory and each left partition scans the
shared
right-side batches. A repartitioned implementation can be evaluated
separately
without changing the ASOF semantics introduced here.

Floating-point equality keys are rejected in this base layer because
Arrow's
required sort order distinguishes `-0.0` from `+0.0` while join equality
does
not. apache#24375 adds the required ordering normalization as an independently
reviewable layer.

## What changes are included in this PR?

- Add `AsOfJoinExec` for left-preserving, Snowflake-style ASOF
semantics.
- Coalesce and collect the ordered right input once, then share it
across all
  left partitions.
- Keep the left input partitioned so each partition can scan
independently and
  preserve the left-side output partitioning.
- Preserve merge state across input and output batch boundaries.
- Reserve each retained Arrow buffer exactly once, including when
right-side
batches are zero-copy slices, and expose build, match, and output
metrics.
- Define output properties and statistics for the broadcast execution
model.
- Reject floating-point equality keys until apache#24375 supplies a
sort/equality
  contract that handles signed zero correctly.
- Add physical operator tests covering match directions, equality
groups,
batch boundaries, unmatched rows, invalid contracts, shared-buffer
memory
accounting, multi-partition broadcast execution, and float-key
rejection.

## Are these changes tested?

Yes:

- `cargo fmt --all`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo test -p datafusion-physical-plan joins::asof_join
--all-features`
- Extended workspace tests from the contributor guide
- FFI integration tests

## Are there any user-facing changes?

This adds a new physical operator API. The base operator deliberately
rejects
floating-point equality keys; apache#24375 adds full Float16, Float32, and
Float64
support. SQL and DataFrame APIs are left to later dependent PRs.

---------

Co-authored-by: Yongting You <[email protected]>

## Which issue does this PR close?

- Part of #318.
- Umbrella PR: apache#23738.
- Depends on apache#23828 (merged).

## Rationale for this change

This is the logical-planning layer of the ASOF JOIN stack. It defines
the
logical contract and planner behavior separately from the SQL frontend
and
serialization formats.

apache#23828 is merged, so this PR's diff against `main` is the isolated
logical
layer. It no longer depends on the optional floating-point follow-up
apache#24375.

## What changes are included in this PR?

- Add `LogicalPlan::AsOfJoin`, `AsOfJoin`, and `AsOfMatch`.
- Validate deterministic expressions, input ownership, supported match
  operators, equality-key types, and USING constraints.
- Add `LogicalPlanBuilder` entry points and schema construction that
preserves
both qualified `USING` keys while exposing one unqualified wildcard key.
- Integrate ASOF joins with tree transforms, display, type coercion,
projection
  pruning, row bounds, and physical planning.
- Plan logical ASOF joins to the broadcast-based `AsOfJoinExec` from
apache#23828.
- Fail closed at proto, SQL unparser, and Substrait boundaries until
their
  owning stack layers add explicit support.
- Defer ASOF-specific functional-dependency refinement to apache#24799 and
filter
  pushdown to apache#24801 so each optimization can be reviewed independently.

## Are these changes tested?

Yes:

- `cargo fmt --all`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo test -p datafusion-expr min_rows_of_joins --all-features`
- `cargo test -p datafusion-substrait
asof_join_fails_closed_until_substrait_has_an_extension --all-features`
- The extended workspace test command from the contributor guide

## Are there any user-facing changes?

This adds logical-plan and builder APIs for ASOF joins. SQL syntax,
DataFrame
APIs, and plan serialization are intentionally left to dependent stack
PRs.
Floating equality keys remain rejected by the merged physical operator
unless
the independent follow-up apache#24375 is also included.

As with any new public `LogicalPlan` variant, downstream exhaustive
matches must
add an arm. The variant is appended so existing variants retain their
`PartialOrd` ordering; maintainers should still treat the enum addition
as a
Rust source-compatibility break.

This PR can be reviewed independently now that apache#23828 has merged. The
optimization follow-ups apache#24799 and apache#24801 are not required by the core
ASOF
stack.

## Which issue does this PR close?

- Part of #318.
- Umbrella PR: apache#23738.
- Builds on apache#23829 and apache#23828, both merged.

## Rationale for this change

This is the SQL frontend layer of the ASOF JOIN stack. It adds syntax
and
unparsing on top of the merged physical and logical contracts. The PR
now
contains only the isolated SQL frontend diff.

## What changes are included in this PR?

- Plan `ASOF JOIN ... MATCH_CONDITION (...)` with optional `ON` or
`USING`
  equality keys.
- Reject unsupported match shapes and non-equality `ON` predicates.
- Unparse ASOF joins while preserving right-side candidate preselection
and
  nested join scope.
- Document the supported SQL syntax and semantics, including qualified
`USING` keys, left-partitioned broadcast execution, the full-right
memory
requirement, repeated scans, and the absence of spill/repartitioned
ASOF.
- Add SQL integration and sqllogictest coverage for all four match
directions,
coercion, equality-free joins, USING, invalid contracts, EXPLAIN,
boundedness,
  and optimized-plan round trips.
- Verify the broadcast topology with a multi-partition left input: left
  partitioning is preserved while the right input is single-partitioned.

## Are these changes tested?

Yes:

- `cargo fmt --all`
- `./ci/scripts/doc_prettier_check.sh --write --allow-dirty`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo test -p datafusion --test core_integration asof --all-features`
- `cargo test -p datafusion-sqllogictest --test sqllogictests
--all-features -- asof_join`
- The extended workspace test command from the contributor guide

## Are there any user-facing changes?

Users can express Snowflake-style ASOF joins in SQL with
`MATCH_CONDITION`, optional equality keys, and `<`, `<=`, `>`, or `>=`
match
directions. With `USING`, wildcard output exposes one unqualified key
while
both qualified input keys remain addressable. The user guide also
documents
the initial broadcast strategy and its memory/no-spill limitations.

apache#23829 and apache#23828 are merged. This is the next core layer in the ASOF
stack and
does not depend on the optional floating-point follow-up apache#24375.

## Which issue does this PR close?

- Part of #318.
- Umbrella PR: apache#23738.

## Rationale for this change

This is the DataFrame API layer of the ASOF JOIN stack. It exposes the
logical contract already available on `main` through the fluent API
without coupling DataFrame callers to SQL.

## What changes are included in this PR?

- Add `DataFrame::join_asof` with an optional equality expression and
one ordered match expression.
- Add `DataFrame::join_asof_using` for same-named equality keys.
- Cover both entry points with a DataFrame execution test.

## Are these changes tested?

Yes:

- `cargo fmt --all`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo test -p datafusion --test core_integration dataframe::join_asof
--all-features`
- The extended workspace test command from the contributor guide

## Are there any user-facing changes?

Users can build ASOF joins through `DataFrame::join_asof` and
`DataFrame::join_asof_using`. Both methods use the same expression
validation and left-preserving semantics as the logical builder and SQL
API.

## Which issue does this PR close?

- Part of #318.
- Umbrella PR: apache#23738.
- Follows apache#23829 (merged).

## Rationale for this change

This is the serialization layer of the ASOF JOIN stack. It gives logical
and
physical ASOF plans explicit protobuf representations without coupling
wire
format review to SQL or DataFrame APIs.

apache#23829 and its prerequisites are merged. After restacking onto the
current
`main`, this PR now contains only the serialization layer.

## What changes are included in this PR?

- Add protobuf messages and enum values for logical and physical ASOF
joins.
- Encode and decode equality keys, ordered match expressions, match
direction,
  join constraint, and right output indices.
- Implement physical serialization through `ExecutionPlan::try_to_proto`
and
  `AsOfJoinExec::try_from_proto`, following the current self-serializing
  execution-plan pattern.
- Regenerate `prost` and `pbjson` sources with the repository generator.
The
  physical-plan oneof uses the next append-only tag after
  `PiecewiseMergeJoinExec`.
- Add logical round trips for all four match directions and a physical
  `AsOfJoinExec` round trip.

## Are these changes tested?

Yes:

- `./datafusion/proto-models/regen.sh`
- `cargo fmt --all`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `./dev/rust_lint.sh`
- `cargo test -p datafusion-proto roundtrip_asof_join --all-features`
- The extended workspace test command from the contributor guide

## Are there any user-facing changes?

Logical and physical ASOF join plans can be serialized through
`datafusion-proto`. The additions use new messages and append-only
oneof/enum
tags, so existing wire tags are not reused. Generated public Rust enums
gain new
variants, however, so downstream exhaustive matches must add arms; this
is a
Rust source-compatibility break even though the wire additions are
compatible.

This PR can now be reviewed independently. It does not depend on the
optional
floating-point follow-up apache#24375.

## Which issue does this PR close?

- Part of #318.
- Umbrella PR: apache#23738.
- All required ASOF feature layers (apache#23828 through apache#23832) are merged.

## Rationale for this change

This is the final core benchmark layer of the ASOF JOIN stack. It uses
the
standard SQL benchmark framework so the workloads share the existing
runner,
result format, and Criterion mode.

## What changes are included in this PR?

- Add an `asof_join` SQL benchmark suite, runnable with
  `./bench.sh run asof_join` or `benchmark_runner`.
- Add seven workloads covering relative input sizes, available ordering,
equality-group cardinality and skew, match direction, and payload width.
- Describe each workload and its explicit ASOF match semantics in the
query.
- Generate pre-sorted Parquet inputs under `DATA_DIR/asof_join` with
`./bench.sh data asof_join`. The benchmark load hook only registers them
with
`WITH ORDER`, keeping data generation and input sorting outside the
measured
  query.
- Assert that every workload plans to `AsOfJoinExec`.

## Are these changes tested?

Yes:

- `cargo fmt --all`
- `./ci/scripts/doc_prettier_check.sh --write --allow-dirty`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo test -p datafusion-benchmarks --lib --bins`
- `DATA_DIR=/tmp/asof_join CARGO_COMMAND='cargo run' ./bench.sh data
asof_join`
- `DATA_DIR=/tmp/asof_join CARGO_COMMAND='cargo run' ./bench.sh run
asof_join 7`
- `cargo run -p datafusion-benchmarks --bin benchmark_runner --
asof_join --iterations 1 --partitions 4 --path /tmp/asof_join`
- `cargo run -p datafusion-benchmarks --bin benchmark_runner --
asof_join --query 7 --partitions 4 --path /tmp/asof_join --criterion`

The pre-sorted workload's physical plan contains `AsOfJoinExec` without
an
input `SortExec`. These smoke runs validate the suite and are not
presented as
performance claims.

## Are there any user-facing changes?

This adds ASOF workloads to the repository benchmark tooling. It does
not
change query semantics or runtime behavior.

This is the final core item tracked by apache#23738. It does not depend on the
optional floating-point follow-up apache#24375.

## Which issue does this PR close?

- Part of #318.
- Umbrella PR: apache#23738.
- Follow-up to the merged physical operator in apache#23828.

## Rationale for this change

`AsOfJoinExec` already supports an internal output projection, but
physical
projection pushdown cannot currently populate it. Queries that need only
a
subset of ASOF output columns therefore materialize unused join columns
before
the outer projection removes them.

## What changes are included in this PR?

- Implement the standard physical projection-embedding hook for
  `AsOfJoinExec`.
- Preserve empty-projection row counts and decline unsafe repeated
embedding.
- Preserve projected plan properties, including dropping output ordering
when
  its sort expressions are no longer available.
- Add focused execution coverage and update the affected ASOF
physical-plan
  expectations.

## Are these changes tested?

Yes. The focused ASOF tests, the complete SQL logic test suite, Clippy
with all
targets and features, and the extended workspace test command from the
contributor guide pass.

I also compared the merged benchmark Q03 on a dedicated EC2
`c7i.4xlarge` with
four partitions. Each revision used its own release build, three
warmups, and
40 measured iterations in ABBA order.

| Revision | Median |
| --- | ---: |
| `main` (`133111f43`) | 107.6 ms |
| This PR (`72ee47918`) | 106.1 ms |

That is a modest 1.3% improvement on Q03. The deterministic benefit is
visible
in the physical plan: the required output columns are embedded in
`AsOfJoinExec`, so unused join columns are not constructed. A residual
`ProjectionExec` is retained only when aliases or expressions still
require it.

## Are there any user-facing changes?

No SQL or logical API changes. Physical plans can avoid materializing
ASOF
output columns that are not required downstream.

)

## Which issue does this PR close?

- Part of #318.
- Umbrella PR: apache#23738.

## Rationale for this change

An ASOF join emits exactly one output row for every left row. Unlike a
general left join, it cannot duplicate left rows, so functional
dependencies from the left input keep their original dependency mode.

Preserving this metadata lets `EliminateJoin` prove that an ASOF result
is unique on a left primary key and remove a downstream redundant join.
Right-side dependencies also remain valid, but are nullable and
non-unique because one right row may match multiple left rows.

## What changes are included in this PR?

This PR restores the left input’s dependency modes and preserves the
right input’s nullable `Multi` dependencies when building the ASOF
output schema. It adds focused schema coverage and SQLLogicTests for
downstream join elimination and right-side grouping.

## What is the testing strategy for this PR?

`asof_join_schema_preserves_dependencies` verifies left and right
dependency modes, nullability, and target columns. The `asof_join.slt`
cases verify end to end that left-side uniqueness removes a redundant
downstream LEFT JOIN and that a right-side primary key still permits
selecting dependent columns when grouping by the key.

## Are there any user-facing changes?

No query results or public APIs change. This improves optimizer metadata
for logical ASOF joins.

## Which issue does this PR close?

- Part of #318.
- Umbrella PR: apache#23738.
- Follow-up to apache#23829.

## Rationale for this change

An ASOF join emits each left row exactly once and does not change
left-side
values. A deterministic predicate that references only the left input
can
therefore run before matching. Right-only and mixed predicates must
remain
above the join because unmatched right fields are NULL-padded, and
volatile
predicates must keep their original evaluation point.

Pushing the safe subset reduces the rows entering the ASOF join and can
expose
the predicate to further scan-level pushdown.

## Performance

Q08 models a high-selectivity case with 10M ordered left rows, 100K
ordered
right rows, and a left predicate that retains 1% of the input. It is
included
in the ASOF benchmark suite.

On an Apple M4 Max, using a release build and two 7-iteration runs per
revision:

| Revision | Median elapsed |
| --- | ---: |
| `1cfc35579` (then-current `main`) | 97.5 ms |
| This change | 15.8 ms |

This workload is **6.17x faster** (**83.8% lower elapsed time**) because
the
join processes 100K left rows instead of 10M.

```shell
target/release/benchmark_runner asof_join --query 8 --iterations 7
```

## What changes are included in this PR?

- Push deterministic left-only conjuncts below logical ASOF joins.
- Keep right-only, mixed, and volatile predicates above the join.
- Cover the observable optimizer behavior in `asof_join.slt`.
- Add Q08 for selective left-side predicates to the ASOF benchmark
suite.

## Are these changes tested?

- `cargo fmt --all -- --check`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo test --profile ci -p datafusion-optimizer --all-features`
- `cargo test --profile ci --test sqllogictests -- asof_join.slt`
- `cargo run --profile ci --bin benchmark_runner -- asof_join --query 8
--iterations 1`
- The extended workspace test command from the contributor guide

## Are there any user-facing changes?

No result semantics or public APIs change. Eligible ASOF joins may
execute with
fewer left-side rows after logical optimization.

## Which issue does this PR close?

- Part of #318.
- Follow-up to apache#24801.

## Rationale for this change

When a filter selects one ASOF equality-key group on the left, rows in
other right-side groups cannot match any surviving left row. The
broadcast join still sorts and retains those right rows unless the same
key filter is applied to the right input.

## Performance

Q09 has 50M rows on each side across 500K equality groups. `WHERE l.key
= 42` retains one group and produces 100 rows with both revisions. On an
Apple M4 Max, using independent release builds and two 7-iteration runs
per revision, the median elapsed time across all 14 iterations was:

| Revision | Median elapsed |
| --- | ---: |
| `main` without this optimization (`92559afc0`) | 972.1 ms |
| This change (`c7522fa37`) | 78.5 ms |

This selective keyed workload is about **12.4x faster** (91.9% lower
elapsed time). The benchmark file and command were the same for both
binaries:

```shell
target/release/benchmark_runner asof_join --query 9 --iterations 7
```

## What changes are included in this PR?

- Mirror a pushed `left_key = literal` predicate to the corresponding
right key when both join keys are direct columns of the same type.
- Keep the original left filter and ASOF null-padding behavior
unchanged. Other predicate shapes and coerced or expression-based keys
are not inferred.
- Add result and plan coverage in `asof_join.slt`, plus a keyed
selective workload as ASOF benchmark Q09.

## What is the testing strategy for this PR?

The SLT cases check the mirrored plan, matching and unmatched results, a
right-side `IS NULL` filter, and a left disjunction that must not prune
the right input. Q09 runs the user-visible query with 50M rows per side
across 500K groups, selecting one group.

## Are there any user-facing changes?

No API or result changes. Eligible ASOF joins can sort and retain fewer
right-side rows.

## Which issue does this PR close?

- Part of #318.
- Follow-up to apache#25574.

## Rationale for this change

apache#25574 mirrors a selective `left_key = literal` filter to the right side
of an ASOF join. The same equality guarantee applies to any
deterministic predicate that references only direct left equality-key
columns, including `IN`, ranges, and `OR` expressions. Mirroring those
predicates avoids sorting and broadcasting right-side groups that cannot
match a surviving left row.

## Performance

Q10 has 50M rows on each side across 500K equality groups. `WHERE l.key
IN (42, 43)` retains two groups and produces 200 rows with both
revisions. On an Apple M4 Max, using independent release builds and two
7-iteration runs per revision, the median elapsed time across all 14
iterations was:

| Revision | Median elapsed |
| --- | ---: |
| Current main after apache#25574 (`acf5c89e6`) | 984.9 ms |
| This change (`5763d505a`) | 82.3 ms |

This selective compound-key workload is about **12.0x faster** (91.6%
lower elapsed time). The benchmark file and command were the same for
both binaries:

```shell
target/release/benchmark_runner asof_join --query 10 --iterations 7
```

## What changes are included in this PR?

- Build an ASOF-specific column mapping for direct equality keys.
- Mirror a pushed deterministic predicate only when every referenced
column has a right-key mapping.
- Support compound `IN`, range, and `OR` predicates while leaving mixed
key/non-key predicates entirely off the right input.
- Add plan coverage and ASOF benchmark Q10.

Coerced and expression-based equality keys remain excluded because they
cannot be replaced column-for-column.

## What is the testing strategy for this PR?

The new `asof_join.slt` cases verify compound predicate mirroring and
the all-or-nothing behavior for a mixed key/non-key `OR` predicate. Q10
covers the user-visible selective workload.

Also run:

- `cargo fmt --all -- --check`
- `cargo clippy --all-targets --all-features -- -D warnings`
- The extended workspace test suite, including all 524 sqllogictest
files

## Are there any user-facing changes?

No API or result changes. More eligible ASOF joins can sort and retain
fewer right-side rows.

---------

Co-authored-by: Jay Zhan <[email protected]>

## Which issue does this PR close?

- Part of #318.
- Follow-up to apache#23830.

## Rationale for this change

ASOF SQL behavior was split between sqllogictest and Rust integration
tests. SQL-visible results and plan shapes are easier to review and
maintain in SLT, while Rust coverage remains useful for execution setup
that SLT cannot express.

## What changes are included in this PR?

- Move duplicated coercion, EXPLAIN, `USING` output, and validation
coverage into `asof_join.slt`.
- Add result and plan-shape coverage for right-output filtering,
projection pruning, expression operands, empty inputs, duplicate left
rows, self joins, and multi-key `USING`.
- Keep Rust tests that require multi-partition physical plans, unbounded
inputs, cross-batch execution, or direct plan-to-SQL roundtrips.

## Are these changes tested?

Yes:

- `cargo fmt --all -- --check`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo test --profile ci -p datafusion-sqllogictest --test
sqllogictests -- asof_join`
- `cargo test --profile ci -p datafusion --test core_integration
--all-features -- asof`
- Extended workspace tests from the contributor guide

## Are there any user-facing changes?

No. This consolidates and expands coverage for the existing ASOF SQL
contract.