duckdb::ArrowMergeEvent Class Reference

Inheritance diagram for duckdb::ArrowMergeEvent:

Collaboration diagram for duckdb::ArrowMergeEvent:

Public Member Functions
	ArrowMergeEvent (ArrowQueryResult &result, BatchedDataCollection &batches, Pipeline &pipeline_p)
void	Schedule () override
Public Member Functions inherited from duckdb::BasePipelineEvent
	BasePipelineEvent (shared_ptr< Pipeline > pipeline)
	BasePipelineEvent (Pipeline &pipeline)
void	PrintPipeline () override
Public Member Functions inherited from duckdb::Event
	Event (Executor &executor)
virtual void	FinishEvent ()
	Called right after the event is finished.
virtual void	FinalizeFinish ()
	Called after the event is entirely finished.
void	FinishTask ()
void	Finish ()
void	AddDependency (Event &event)
bool	HasDependencies () const
const vector< reference< Event > > &	GetParentsVerification () const
void	CompleteDependency ()
void	SetTasks (vector< shared_ptr< Task > > tasks)
void	InsertEvent (shared_ptr< Event > replacement_event)
bool	IsFinished () const
ClientContext &	GetClientContext ()
template<class TARGET >
TARGET &	Cast ()
template<class TARGET >
const TARGET &	Cast () const
Public Member Functions inherited from duckdb::enable_shared_from_this< Event >
shared_ptr< Event >	shared_from_this ()
shared_ptr< Event const >	shared_from_this () const

Public Attributes
ArrowQueryResult &	result
BatchedDataCollection &	batches
Public Attributes inherited from duckdb::BasePipelineEvent
shared_ptr< Pipeline >	pipeline
	The pipeline that this event belongs to.

Private Attributes
idx_t	record_batch_size
	The max size of a record batch to output.

Additional Inherited Members
Protected Member Functions inherited from duckdb::enable_shared_from_this< Event >
	enable_shared_from_this (enable_shared_from_this const &) noexcept
enable_shared_from_this &	operator= (enable_shared_from_this const &) noexcept
Protected Attributes inherited from duckdb::Event
Executor &	executor
atomic< idx_t >	finished_tasks
	The current threads working on the event.
atomic< idx_t >	total_tasks
	The maximum amount of threads that can work on the event.
atomic< idx_t >	finished_dependencies
idx_t	total_dependencies
	The total amount of dependencies.
vector< weak_ptr< Event > >	parents
	The events that depend on this event to run.
vector< reference< Event > >	parents_raw
	Raw pointers to the parents (used for verification only)
atomic< bool >	finished
	Whether or not the event is finished executing.

Constructor & Destructor Documentation

ArrowMergeEvent()

duckdb::ArrowMergeEvent::ArrowMergeEvent	(	ArrowQueryResult &	result,
		BatchedDataCollection &	batches,
		Pipeline &	pipeline_p
	)

    : BasePipelineEvent(pipeline_p), result(result), batches(batches) {
    record_batch_size = result.BatchSize();
}

Member Function Documentation

Schedule()

void duckdb::ArrowMergeEvent::Schedule ( )

overridevirtual

Implements duckdb::Event.

                               {
    vector<shared_ptr<Task>> tasks;
 
    BatchesToTaskTransformer transformer(batches);
    vector<BatchesForTask> task_data;
    bool finished = false;
    // First we convert our list of batches into units of Storage::ROW_GROUP_SIZE tuples each
    while (!finished) {
        idx_t tuples_for_task = 0;
        idx_t start_index = transformer.GetIndex();
        idx_t end_index = start_index;
        while (tuples_for_task < DEFAULT_ROW_GROUP_SIZE) {
            idx_t batch_size;
            if (!transformer.TryGetNextBatchSize(batch_size)) {
                finished = true;
                break;
            }
            end_index++;
            tuples_for_task += batch_size;
        }
        if (start_index == end_index) {
            break;
        }
        BatchesForTask batches_for_task;
        batches_for_task.tuple_count = tuples_for_task;
        batches_for_task.batches = batches.BatchRange(start_index, end_index);
        task_data.push_back(batches_for_task);
    }
 
    // Now we produce tasks from these units
    // Every task is given a scan_state created from the range of batches
    // and a vector of indices indicating the arrays (record batches) they should populate
    idx_t record_batch_index = 0;
    for (auto &data : task_data) {
        const auto tuples = data.tuple_count;
 
        auto full_batches = tuples / record_batch_size;
        auto remainder = tuples % record_batch_size;
        auto total_batches = full_batches + !!remainder;
 
        vector<idx_t> record_batch_indices(total_batches);
        for (idx_t i = 0; i < total_batches; i++) {
            record_batch_indices[i] = record_batch_index++;
        }
 
        BatchCollectionChunkScanState scan_state(batches, data.batches, pipeline->executor.context);
        tasks.push_back(make_uniq<ArrowBatchTask>(result, std::move(record_batch_indices), pipeline->executor,
                                                  shared_from_this(), std::move(scan_state), result.names,
                                                  record_batch_size));
    }
 
    // Allocate the list of record batches inside the query result
    {
        vector<unique_ptr<ArrowArrayWrapper>> arrays;
        arrays.resize(record_batch_index);
        for (idx_t i = 0; i < record_batch_index; i++) {
            arrays[i] = make_uniq<ArrowArrayWrapper>();
        }
        result.SetArrowData(std::move(arrays));
    }
    D_ASSERT(!tasks.empty());
    SetTasks(std::move(tasks));
}

---

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

• external/duckdb/duckdb.cpp