datafusion.plan#

This module supports physical and logical plans in DataFusion.

Classes#

`ExecutionPlan`	Represent nodes in the DataFusion Physical Plan.
`LogicalPlan`	Logical Plan.
`Metric`	A single execution metric with name, value, partition, and labels.
`MetricsSet`	A set of metrics for a single execution plan operator.
`PhysicalPartitioning`	How a physical plan's output rows are spread across its partitions.

Module Contents#

class datafusion.plan.ExecutionPlan(plan: datafusion._internal.ExecutionPlan)#

Represent nodes in the DataFusion Physical Plan.

This constructor should not be called by the end user.

__repr__() → str#: Print a string representation of the physical plan.

children() → list[ExecutionPlan]#

Get a list of children ExecutionPlan that act as inputs to this plan.

The returned list will be empty for leaf nodes such as scans, will contain a single value for unary nodes, or two values for binary nodes (such as joins).

collect_metrics() → list[tuple[str, MetricsSet]]#

Return runtime statistics for each step of the query execution.

DataFusion executes a query as a pipeline of operators — for example a data source scan, followed by a filter, followed by a projection. After the DataFrame has been executed (via collect(), execute_stream(), etc.), each operator records statistics such as how many rows it produced and how much CPU time it consumed.

Each entry in the returned list corresponds to one operator that recorded metrics. The first element of the tuple is the operator’s description string — the same text shown by display_indent() — which identifies both the operator type and its key parameters, for example "FilterExec: column1@0 > 1" or "DataSourceExec: partitions=1".

Returns:: A list of (description, MetricsSet) tuples ordered from the outermost operator (top of the execution tree) down to the data-source leaves. Only operators that recorded at least one metric are included. Returns an empty list if called before the DataFrame has been executed.

display() → str#: Print the physical plan.

display_indent() → str#: Print an indented form of the physical plan.

static from_bytes(ctx: datafusion.context.SessionContext, data: bytes) → ExecutionPlan#

Create an ExecutionPlan from serialized protobuf bytes.

Decoding routes through the codecs installed on ctx with with_physical_extension_codec().

Unlike datafusion.Expr.from_bytes(), ctx is required and positional, and there is no fallback to a worker or global context. ctx need share nothing with the session that encoded the plan: a scan over a table registered from record batches decodes here, because the batches travel inside the encoded scan.

static from_proto(ctx: datafusion.context.SessionContext, data: bytes) → ExecutionPlan#: Deprecated alias for from_bytes().

metrics() → MetricsSet | None#

Return metrics for this plan node, or None if this plan has no MetricsSet.

Some operators (e.g. DataSourceExec) eagerly initialize a MetricsSet when the plan is created, so this may return a set even before execution. Metric values (such as output_rows) are only meaningful after the DataFrame has been executed.

to_bytes(ctx: datafusion.context.SessionContext | None = None) → bytes#

Convert an ExecutionPlan into serialized protobuf bytes.

When ctx is supplied, encoding routes through the codecs installed on it with with_physical_extension_codec().

Unlike LogicalPlan.to_bytes(), a plan reading a table registered from record batches does round-trip: the batches travel inside the encoded scan.

Round-tripping through this method and from_bytes() is how an extension library checks that its own codec claimed its nodes, rather than a codec installed earlier in the chain — see When codec order matters.

to_proto() → bytes#: Deprecated alias for to_bytes().

_raw_plan#

property output_partitioning: PhysicalPartitioning#

Returns how this plan’s output rows are spread across its partitions.

Where partition_count gives only the number of partitions, this also reports the scheme, so a caller executing partitions separately can tell whether they are hash-distributed on known keys or merely counted. A plan does not necessarily partition the way it was asked to; see Checking what the plan actually does.

Examples

>>> import pyarrow as pa
>>> from datafusion import SessionConfig, SessionContext
>>> ctx = SessionContext(SessionConfig().with_target_partitions(4))
>>> ctx.register_record_batches("t", [
...     [pa.record_batch({"a": [1, 2, 3]})],
...     [pa.record_batch({"a": [4, 5, 6]})],
... ])
>>> ctx.sql("select a from t").execution_plan().output_partitioning
UnknownPartitioning(2)

