feat(kernel): consume the folded directResults execute()

The kernel's `Statement.execute()` is now the directResults call
(databricks/databricks-sql-kernel#136): it returns either a terminal
`ExecutedStatement` (the query finished within the server inline wait) or a
still-running `ExecutedAsyncStatement`, instead of always blocking to terminal.

`KernelDatabricksClient.execute_command` now feature-detects the arm (via
`await_result`, present only on the async handle):
  - terminal -> build the result set from the `ExecutedStatement` as before;
  - still-running -> register the handle (so `cursor.cancel()` / `close()` reach
    it), then `await_result()` to drive it to terminal — preserving execute()'s
    blocking contract.

`cursor.execute()` behaviour is unchanged for callers (still blocks to a ready
result set); this just adapts to the kernel's new single result-or-handle
return. Mid-run cancel still works via the `_sync_cancellers` canceller
registered before execute.

Tested e2e against a live warehouse: CREATE/INSERT commit, 100k read, error
surfaces at execute, mid-run cancel (~130ms). Latency vs Thrift: SELECT 1
~parity, range(100000) ~parity.

Co-authored-by: Isaac
Signed-off-by: Madhavendra Rathore <madhavendra.rathore@databricks.com>

Author: msrathore-db

src/databricks/sql/backend/kernel/client.py

@@ -472,19 +472,33 @@ def execute_command(
                 except Exception:
                     # Canceller is best-effort; never block execute on it.
                     pass
-                executed = stmt.execute()
-                # Execute succeeded: the kernel now owns the statement
-                # lifecycle. It auto-closes the server statement when the
-                # result stream is fully drained (``ExecutedStatement::
-                # next_batch`` end-of-stream), with the executed handle's
-                # ``Drop`` as the backstop for partial/abandoned reads.
-                # So we must NOT close ``stmt`` here: a premature
-                # ``CloseStatement`` at execute-return broke lazy
-                # CloudFetch chunk-link fetches (``get_result_chunks``
-                # against the live statement) for large paginated-link
-                # results. Closing here is left ONLY for the error path
-                # below, where no executed handle / result set was
-                # produced to reap it.
+                # ``execute()`` is now the kernel's directResults call: it returns
+                # either a terminal ``ExecutedStatement`` (finished within the
+                # server inline wait) or an ``ExecutedAsyncStatement`` (still
+                # running). Feature-detect via ``await_result`` (only on the
+                # async arm).
+                result = stmt.execute()
+                if hasattr(result, "await_result"):
+                    # Still-running arm. Register the handle so ``cursor.cancel()``
+                    # (cancel_command) and ``close_command`` can reach it, then
+                    # drive it to terminal — preserving ``execute()``'s blocking
+                    # contract. The ``_sync_cancellers`` canceller registered
+                    # above also covers a cancel issued during this drive.
+                    command_id = CommandId.from_sea_statement_id(result.statement_id)
+                    cursor.active_command_id = command_id
+                    with self._async_handles_lock:
+                        self._async_handles[command_id.guid] = result
+                        self._async_statements[command_id.guid] = stmt
+                    close_stmt = False
+                    stream = result.await_result()
+                    return self._make_result_set(stream, cursor, command_id)
+                # Terminal arm: an ``ExecutedStatement`` with the result inline.
+                executed = result
+                # The kernel now owns the statement lifecycle (auto-closes on
+                # stream drain, with ``Drop`` as backstop), so we must NOT close
+                # ``stmt`` here — a premature ``CloseStatement`` breaks lazy
+                # CloudFetch chunk-link fetches. Closing is left to the error
+                # path below.
                 close_stmt = False
             except Exception as exc:
                 raise _wrap_kernel_exception("execute_command", exc) from exc