feat(sf_core): atomically take result streams

[zeroshade]

Summary

• add DatabaseDriverV1::result_set_take_stream, a one-shot API that atomically deregisters a result-set handle before constructing its Arrow reader
• guarantee the handle stays released when reader construction fails
• add an internal HandleManager::take_obj primitive so concurrent consumers are first-caller-wins under one write lock
• share stream construction between retained result_set_get_stream and consuming result_set_take_stream
• migrate statement_prepare to the one-shot lifecycle

Motivation

In-process wrappers currently repeat this lifecycle in execution, metadata, and ingestion paths:

1. call result_set_get_stream
2. call result_set_release on success
3. remember to release separately on acquisition failure
4. decide which error wins when both operations fail

The new method centralizes that ownership transition in sf_core. The Rust ADBC driver can replace its shared acquire/release helper with one result_set_take_stream call, while protobuf clients retain the existing independent get/release methods.

Semantics

• the handle is consumed before stream construction starts
• only one caller can take a live handle
• a successful take returns the same reader as result_set_get_stream
• construction failure does not restore or leak the handle
• result_set_get_stream remains reusable and backward compatible

Testing

• cargo test -p sf_core --lib -- --test-threads=1 — 1700 passed, 1 ignored
• targeted success, duplicate-take, release-after-take, and construction-failure cleanup tests
• cargo check -p sf_core --all-targets
• cargo check -p sf_core --no-default-features
• cargo clippy -p sf_core --lib
• cargo doc -p sf_core --no-deps
• cargo fmt --all -- --check
• git diff --check

The full suite was run serially because existing resolver tests mutate process environment and can race under the default parallel test runner.

[Copilot]

Pull request overview

This PR introduces a one-shot result-set streaming API to make result-set handle consumption atomic and failure-safe, reducing duplicate acquire/release logic across callers and tightening concurrency semantics around handle ownership.

Changes:

• Added DatabaseDriverV1::result_set_take_stream to atomically deregister a result-set handle before building its Arrow reader, ensuring the handle stays released even on construction failure.
• Introduced HandleManager::take_obj as an internal primitive to implement first-caller-wins consumption under a single write lock.
• Refactored stream construction to be shared between result_set_get_stream (retained handle) and result_set_take_stream (consuming handle), and migrated statement_prepare to the consuming lifecycle.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

File	Description
sf_core/src/handle_manager.rs	Adds atomic handle consumption via take_obj, and re-implements delete_handle on top of it.
sf_core/src/apis/database_driver_v1/statement.rs	Switches statement_prepare to use the new one-shot result-set streaming API.
sf_core/src/apis/database_driver_v1/result_set.rs	Extracts shared stream construction and adds result_set_take_stream plus targeted tests for consumption and failure cleanup.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.

Suppressed comments (1)

sf_core/src/handle_manager.rs:116

• HandleManager::delete_handle no longer creates its own tracing span, which makes it inconsistent with the rest of HandleManager’s public methods (add_handle, get_obj, drain_matching) and also means delete operations will only show up under the nested HandleManager::take_obj span. Adding a HandleManager::delete_handle span keeps traces searchable and preserves the method-level observability that the other operations have.

    pub fn delete_handle(&self, handle: Handle) -> bool {
        self.take_obj(handle).is_some()
    }