A group-by redistributes rows, so the plan reports the keys:

>>> grouped = ctx.sql("select a, count(*) from t group by a")
>>> partitioning = grouped.execution_plan().output_partitioning
>>> partitioning.scheme
'Hash'
>>> partitioning.partition_count
4

property partition_count: int#

Returns the number of partitions in the physical plan.

Examples

>>> from datafusion import SessionContext
>>> ctx = SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> df.execution_plan().partition_count
1

class datafusion.plan.LogicalPlan(plan: datafusion._internal.LogicalPlan)#

Logical Plan.

A LogicalPlan is a node in a tree of relational operators (such as Projection or Filter).

Represents transforming an input relation (table) to an output relation (table) with a potentially different schema. Plans form a dataflow tree where data flows from leaves up to the root to produce the query result.

A LogicalPlan can be created by the SQL query planner, the DataFrame API, or programmatically (for example custom query languages).

This constructor should not be called by the end user.

__eq__(other: object) → bool#: Test equality.

__repr__() → str#: Generate a printable representation of the plan.

display() → str#: Print the logical plan.

display_graphviz() → str#

Print the graph visualization of the logical plan.

Returns a formattable structure that produces lines meant for graphical display using the DOT language. This format can be visualized using software from graphviz.

display_indent() → str#: Print an indented form of the logical plan.

display_indent_schema() → str#: Print an indented form of the schema for the logical plan.

static from_bytes(ctx: datafusion.context.SessionContext, data: bytes) → LogicalPlan#

Create a LogicalPlan from serialized protobuf bytes.

Decoding routes through the codecs installed on ctx with with_logical_extension_codec(). Tables created in memory from record batches are currently not supported.

Unlike datafusion.Expr.from_bytes(), ctx is required and positional, and there is no fallback to a worker or global context.

static from_proto(ctx: datafusion.context.SessionContext, data: bytes) → LogicalPlan#: Deprecated alias for from_bytes().

inputs() → list[LogicalPlan]#: Returns the list of inputs to the logical plan.

to_bytes(ctx: datafusion.context.SessionContext | None = None) → bytes#

Convert a LogicalPlan to serialized protobuf bytes.

When ctx is supplied, encoding routes through the codecs installed on it with with_logical_extension_codec(), so extension codecs see the encode path. With ctx=None a default codec is used. Tables created in memory from record batches are currently not supported.

to_proto() → bytes#: Deprecated alias for to_bytes().

to_variant() → Any#: Convert the logical plan into its specific variant.

_raw_plan#

class datafusion.plan.Metric(raw: datafusion._internal.Metric)#

A single execution metric with name, value, partition, and labels.

This constructor should not be called by the end user.

__repr__() → str#: Return a string representation of the metric.

labels() → dict[str, str]#

Return the labels associated with this metric.

Labels provide additional context for a metric. For example:

metric.labels()
# {'output_type': 'final'}

_raw#

property name: str#: The name of this metric (e.g. output_rows).

property partition: int | None#

The 0-based partition index this metric applies to.

Returns None for metrics that are not partition-specific (i.e. they apply globally across all partitions of the operator).

property value: int | datetime.datetime | None#

The value of this metric.

Returns an int for counters, gauges, and time-based metrics (nanoseconds), a datetime (UTC) for start_timestamp / end_timestamp metrics, or None when the value has not been set or is not representable.

property value_as_datetime: datetime.datetime | None#

The value as a UTC datetime for timestamp metrics.

Returns None for all non-timestamp metrics and for timestamp metrics whose value has not been set (e.g. before execution).

class datafusion.plan.MetricsSet(raw: datafusion._internal.MetricsSet)#

A set of metrics for a single execution plan operator.

A physical plan operator runs independently across one or more partitions. metrics() returns the raw per-partition Metric objects. The convenience properties (output_rows, elapsed_compute, etc.) automatically sum the named metric across all partitions, giving a single aggregate value for the operator as a whole.

This constructor should not be called by the end user.

__repr__() → str#: Return a string representation of the metrics set.

metrics() → list[Metric]#: Return all individual metrics in this set.

sum_by_name(name: str) → int | None#

Sum the named metric across all partitions.

