Inheritance diagram for duckdb::PhysicalArrowCollector:

[legend]

Collaboration diagram for duckdb::PhysicalArrowCollector:

[legend]

Public Member Functions
	PhysicalArrowCollector (PhysicalPlan &physical_plan, PreparedStatementData &data, bool parallel, idx_t batch_size)

SinkResultType	Sink (ExecutionContext &context, DataChunk &chunk, OperatorSinkInput &input) const override

SinkCombineResultType	Combine (ExecutionContext &context, OperatorSinkCombineInput &input) const override

unique_ptr< QueryResult >	GetResult (GlobalSinkState &state) const override
	The final method used to fetch the query result from this operator.

unique_ptr< GlobalSinkState >	GetGlobalSinkState (ClientContext &context) const override

unique_ptr< LocalSinkState >	GetLocalSinkState (ExecutionContext &context) const override

SinkFinalizeType	Finalize (Pipeline &pipeline, Event &event, ClientContext &context, OperatorSinkFinalizeInput &input) const override

bool	ParallelSink () const override

bool	SinkOrderDependent () const override

Public Member Functions inherited from duckdb::PhysicalResultCollector
	PhysicalResultCollector (PhysicalPlan &physical_plan, PreparedStatementData &data)

bool	IsSink () const override

vector< const_reference< PhysicalOperator > >	GetChildren () const override

void	BuildPipelines (Pipeline &current, MetaPipeline &meta_pipeline) override

bool	IsSource () const override

virtual bool	IsStreaming () const
	Whether this is a streaming result collector.

Public Member Functions inherited from duckdb::PhysicalOperator
	PhysicalOperator (PhysicalPlan &physical_plan, PhysicalOperatorType type, vector< LogicalType > types, idx_t estimated_cardinality)

	PhysicalOperator (const PhysicalOperator &other)=delete
	Deleted copy constructors.

PhysicalOperator &	operator= (const PhysicalOperator &)=delete

virtual string	GetName () const

virtual InsertionOrderPreservingMap< string >	ParamsToString () const

virtual string	ToString (ExplainFormat format=ExplainFormat::DEFAULT) const

void	Print () const

const vector< LogicalType > &	GetTypes () const
	Return a vector of the types that will be returned by this operator.

virtual bool	Equals (const PhysicalOperator &other) const

idx_t	EstimatedThreadCount () const
	Functions to help decide how to set up pipeline dependencies.

bool	CanSaturateThreads (ClientContext &context) const

virtual void	Verify ()

virtual unique_ptr< OperatorState >	GetOperatorState (ExecutionContext &context) const

virtual unique_ptr< GlobalOperatorState >	GetGlobalOperatorState (ClientContext &context) const

virtual OperatorResultType	Execute (ExecutionContext &context, DataChunk &input, DataChunk &chunk, GlobalOperatorState &gstate, OperatorState &state) const

virtual OperatorFinalizeResultType	FinalExecute (ExecutionContext &context, DataChunk &chunk, GlobalOperatorState &gstate, OperatorState &state) const

virtual OperatorFinalResultType	OperatorFinalize (Pipeline &pipeline, Event &event, ClientContext &context, OperatorFinalizeInput &input) const

virtual bool	ParallelOperator () const

virtual bool	RequiresFinalExecute () const

virtual bool	RequiresOperatorFinalize () const

virtual OrderPreservationType	OperatorOrder () const
	The influence the operator has on order (insertion order means no influence)

virtual unique_ptr< LocalSourceState >	GetLocalSourceState (ExecutionContext &context, GlobalSourceState &gstate) const

virtual unique_ptr< GlobalSourceState >	GetGlobalSourceState (ClientContext &context) const

SourceResultType	GetData (ExecutionContext &context, DataChunk &chunk, OperatorSourceInput &input) const

virtual OperatorPartitionData	GetPartitionData (ExecutionContext &context, DataChunk &chunk, GlobalSourceState &gstate, LocalSourceState &lstate, const OperatorPartitionInfo &partition_info) const

