[HyukjinKwon]

What changes were proposed in this pull request?

SPARK-61180: PySpark Connect's DataFrame._joinAsOf(other, leftAsOfColumn, rightAsOfColumn, on=None, how=None, *, tolerance=None, allowExactMatches=True, direction="backward") builds the Spark Connect AsOfJoin relation. pandas-on-Spark's merge_asof calls it.

• Rust core: DataFrame::as_of_join(&right, AsOfJoinOptions) with a new AsOfJoinOptions struct (AsOfJoinOptions::new(left_as_of, right_as_of) sets the PySpark defaults: inner, exact matches allowed, "backward"). The plan is a new LogicalPlan::AsOfJoin, serialized to AsOfJoin with left_as_of, right_as_of, join_expr or using_columns, join_type, tolerance, allow_exact_matches and direction. A DataFrame of another session fails with SESSION_NOT_SAME.
• PyO3: DataFrame._joinAsOf with PySpark's signature. how defaults to inner and must be a string. tolerance must be a Column. on is a name, a Column, or a list of names or Columns, as in plan.AsOfJoin. A string as-of column is a column of its DataFrame, as df._col(name). join_type and direction are sent as they are, so the server checks them as it does for PySpark.

Why are the changes needed?

ps.merge_asof failed, because _joinAsOf did not exist and the name resolved through DataFrame.__getattr__ to a Column:

>>> ps.merge_asof(ps.DataFrame(L), ps.DataFrame(R), on="t")
TypeError: 'pyspark.sql.connect.column.Column' object is not callable

PySpark returns [[1, 1.0, 'a', nan, None], [5, 5.0, 'b', 4.0, 'y']] for L = [(1, 1.0, "a"), (5, 5.0, "b")] ("t int, x double, k string") and R = [(2, 2.0, "x"), (4, 4.0, "y")] ("t int, y double, k string").

Does this PR introduce any user-facing change?

Yes. DataFrame._joinAsOf and pandas.merge_asof (pandas-on-Spark) work on the Rust client as they do on PySpark Connect. Before, they raised the TypeError above.

How was this patch tested?

• test_join_as_of_sends_the_as_of_join_as_pyspark (offline, python/tests/test_dropin_offline.py): checks the fields of the sent AsOfJoin relation for a list on with how, tolerance, allowExactMatches=False and direction, for the defaults, and for a Column on.
• test_join_as_of_and_merge_asof_give_the_rows_of_pyspark (e2e, python/tests/test_dropin_e2e.py): the rows of _joinAsOf for each direction, allowExactMatches, tolerance, on as a list, and the inner default, and the rows of ps.merge_asof with on, by, direction, tolerance, and left_on/right_on/left_by/right_by.
• as_of_join_sends_every_field_of_the_as_of_join, as_of_join_defaults_and_sends_a_condition_as_join_expr and as_of_join_rejects_a_dataframe_of_another_session (Rust unit tests, crates/spark-connect/src/dataframe.rs).
• The offline test fails without this change, with TypeError: 'pyspark.sql.connect.column.Column' object is not callable. The same cases give the same schemas, rows, and error messages on official PySpark 4.3.0 against the same server. The explain text matches apart from plan IDs.
• The lib tests of both crates (611 passed), the build of all tests, also with wasm-udf, the offline Python suites (2228 passed, 8 skipped), and the live drop-in and Arrow suites (211 passed) and live cases (10 passed) against a local Spark 4.3.0 server. Clippy reports no diagnostics in the changed lines.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Opus 5.5

This pull request and its description were written by Isaac.