Struct datafusion::datasource::physical_plan::parquet::ParquetExec

pub struct ParquetExec { /* private fields */ }

Execution plan for reading one or more Parquet files.

            ▲
            │
            │  Produce a stream of
            │  RecordBatches
            │
┌───────────────────────┐
│                       │
│      ParquetExec      │
│                       │
└───────────────────────┘
            ▲
            │  Asynchronously read from one
            │  or more parquet files via
            │  ObjectStore interface
            │
            │
  .───────────────────.
 │                     )
 │`───────────────────'│
 │    ObjectStore      │
 │.───────────────────.│
 │                     )
  `───────────────────'

Example: Create a ParquetExec

// Create a ParquetExec for reading `file1.parquet` with a file size of 100MB
let file_scan_config = FileScanConfig::new(object_store_url, file_schema)
   .with_file(PartitionedFile::new("file1.parquet", 100*1024*1024));
let exec = ParquetExec::builder(file_scan_config)
  // Provide a predicate for filtering row groups/pages
  .with_predicate(predicate)
  .build();

Features

Supports the following optimizations:

• Concurrent reads: Can read from one or more files in parallel as multiple partitions, including concurrently reading multiple row groups from a single file.

• Predicate push down: skips row groups and pages based on min/max/null_counts in the row group metadata, the page index and bloom filters.

• Projection pushdown: reads and decodes only the columns required.

• Limit pushdown: stop execution early after some number of rows are read.

• Custom readers: customize reading parquet files, e.g. to cache metadata, coalesce I/O operations, etc. See ParquetFileReaderFactory for more details.

• Schema adapters: read parquet files with different schemas into a unified table schema. This can be used to implement “schema evolution”. See SchemaAdapterFactory for more details.

• metadata_size_hint: controls the number of bytes read from the end of the file in the initial I/O when the default ParquetFileReaderFactory. If a custom reader is used, it supplies the metadata directly and this parameter is ignored. ParquetExecBuilder::with_metadata_size_hint for more details.

• User provided ParquetAccessPlans to skip row groups and/or pages based on external information. See “Implementing External Indexes” below

Implementing External Indexes

It is possible to restrict the row groups and selections within those row groups that the ParquetExec will consider by providing an initial ParquetAccessPlan as extensions on PartitionedFile. This can be used to implement external indexes on top of parquet files and select only portions of the files.

The ParquetExec will try and reduce any provided ParquetAccessPlan further based on the contents of ParquetMetadata and other settings.

Example of providing a ParquetAccessPlan

// create an access plan to scan row group 0, 1 and 3 and skip row groups 2 and 4
let mut access_plan = ParquetAccessPlan::new_all(5);
access_plan.skip(2);
access_plan.skip(4);
// provide the plan as extension to the FileScanConfig
let partitioned_file = PartitionedFile::new("my_file.parquet", 1234)
  .with_extensions(Arc::new(access_plan));
// create a ParquetExec to scan this file
let file_scan_config = FileScanConfig::new(ObjectStoreUrl::local_filesystem(), schema())
    .with_file(partitioned_file);
// this parquet exec will not even try to read row groups 2 and 4. Additional
// pruning based on predicates may also happen
let exec = ParquetExec::builder(file_scan_config).build();

For a complete example, see the [advanced_parquet_index example]).

Execution Overview

• Step 1: ParquetExec::execute is called, returning a FileStream configured to open parquet files with a ParquetOpener.

• Step 2: When the stream is polled, the ParquetOpener is called to open the file.

• Step 3: The ParquetOpener gets the ParquetMetaData (file metadata) via ParquetFileReaderFactory, creating a ParquetAccessPlan by applying predicates to metadata. The plan and projections are used to determine what pages must be read.

• Step 4: The stream begins reading data, fetching the required pages and incrementally decoding them.

• Step 5: As each [RecordBatch] is read, it may be adapted by a SchemaAdapter to match the table schema. By default missing columns are filled with nulls, but this can be customized via SchemaAdapterFactory.

Implementations

impl ParquetExec

pub fn new( base_config: FileScanConfig, predicate: Option<Arc<dyn PhysicalExpr>>, metadata_size_hint: Option<usize>, table_parquet_options: TableParquetOptions, ) -> Self

👎Deprecated since 39.0.0: use ParquetExec::builder or ParquetExecBuilder

Create a new Parquet reader execution plan provided file list and schema.

pub fn builder(file_scan_config: FileScanConfig) -> ParquetExecBuilder

Return a ParquetExecBuilder.

See example on ParquetExec and ParquetExecBuilder for specifying parquet table options.

pub fn base_config(&self) -> &FileScanConfig

FileScanConfig that controls this scan (such as which files to read)

pub fn table_parquet_options(&self) -> &TableParquetOptions