virtual bool	ParallelSource () const

virtual bool	SupportsPartitioning (const OperatorPartitionInfo &partition_info) const

virtual OrderPreservationType	SourceOrder () const
	The type of order emitted by the operator (as a source)

virtual ProgressData	GetProgress (ClientContext &context, GlobalSourceState &gstate) const
	Returns the current progress percentage, or a negative value if progress bars are not supported.

virtual ProgressData	GetSinkProgress (ClientContext &context, GlobalSinkState &gstate, const ProgressData source_progress) const
	Returns the current progress percentage, or a negative value if progress bars are not supported.

virtual InsertionOrderPreservingMap< string >	ExtraSourceParams (GlobalSourceState &gstate, LocalSourceState &lstate) const

virtual void	PrepareFinalize (ClientContext &context, GlobalSinkState &sink_state) const

virtual SinkNextBatchType	NextBatch (ExecutionContext &context, OperatorSinkNextBatchInput &input) const

virtual OperatorPartitionInfo	RequiredPartitionInfo () const

virtual vector< const_reference< PhysicalOperator > >	GetSources () const

bool	AllSourcesSupportBatchIndex () const

template<class TARGET >
TARGET &	Cast ()

template<class TARGET >
const TARGET &	Cast () const

Static Public Member Functions
static unique_ptr< PhysicalOperator >	Create (ClientContext &context, PreparedStatementData &data, idx_t batch_size)

Static Public Member Functions inherited from duckdb::PhysicalResultCollector
static unique_ptr< PhysicalOperator >	GetResultCollector (ClientContext &context, PreparedStatementData &data)

Static Public Member Functions inherited from duckdb::PhysicalOperator
static void	SetEstimatedCardinality (InsertionOrderPreservingMap< string > &result, idx_t estimated_cardinality)

static idx_t	GetMaxThreadMemory (ClientContext &context)
	The maximum amount of memory the operator should use per thread.

static OperatorCachingMode	SelectOperatorCachingMode (ExecutionContext &context)
	Whether operator caching is allowed in the current execution context.

Public Attributes
idx_t	record_batch_size
	User provided batch size.

bool	parallel

Public Attributes inherited from duckdb::PhysicalResultCollector
StatementType	statement_type

StatementProperties	properties

QueryResultMemoryType	memory_type

PhysicalOperator &	plan

vector< string >	names

Public Attributes inherited from duckdb::PhysicalOperator
ArenaLinkedList< reference< PhysicalOperator > >	children
	The child operators.

PhysicalOperatorType	type
	The physical operator type.

vector< LogicalType >	types
	The return types.

idx_t	estimated_cardinality
	The estimated cardinality.

unique_ptr< GlobalSinkState >	sink_state
	The global sink state.

unique_ptr< GlobalOperatorState >	op_state
	The global operator state.

mutex	lock
	Lock for (re)setting any of the operator states.

Additional Inherited Members
Static Public Attributes inherited from duckdb::PhysicalResultCollector
static constexpr const PhysicalOperatorType	TYPE = PhysicalOperatorType::RESULT_COLLECTOR

Static Public Attributes inherited from duckdb::PhysicalOperator
static constexpr const PhysicalOperatorType	TYPE = PhysicalOperatorType::INVALID

Protected Member Functions inherited from duckdb::PhysicalResultCollector
unique_ptr< ColumnDataCollection >	CreateCollection (ClientContext &context) const

Protected Member Functions inherited from duckdb::PhysicalOperator
virtual SourceResultType	GetDataInternal (ExecutionContext &context, DataChunk &chunk, OperatorSourceInput &input) const

Constructor & Destructor Documentation

◆ PhysicalArrowCollector()

duckdb::PhysicalArrowCollector::PhysicalArrowCollector	(	PhysicalPlan &	physical_plan,
		PreparedStatementData &	data,
		bool	parallel,
		idx_t	batch_size
	)

inline

