duckdb::DuckDBAdbcConnectionWrapper Struct Reference

Collaboration diagram for duckdb::DuckDBAdbcConnectionWrapper:

Public Member Functions
void	RegisterStream (duckdb_adbc::DuckDBAdbcStreamWrapper *stream)
	Register a stream wrapper so it can be materialized if another query runs on this connection.
void	UnregisterStream (duckdb_adbc::DuckDBAdbcStreamWrapper *stream)
	Unregister a stream wrapper.
void	MaterializeStreams ()
	Materialize all active streams, fetching remaining data into memory.
void	DetachAndClearStreams ()
	Detach all streams from this connection and clear the list (called on connection release).

Public Attributes
duckdb_connection	connection
unordered_map< string, string >	options

Private Attributes
mutex	stream_mutex
vector< duckdb_adbc::DuckDBAdbcStreamWrapper * >	active_streams

Member Function Documentation

RegisterStream()

void duckdb::DuckDBAdbcConnectionWrapper::RegisterStream

(

duckdb_adbc::DuckDBAdbcStreamWrapper *

stream

)

Register a stream wrapper so it can be materialized if another query runs on this connection.

                                                                                                 {
    const duckdb::lock_guard<duckdb::mutex> guard(stream_mutex);
    active_streams.push_back(stream);
}

UnregisterStream()

void duckdb::DuckDBAdbcConnectionWrapper::UnregisterStream

(

duckdb_adbc::DuckDBAdbcStreamWrapper *

stream

)

Unregister a stream wrapper.

                                                                                                   {
    const duckdb::lock_guard<duckdb::mutex> guard(stream_mutex);
    auto it = std::find(active_streams.begin(), active_streams.end(), stream);
    if (it != active_streams.end()) {
        active_streams.erase(it);
    }
}

MaterializeStreams()

void duckdb::DuckDBAdbcConnectionWrapper::MaterializeStreams

(

)

Materialize all active streams, fetching remaining data into memory.

                                                           {
    const duckdb::lock_guard<duckdb::mutex> guard(stream_mutex);
    for (auto *result_wrapper : active_streams) {
        if (!result_wrapper || result_wrapper->materialized) {
            continue;
        }
 
        // Collect remaining batches from the streaming result
        duckdb::vector<ArrowArray> batches;
        auto arrow_options = duckdb_result_get_arrow_options(&result_wrapper->result);
        while (true) {
            ArrowArray array;
            std::memset(&array, 0, sizeof(ArrowArray));
 
            auto duckdb_chunk = duckdb_fetch_chunk(result_wrapper->result);
            if (!duckdb_chunk) {
                break;
            }
            auto conversion_err = duckdb_data_chunk_to_arrow(arrow_options, duckdb_chunk, &array);
            duckdb_destroy_data_chunk(&duckdb_chunk);
 
            if (conversion_err) {
                duckdb_destroy_error_data(&conversion_err);
                if (array.release) {
                    array.release(&array);
                }
                break;
            }
            batches.push_back(array);
        }
        duckdb_destroy_arrow_options(&arrow_options);
 
        // Store materialized data
        auto mat = static_cast<duckdb_adbc::MaterializedData *>(malloc(sizeof(duckdb_adbc::MaterializedData)));
        if (!mat) {
            // Allocation failed — release fetched batches and skip materialization
            for (auto &batch : batches) {
                if (batch.release) {
                    batch.release(&batch);
                }
            }
            continue;
        }
        mat->current = 0;
        mat->count = static_cast<idx_t>(batches.size());
        if (!batches.empty()) {
            mat->batches = static_cast<ArrowArray *>(malloc(sizeof(ArrowArray) * batches.size()));
            if (!mat->batches) {
                // Allocation failed — release fetched batches and skip materialization
                for (auto &batch : batches) {
                    if (batch.release) {
                        batch.release(&batch);
                    }
                }
                free(mat);
                continue;
            }
            for (idx_t i = 0; i < batches.size(); i++) {
                mat->batches[i] = batches[i];
            }
        } else {
            mat->batches = nullptr;
        }
        result_wrapper->materialized = mat;
    }
}

DetachAndClearStreams()

void duckdb::DuckDBAdbcConnectionWrapper::DetachAndClearStreams

(

)

Detach all streams from this connection and clear the list (called on connection release).

                                                              {
    const duckdb::lock_guard<duckdb::mutex> guard(stream_mutex);
    for (auto *stream_wrapper : active_streams) {
        if (stream_wrapper) {
            stream_wrapper->conn_wrapper = nullptr;
        }
    }
    active_streams.clear();
}

---

The documentation for this struct was generated from the following file:

• external/duckdb/duckdb.cpp