Options passed to the parquet reader for this scan

pub fn predicate(&self) -> Option<&Arc<dyn PhysicalExpr>>

Optional predicate.

pub fn pruning_predicate(&self) -> Option<&Arc<PruningPredicate>>

Optional reference to this parquet scan’s pruning predicate

pub fn with_parquet_file_reader_factory( self, parquet_file_reader_factory: Arc<dyn ParquetFileReaderFactory>, ) -> Self

Optional user defined parquet file reader factory.

See documentation on ParquetExecBuilder::with_parquet_file_reader_factory

pub fn with_schema_adapter_factory( self, schema_adapter_factory: Arc<dyn SchemaAdapterFactory>, ) -> Self

Optional schema adapter factory.

See documentation on ParquetExecBuilder::with_schema_adapter_factory

pub fn with_pushdown_filters(self, pushdown_filters: bool) -> Self

If true, any filter Exprs on the scan will converted to a RowFilter in the ParquetRecordBatchStream. These filters are applied by the parquet decoder to skip unecessairly decoding other columns which would not pass the predicate. Defaults to false

pub fn with_reorder_filters(self, reorder_filters: bool) -> Self

If true, the RowFilter made by pushdown_filters may try to minimize the cost of filter evaluation by reordering the predicate Exprs. If false, the predicates are applied in the same order as specified in the query. Defaults to false.

pub fn with_enable_page_index(self, enable_page_index: bool) -> Self

If enabled, the reader will read the page index This is used to optimise filter pushdown via RowSelector and RowFilter by eliminating unnecessary IO and decoding

pub fn with_bloom_filter_on_read(self, bloom_filter_on_read: bool) -> Self

If enabled, the reader will read by the bloom filter

pub fn with_bloom_filter_on_write( self, enable_bloom_filter_on_write: bool, ) -> Self

If enabled, the writer will write by the bloom filter

Trait Implementations

impl Clone for ParquetExec

fn clone(&self) -> ParquetExec

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for ParquetExec

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl DisplayAs for ParquetExec

fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter<'_>) -> Result

Format according to DisplayFormatType, used when verbose representation looks different from the default one

impl ExecutionPlan for ParquetExec

fn as_any(&self) -> &dyn Any

Return a reference to Any that can be used for downcasting

fn repartitioned( &self, target_partitions: usize, config: &ConfigOptions, ) -> Result<Option<Arc<dyn ExecutionPlan>>>

Redistribute files across partitions according to their size See comments on FileGroupPartitioner for more detail.

fn name(&self) -> &'static str

Short name for the ExecutionPlan, such as ‘ParquetExec’.

fn properties(&self) -> &PlanProperties

Return properties of the output of the ExecutionPlan, such as output ordering(s), partitioning information etc.

fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>>

Get a list of children ExecutionPlans 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).

fn with_new_children( self: Arc<Self>, _: Vec<Arc<dyn ExecutionPlan>>, ) -> Result<Arc<dyn ExecutionPlan>>

Returns a new ExecutionPlan where all existing children were replaced by the children, in order

fn execute( &self, partition_index: usize, ctx: Arc<TaskContext>, ) -> Result<SendableRecordBatchStream>

Begin execution of partition, returning a Stream of RecordBatches.

fn metrics(&self) -> Option<MetricsSet>

Return a snapshot of the set of Metrics for this ExecutionPlan. If no Metrics are available, return None.

fn statistics(&self) -> Result<Statistics>

Returns statistics for this ExecutionPlan node. If statistics are not available, should return Statistics::new_unknown (the default), not an error.

fn with_fetch(&self, limit: Option<usize>) -> Option<Arc<dyn ExecutionPlan>>

Returns a fetching variant of this ExecutionPlan node, if it supports fetch limits. Returns None otherwise.

fn static_name() -> &'static str

where Self: Sized,

Short name for the ExecutionPlan, such as ‘ParquetExec’. Like name but can be called without an instance.

fn schema(&self) -> Arc<Schema>

Get the schema for this execution plan

fn required_input_distribution(&self) -> Vec<Distribution>

Specifies the data distribution requirements for all the children for this ExecutionPlan, By default it’s [Distribution::UnspecifiedDistribution] for each child,

fn required_input_ordering(&self) -> Vec<Option<Vec<PhysicalSortRequirement>>>

Specifies the ordering required for all of the children of this ExecutionPlan.

fn maintains_input_order(&self) -> Vec<bool>

Returns false if this ExecutionPlan’s implementation may reorder rows within or between partitions.

fn benefits_from_input_partitioning(&self) -> Vec<bool>

Specifies whether the ExecutionPlan benefits from increased parallelization at its input for each child.

fn supports_limit_pushdown(&self) -> bool

Returns true if a limit can be safely pushed down through this ExecutionPlan node.