46071 : PhysicalResultCollector(physical_plan, data), record_batch_size(batch_size), parallel(parallel) {

46072 }

duckdb::PhysicalArrowCollector::record_batch_size

idx_t record_batch_size

User provided batch size.

Definition duckdb.cpp:46088

Member Function Documentation

◆ Create()

unique_ptr< PhysicalOperator > duckdb::PhysicalArrowCollector::Create	(	ClientContext &	context,
		PreparedStatementData &	data,
		idx_t	batch_size
	)

static

                                                                              {
    auto &physical_plan = *data.physical_plan;
    auto &root = physical_plan.Root();
 
    if (!PhysicalPlanGenerator::PreserveInsertionOrder(context, root)) {
        // Not an order-preserving plan: use the parallel materialized collector.
        return make_uniq<PhysicalArrowCollector>(physical_plan, data, true, batch_size);
    }
 
    if (!PhysicalPlanGenerator::UseBatchIndex(context, root)) {
        // Order-preserving plan, and we cannot use the batch index: use single-threaded result collector.
        return make_uniq<PhysicalArrowCollector>(physical_plan, data, false, batch_size);
    }
 
    // Order-preserving plan, and we can use the batch index: use a batch collector.
    return make_uniq<PhysicalArrowBatchCollector>(physical_plan, data, batch_size);
}

◆ Sink()

SinkResultType duckdb::PhysicalArrowCollector::Sink	(	ExecutionContext &	context,
		DataChunk &	chunk,
		OperatorSinkInput &	input
	)		const

overridevirtual

The sink method is called constantly with new input, as long as new input is available. Note that this method CAN be called in parallel, proper locking is needed when accessing dat a inside the GlobalSinkState.

Reimplemented from duckdb::PhysicalOperator.

                                                                            {
    auto &lstate = input.local_state.Cast<ArrowCollectorLocalState>();
    // Append to the appender, up to chunk size
 
    auto count = chunk.size();
    auto &appender = lstate.appender;
    D_ASSERT(count != 0);
 
    idx_t processed = 0;
    do {
        if (!appender) {
            // Create the appender if we haven't started this chunk yet
            auto properties = context.client.GetClientProperties();
            D_ASSERT(processed < count);
            auto initial_capacity = MinValue(record_batch_size, count - processed);
            appender = make_uniq<ArrowAppender>(types, initial_capacity, properties,
                                                ArrowTypeExtensionData::GetExtensionTypes(context.client, types));
        }
 
        // Figure out how much we can still append to this chunk
        auto row_count = appender->RowCount();
        D_ASSERT(record_batch_size > row_count);
        auto to_append = MinValue(record_batch_size - row_count, count - processed);
 
        // Append and check if the chunk is finished
        appender->Append(chunk, processed, processed + to_append, count);
        processed += to_append;
        row_count = appender->RowCount();
        if (row_count >= record_batch_size) {
            lstate.FinishArray();
        }
    } while (processed < count);
    return SinkResultType::NEED_MORE_INPUT;
}

Here is the call graph for this function:

◆ Combine()

SinkCombineResultType duckdb::PhysicalArrowCollector::Combine	(	ExecutionContext &	context,
		OperatorSinkCombineInput &	input
	)		const

overridevirtual

The combine is called when a single thread has completed execution of its part of the pipeline, it is the final time that a specific LocalSinkState is accessible. This method can be called in parallel while other Sink() or Combine() calls are active on the same GlobalSinkState.

Reimplemented from duckdb::PhysicalOperator.

                                                                                             {
    auto &gstate = input.global_state.Cast<ArrowCollectorGlobalState>();
    auto &lstate = input.local_state.Cast<ArrowCollectorLocalState>();
    auto &last_appender = lstate.appender;
    auto &arrays = lstate.finished_arrays;
    if (arrays.empty() && !last_appender) {
        // Nothing to do
        return SinkCombineResultType::FINISHED;
    }
    if (last_appender) {
        // FIXME: we could set these aside and merge them in a finalize event in an effort to create more balanced
        // chunks out of these remnants
        lstate.FinishArray();
    }
    // Collect all the finished arrays
    lock_guard<mutex> l(gstate.glock);
    // Move the arrays from our local state into the global state
    gstate.chunks.insert(gstate.chunks.end(), std::make_move_iterator(arrays.begin()),
                         std::make_move_iterator(arrays.end()));
    arrays.clear();
    gstate.tuple_count += lstate.tuple_count;
    return SinkCombineResultType::FINISHED;
}