Useful for accessing any metric not exposed as a first-class property. Returns None if no metric with the given name was recorded.

Parameters:: name – The metric name, e.g. "output_rows" or "elapsed_compute".

_raw#

property elapsed_compute: int | None#

Total CPU time (in nanoseconds) spent inside this operator’s execute loop.

Summed across all partitions. Returns None if no elapsed_compute metric was recorded.

property output_rows: int | None#: Sum of output_rows across all partitions.

property spill_count: int | None#

Number of times this operator spilled data to disk due to memory pressure.

This is a count of spill events, not a byte count. Summed across all partitions. Returns None if no spill_count metric was recorded.

property spilled_bytes: int | None#: Sum of spilled_bytes across all partitions.

property spilled_rows: int | None#: Sum of spilled_rows across all partitions.

class datafusion.plan.PhysicalPartitioning(partitioning: datafusion._internal.PhysicalPartitioning)#

How a physical plan’s output rows are spread across its partitions.

Returned by ExecutionPlan.output_partitioning. This is the partitioning a built plan has. Distinct from datafusion.expr.Partitioning, the logical partitioning a Repartition node records and hands back from partitioning_scheme() — a request, which the plan need not honour. See Checking what the plan actually does.

This constructor should not be called by the end user.

__eq__(other: object) → bool#

Compare two partitionings by scheme, count and hash expressions.

Equality is structural, and does not mirror DataFusion’s own comparison of the underlying type, under which two UnknownPartitioning values of the same width are unequal.

Examples

>>> from datafusion import SessionContext
>>> ctx = SessionContext()
>>> a = ctx.from_pydict({"a": [1, 2, 3]}).execution_plan()
>>> b = ctx.from_pydict({"b": [4, 5, 6]}).execution_plan()
>>> a.output_partitioning == b.output_partitioning
True
>>> a.output_partitioning == "UnknownPartitioning(1)"
False

__hash__() → int#

Hash the partitioning, consistently with __eq__().

Examples

>>> from datafusion import SessionContext
>>> ctx = SessionContext()
>>> a = ctx.from_pydict({"a": [1, 2, 3]}).execution_plan()
>>> b = ctx.from_pydict({"b": [4, 5, 6]}).execution_plan()
>>> len({a.output_partitioning, b.output_partitioning})
1

__repr__() → str#

Print a string representation of the partitioning.

Examples

>>> from datafusion import SessionContext
>>> ctx = SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> repr(df.execution_plan().output_partitioning)
'UnknownPartitioning(1)'

_key() → tuple[str, int, tuple[str, Ellipsis] | None]#

_raw_partitioning#

property hash_expressions: list[str] | None#

The expressions rows are hashed on, or None for other schemes.

Physical expressions have no Python representation, so these are returned in their displayed form. None covers "Range" as well, whose ordering and split points this class does not expose.

Examples

>>> import pyarrow as pa
>>> from datafusion import SessionConfig, SessionContext
>>> ctx = SessionContext(SessionConfig().with_target_partitions(4))
>>> ctx.register_record_batches("t", [
...     [pa.record_batch({"a": [1, 2, 3]})],
...     [pa.record_batch({"a": [4, 5, 6]})],
... ])
>>> scan = ctx.sql("select a from t").execution_plan()
>>> scan.output_partitioning.hash_expressions is None
True
>>> grouped = ctx.sql("select a, count(*) from t group by a")
>>> grouped.execution_plan().output_partitioning.hash_expressions
['a@0']

property partition_count: int#

The number of partitions.

Examples

>>> from datafusion import SessionContext
>>> ctx = SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> df.execution_plan().output_partitioning.partition_count
1

property scheme: Literal['RoundRobinBatch', 'Hash', 'Range', 'UnknownPartitioning']#

Which partitioning scheme this is.

"UnknownPartitioning" means the plan knows how many partitions it has but nothing about how rows are distributed between them, which is the usual case for a file scan. "RoundRobinBatch" and "Hash" come from a repartition the optimizer inserted. "Range" only appears on a plan this package did not build; see Checking what the plan actually does.

Examples

>>> from datafusion import SessionContext
>>> ctx = SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> df.execution_plan().output_partitioning.scheme
'UnknownPartitioning'