◆ GetResult()

unique_ptr< QueryResult > duckdb::PhysicalArrowCollector::GetResult ( GlobalSinkState & state ) const

overridevirtual

The final method used to fetch the query result from this operator.

Implements duckdb::PhysicalResultCollector.

                                                                                        {
    auto &gstate = state_p.Cast<ArrowCollectorGlobalState>();
    return std::move(gstate.result);
}

◆ GetGlobalSinkState()

unique_ptr< GlobalSinkState > duckdb::PhysicalArrowCollector::GetGlobalSinkState ( ClientContext & context ) const

overridevirtual

Reimplemented from duckdb::PhysicalOperator.

                                                                                                   {
    return make_uniq<ArrowCollectorGlobalState>();
}

◆ GetLocalSinkState()

unique_ptr< LocalSinkState > duckdb::PhysicalArrowCollector::GetLocalSinkState ( ExecutionContext & context ) const

overridevirtual

Reimplemented from duckdb::PhysicalOperator.

                                                                                                    {
    return make_uniq<ArrowCollectorLocalState>();
}

◆ Finalize()

SinkFinalizeType duckdb::PhysicalArrowCollector::Finalize	(	Pipeline &	pipeline,
		Event &	event,
		ClientContext &	context,
		OperatorSinkFinalizeInput &	input
	)		const

overridevirtual

The finalize is called when ALL threads are finished execution. It is called only once per pipeline, and is entirely single threaded. If Finalize returns SinkResultType::Finished, the sink is marked as finished

Reimplemented from duckdb::PhysicalOperator.

                                                                                          {
    auto &gstate = input.global_state.Cast<ArrowCollectorGlobalState>();
 
    if (gstate.chunks.empty()) {
        if (gstate.tuple_count != 0) {
            throw InternalException(
                "PhysicalArrowCollector Finalize contains no chunks, but tuple_count is non-zero (%d)",
                gstate.tuple_count);
        }
        gstate.result = make_uniq<ArrowQueryResult>(statement_type, properties, names, types,
                                                    context.GetClientProperties(), record_batch_size);
        return SinkFinalizeType::READY;
    }
 
    gstate.result = make_uniq<ArrowQueryResult>(statement_type, properties, names, types, context.GetClientProperties(),
                                                record_batch_size);
    auto &arrow_result = gstate.result->Cast<ArrowQueryResult>();
    arrow_result.SetArrowData(std::move(gstate.chunks));
 
    return SinkFinalizeType::READY;
}

◆ ParallelSink()

bool duckdb::PhysicalArrowCollector::ParallelSink ( ) const

overridevirtual

Reimplemented from duckdb::PhysicalOperator.

                                                {
    return parallel;
}

◆ SinkOrderDependent()

bool duckdb::PhysicalArrowCollector::SinkOrderDependent ( ) const

overridevirtual

Whether or not the sink operator depends on the order of the input chunks If this is set to true, we cannot do things like caching intermediate vectors

Reimplemented from duckdb::PhysicalOperator.

                                                      {
    return true;
}

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

external/duckdb/duckdb.cpp

Public Member Functions

Static Public Member Functions

Public Attributes

Additional Inherited Members

Constructor & Destructor Documentation

◆ PhysicalArrowCollector()

Member Function Documentation

◆ Create()

◆ Sink()

◆ Combine()

◆ GetResult()

◆ GetGlobalSinkState()

◆ GetLocalSinkState()

◆ Finalize()

◆ ParallelSink()

◆ SinkOrderDependent()