Interface#

classFragment:publicstd::enable_shared_from_this<Fragment>#

A granular piece of aDataset, such as an individual file.

AFragment can be read/scanned separately from other fragments. It yields a collection of RecordBatches when scanned

Note that Fragments have well defined physical schemas which are reconciled by the Datasets which contain them; these physical schemas may differ from a parentDataset’s schema and the physical schemas of sibling Fragments.

Subclassed byarrow::dataset::FileFragment,arrow::dataset::InMemoryFragment

Public Functions

Result<std::shared_ptr<Schema>>ReadPhysicalSchema()#

Return the physical schema of theFragment.

The physical schema is also called the writer schema. This method is blocking and may suffer from high latency filesystem. The schema is cached after being read once, or may be specified at construction.

virtualResult<RecordBatchGenerator>ScanBatchesAsync(conststd::shared_ptr<ScanOptions>&options)=0#

An asynchronous version of Scan.

virtualFuture<std::shared_ptr<InspectedFragment>>InspectFragment(constFragmentScanOptions*format_options,compute::ExecContext*exec_context)#

Inspect a fragment to learn basic information.

This will be called before a scan and a fragment should attach whatever information will be needed to figure out an evolution strategy. This information will then be passed to the call to BeginScan

virtualFuture<std::shared_ptr<FragmentScanner>>BeginScan(constFragmentScanRequest&request,constInspectedFragment&inspected_fragment,constFragmentScanOptions*format_options,compute::ExecContext*exec_context)#

Start a scan operation.

virtualFuture<std::optional<int64_t>>CountRows(compute::Expressionpredicate,conststd::shared_ptr<ScanOptions>&options)#

Count the number of rows in this fragment matching the filter using metadata only.

That is, this method may perform I/O, but will not load data.

If this is not possible, resolve with an empty optional. The fragment can perform I/O (e.g. to read metadata) before it deciding whether it can satisfy the request.

virtualStatusClearCachedMetadata()#

Clear any metadata that may have been cached by this object.

A fragment may typically cache metadata to speed up repeated accesses. In use cases when memory use is more critical than CPU time, calling this function can help reclaim memory.

inlineconstcompute::Expression&partition_expression()const#

An expression which evaluates to true for all data viewed by thisFragment.

Public Static Attributes

staticconstcompute::ExpressionkNoPartitionInformation#

An expression that represents no known partition information.

classDataset:publicstd::enable_shared_from_this<Dataset>#

A container of zero or more Fragments.

ADataset acts as a union of Fragments, e.g. files deeply nested in a directory. ADataset has a schema to which Fragments must align during a scan operation. This is analogous to Avro’s reader and writer schema.

Subclassed byarrow::dataset::FileSystemDataset,arrow::dataset::InMemoryDataset,arrow::dataset::UnionDataset

Public Functions

Result<std::shared_ptr<ScannerBuilder>>NewScan()#

Begin to build a new Scan operation against thisDataset.

Result<FragmentIterator>GetFragments(compute::Expressionpredicate)#

GetFragments returns an iterator of Fragments given a predicate.

Result<FragmentGenerator>GetFragmentsAsync(compute::Expressionpredicate)#

Async versions ofGetFragments.

inlineconstcompute::Expression&partition_expression()const#

An expression which evaluates to true for all data viewed by thisDataset.

May be null, which indicates no information is available.

virtualstd::stringtype_name()const=0#

The name identifying the kind ofDataset.

virtualResult<std::shared_ptr<Dataset>>ReplaceSchema(std::shared_ptr<Schema>schema)const=0#

Return a copy of thisDataset with a different schema.

The copy will view the same Fragments. If the new schema is not compatible with the original dataset’s schema then an error will be raised.

inlineDatasetEvolutionStrategy*evolution_strategy()#

Rules used by this dataset to handle schema evolution.

Partitioning#

enumclassSegmentEncoding:int8_t#

The encoding of partition segments.

Values:

enumeratorNone#

No encoding.

enumeratorUri#

Segment values are URL-encoded.

staticconstexprcharkDefaultHiveNullFallback[]="__HIVE_DEFAULT_PARTITION__"#

The default fallback used for null values in a Hive-style partitioning.

std::ostream&operator<<(std::ostream&os,SegmentEncodingsegment_encoding)#

std::stringStripPrefix(conststd::string&path,conststd::string&prefix)#

std::stringStripPrefixAndFilename(conststd::string&path,conststd::string&prefix)#

Extracts the directory and filename and removes the prefix of a path.

e.g.,StripPrefixAndFilename("/data/year=2019/c.txt","/data")->{"year=2019","c.txt"}

std::vector<std::string>StripPrefixAndFilename(conststd::vector<std::string>&paths,conststd::string&prefix)#

Vector version of StripPrefixAndFilename.

std::vector<std::string>StripPrefixAndFilename(conststd::vector<fs::FileInfo>&files,conststd::string&prefix)#

Vector version of StripPrefixAndFilename.

classPartitioning:publicarrow::util::EqualityComparable<Partitioning>#

#include <arrow/dataset/partition.h>

Interface for parsing partition expressions from string partition identifiers.

For example, the identifier “foo=5” might be parsed to an equality expression between the “foo” field and the value 5.

Some partitionings may store the field names in a metadata store instead of in file paths, for example dataset_root/2009/11/… could be used when the partition fields are “year” and “month”

Paths are consumed from left to right. Paths must be relative to the root of a partition; path prefixes must be removed before passing the path to a partitioning for parsing.

Subclassed byarrow::dataset::FunctionPartitioning,arrow::dataset::KeyValuePartitioning

Public Functions

virtualstd::stringtype_name()const=0#

The name identifying the kind of partitioning.

virtualResult<compute::Expression>Parse(conststd::string&path)const=0#

Parse a path into a partition expression.

inlineconststd::shared_ptr<Schema>&schema()const#

The partition schema.

Public Static Functions

staticstd::shared_ptr<Partitioning>Default()#

A defaultPartitioning which is aDirectoryPartitioning with an empty schema.

structPartitionedBatches#

#include <arrow/dataset/partition.h>

If the input batch shares any fields with this partitioning, produce sub-batches which satisfy mutually exclusive Expressions.

structKeyValuePartitioningOptions#

#include <arrow/dataset/partition.h>

Options for key-value based partitioning (hive/directory).

Subclassed byarrow::dataset::HivePartitioningOptions

Public Members

SegmentEncodingsegment_encoding=SegmentEncoding::Uri#

After splitting a path into components, decode the path components before parsing according to this scheme.

structPartitioningFactoryOptions#

#include <arrow/dataset/partition.h>

Options for inferring a partitioning.

Subclassed byarrow::dataset::HivePartitioningFactoryOptions

Public Members

boolinfer_dictionary=false#

When inferring a schema for partition fields, yield dictionary encoded types instead of plain.

This can be more efficient when materializing virtual columns, and Expressions parsed by the finishedPartitioning will include dictionaries of all unique inspected values for each field.

std::shared_ptr<Schema>schema#

Optionally, an expected schema can be provided, in which case inference will only check discovered fields against the schema and update internal state (such as dictionaries).

SegmentEncodingsegment_encoding=SegmentEncoding::Uri#

After splitting a path into components, decode the path components before parsing according to this scheme.

structHivePartitioningFactoryOptions:publicarrow::dataset::PartitioningFactoryOptions#

#include <arrow/dataset/partition.h>

Options for inferring a hive-style partitioning.

Public Members

std::stringnull_fallback#

The hive partitioning scheme maps null to a hard coded fallback string.

classPartitioningFactory#

#include <arrow/dataset/partition.h>

PartitioningFactory provides creation of a partitioning when the specific schema must be inferred from available paths (no explicit schema is known).

Public Functions

virtualstd::stringtype_name()const=0#

The name identifying the kind of partitioning.

virtualResult<std::shared_ptr<Schema>>Inspect(conststd::vector<std::string>&paths)=0#

Get the schema for the resultingPartitioning.

This may reset internal state, for example dictionaries of unique representations.

virtualResult<std::shared_ptr<Partitioning>>Finish(conststd::shared_ptr<Schema>&schema)const=0#

Create a partitioning using the provided schema (fields may be dropped).

classKeyValuePartitioning:publicarrow::dataset::Partitioning#

#include <arrow/dataset/partition.h>

Subclass for the common case of a partitioning which yields an equality expression for each segment.

Subclassed byarrow::dataset::DirectoryPartitioning,arrow::dataset::FilenamePartitioning,arrow::dataset::HivePartitioning

Public Functions

virtualResult<compute::Expression>Parse(conststd::string&path)constoverride#

Parse a path into a partition expression.

structKey#

#include <arrow/dataset/partition.h>

An unconverted equality expression consisting of a field name and the representation of a scalar value.

classDirectoryPartitioning:publicarrow::dataset::KeyValuePartitioning#

#include <arrow/dataset/partition.h>

DirectoryPartitioning parses one segment of a path for each field in its schema.

All fields are required, so paths passed toDirectoryPartitioning::Parse must contain segments for each field.

For example given schema<year:int16, month:int8> the path “/2009/11” would be parsed to (“year”_ == 2009 and “month”_ == 11)

Public Functions

explicitDirectoryPartitioning(std::shared_ptr<Schema>schema,ArrayVectordictionaries={},KeyValuePartitioningOptionsoptions={})#

If a field in schema is of dictionary type, the corresponding element of dictionaries must be contain the dictionary of values for that field.

inlinevirtualstd::stringtype_name()constoverride#

The name identifying the kind of partitioning.

Public Static Functions

staticstd::shared_ptr<PartitioningFactory>MakeFactory(std::vector<std::string>field_names,PartitioningFactoryOptions={})#

Create a factory for a directory partitioning.

Parameters:

field_names –[in] The names for the partition fields. Types will be inferred.

structHivePartitioningOptions:publicarrow::dataset::KeyValuePartitioningOptions#

#include <arrow/dataset/partition.h>

classHivePartitioning:publicarrow::dataset::KeyValuePartitioning#

#include <arrow/dataset/partition.h>

Multi-level, directory based partitioning originating from Apache Hive with all data files stored in the leaf directories.

Data is partitioned by static values of a particular column in the schema. Partition keys are represented in the form $key=$value in directory names.Field order is ignored, as are missing or unrecognized field names.

For example given schema<year:int16, month:int8, day:int8> the path “/day=321/ignored=3.4/year=2009” parses to (“year”_ == 2009 and “day”_ == 321)

Public Functions

inlineexplicitHivePartitioning(std::shared_ptr<Schema>schema,ArrayVectordictionaries={},std::stringnull_fallback=kDefaultHiveNullFallback)#

If a field in schema is of dictionary type, the corresponding element of dictionaries must be contain the dictionary of values for that field.

inlinevirtualstd::stringtype_name()constoverride#

The name identifying the kind of partitioning.

Public Static Functions

staticstd::shared_ptr<PartitioningFactory>MakeFactory(HivePartitioningFactoryOptions={})#

Create a factory for a hive partitioning.

classFunctionPartitioning:publicarrow::dataset::Partitioning#

#include <arrow/dataset/partition.h>

Implementation provided by lambda or other callable.

Public Functions

inlinevirtualstd::stringtype_name()constoverride#

The name identifying the kind of partitioning.

inlinevirtualResult<compute::Expression>Parse(conststd::string&path)constoverride#

Parse a path into a partition expression.

classFilenamePartitioning:publicarrow::dataset::KeyValuePartitioning#

#include <arrow/dataset/partition.h>

Public Functions

explicitFilenamePartitioning(std::shared_ptr<Schema>schema,ArrayVectordictionaries={},KeyValuePartitioningOptionsoptions={})#

Construct aFilenamePartitioning from its components.

If a field in schema is of dictionary type, the corresponding element of dictionaries must be contain the dictionary of values for that field.

inlinevirtualstd::stringtype_name()constoverride#

The name identifying the kind of partitioning.

Public Static Functions

staticstd::shared_ptr<PartitioningFactory>MakeFactory(std::vector<std::string>field_names,PartitioningFactoryOptions={})#

Create a factory for a filename partitioning.

Parameters:

field_names –[in] The names for the partition fields. Types will be inferred.

classPartitioningOrFactory#

#include <arrow/dataset/partition.h>

Either aPartitioning or aPartitioningFactory.

Public Functions

inlineconststd::shared_ptr<Partitioning>&partitioning()const#

The partitioning (if given).

inlineconststd::shared_ptr<PartitioningFactory>&factory()const#

The partition factory (if given).

Result<std::shared_ptr<Schema>>GetOrInferSchema(conststd::vector<std::string>&paths)#

Get the partition schema, inferring it with the given factory if needed.

Dataset discovery/factories#

structInspectOptions#

#include <arrow/dataset/discovery.h>

Public Members

intfragments=1#

Indicate how many fragments should be inspected to infer the unified dataset schema.

Limiting the number of fragments accessed improves the latency of the discovery process when dealing with a high number of fragments and/or high latency file systems.

The default value of1 inspects the schema of the first (in no particular order) fragment only. If the dataset has a uniform schema for all fragments, this default is the optimal value. In order to inspect all fragments and robustly unify their potentially varying schemas, set this option tokInspectAllFragments. A value of0 disables inspection of fragments altogether so only the partitioning schema will be inspected.

Field::MergeOptionsfield_merge_options=Field::MergeOptions::Defaults()#

Control how to unify types.

By default, types are merged strictly (the type must match exactly, except nulls can be merged with other types).

Public Static Attributes

staticconstexprintkInspectAllFragments=-1#

Seefragments property.

structFinishOptions#

#include <arrow/dataset/discovery.h>

Public Members

std::shared_ptr<Schema>schema=NULLPTR#

Finalize the dataset with this given schema.

If the schema is not provided, infer the schema via the Inspect, see theinspect_options property.

InspectOptionsinspect_options={}#

If the schema is not provided, it will be discovered by passing the following options toDatasetDiscovery::Inspect.

boolvalidate_fragments=false#

Indicate if the givenSchema (when specified), should be validated against the fragments’ schemas.

inspect_options will control how many fragments are checked.

classDatasetFactory#

#include <arrow/dataset/discovery.h>

DatasetFactory provides a way to inspect/discover aDataset’s expected schema before materializing saidDataset.

Subclassed byarrow::dataset::FileSystemDatasetFactory,arrow::dataset::ParquetDatasetFactory,arrow::dataset::UnionDatasetFactory

Public Functions

virtualResult<std::vector<std::shared_ptr<Schema>>>InspectSchemas(InspectOptionsoptions)=0#

Get the schemas of the Fragments andPartitioning.

Result<std::shared_ptr<Schema>>Inspect(InspectOptionsoptions={})#

Get unified schema for the resultingDataset.

Result<std::shared_ptr<Dataset>>Finish()#

Create aDataset.

Result<std::shared_ptr<Dataset>>Finish(std::shared_ptr<Schema>schema)#

Create aDataset with the given schema (seeInspectOptions::schema)

virtualResult<std::shared_ptr<Dataset>>Finish(FinishOptionsoptions)=0#

Create aDataset with the given options.

inlineconstcompute::Expression&root_partition()const#

Optional root partition for the resultingDataset.

inlineStatusSetRootPartition(compute::Expressionpartition)#

Set the root partition for the resultingDataset.

Scanning#

usingTaggedRecordBatchGenerator=std::function<Future<TaggedRecordBatch>()>#

usingTaggedRecordBatchIterator=Iterator<TaggedRecordBatch>#

usingEnumeratedRecordBatchGenerator=std::function<Future<EnumeratedRecordBatch>()>#

usingEnumeratedRecordBatchIterator=Iterator<EnumeratedRecordBatch>#

constexprint64_tkDefaultBatchSize=1<<17#

constexprint32_tkDefaultBatchReadahead=16#

constexprint32_tkDefaultFragmentReadahead=4#

constexprint32_tkDefaultBytesReadahead=1<<25#

voidSetProjection(ScanOptions*options,ProjectionDescrprojection)#

Utility method to set the projection expression and schema.

classFragmentScanOptions#

#include <arrow/dataset/dataset.h>

Per-scan options for fragment(s) in a dataset.

These options are not intrinsic to the format or fragment itself, but do affect the results of a scan. These are options which make sense to change between repeated reads of the same dataset, such as format-specific conversion options (that do not affect the schema).

Subclassed byarrow::dataset::CsvFragmentScanOptions,arrow::dataset::IpcFragmentScanOptions,arrow::dataset::JsonFragmentScanOptions,arrow::dataset::ParquetFragmentScanOptions

structScanOptions#

#include <arrow/dataset/scanner.h>

Scan-specific options, which can be changed between scans of the same dataset.

Public Functions

std::vector<FieldRef>MaterializedFields()const#

Return a vector of FieldRefs that require materialization.

This is usually the union of the fields referenced in the projection and the filter expression. Examples:

• SELECTa,bWHEREa<2&&c>1 => [“a”, “b”, “a”, “c”]

• SELECTa+b<3WHEREa>1 => [“a”, “b”, “a”]

This is needed for expression where a field may not be directly used in the final projection but is still required to evaluate the expression.

This is used byFragment implementations to apply the column sub-selection optimization.

Public Members

compute::Expressionfilter=compute::literal(true)#

A row filter (which will be pushed down to partitioning/reading if supported).

compute::Expressionprojection#

A projection expression (which can add/remove/rename columns).

std::shared_ptr<Schema>dataset_schema#

Schema with which batches will be read from fragments.

This is also known as the “reader schema” it will be used (for example) in constructing CSV file readers to identify column types for parsing. Usually only a subset of its fields (see MaterializedFields) will be materialized during a scan.

std::shared_ptr<Schema>projected_schema#

Schema of projected record batches.

This is independent of dataset_schema as its fields are derived from the projection. For example, let

dataset_schema = {“a”: int32, “b”: int32, “id”: utf8} projection = project({equal(field_ref(“a”), field_ref(“b”))}, {“a_plus_b”})

(no filter specified). In this case, the projected_schema would be

{“a_plus_b”: int32}

int64_tbatch_size=kDefaultBatchSize#

Maximum row count for scanned batches.

int32_tbatch_readahead=kDefaultBatchReadahead#

How many batches to read ahead within a fragment.

Set to 0 to disable batch readahead

Note: May not be supported by all formats Note: Will be ignored if use_threads is set to false

int32_tfragment_readahead=kDefaultFragmentReadahead#

How many files to read ahead.

Set to 0 to disable fragment readahead

Note: May not be enforced by all scanners Note: Will be ignored if use_threads is set to false

MemoryPool*pool=arrow::default_memory_pool()#

A pool from which materialized and scanned arrays will be allocated.

io::IOContextio_context#

IOContext for any IO tasks.

Note: The IOContext executor will be ignored if use_threads is set to false

arrow::internal::Executor*cpu_executor=NULLPTR#

Executor for any CPU tasks.

If null, the global CPU executor will be used

Note: The Executor will be ignored if use_threads is set to false

booluse_threads=false#

If true the scanner will scan in parallel.

Note: If true, this will use threads from both the cpu_executor and the io_context.executor Note: This must be true in order for any readahead to happen

booladd_augmented_fields=true#

If true the scanner will add augmented fields to the output schema.

boolcache_metadata=true#

Whether to cache metadata when scanning.

Fragments may typically cache metadata to speed up repeated accesses. However, in use cases where a single scan is done, or if memory use is more critical than CPU time, setting this option to false can lessen memory use.

std::shared_ptr<FragmentScanOptions>fragment_scan_options#

Fragment-specific scan options.

acero::BackpressureOptionsbackpressure=acero::BackpressureOptions::DefaultBackpressure()#

Parameters which control when the plan should pause for a slow consumer.

structScanV2Options:publicarrow::acero::ExecNodeOptions#

#include <arrow/dataset/scanner.h>

Scan-specific options, which can be changed between scans of the same dataset.

A dataset consists of one or more individual fragments. A fragment is anything that is independently scannable, often a file.

Batches from all fragments will be converted to a single schema. This unified schema is referred to as the “dataset schema” and is the output schema for this node.

Individual fragments may have schemas that are different from the dataset schema. This is sometimes referred to as the physical or fragment schema. Conversion from the fragment schema to the dataset schema is a process known as evolution.

Public Members

std::shared_ptr<Dataset>dataset#

The dataset to scan.

compute::Expressionfilter=compute::literal(true)#

A row filter.

The filter expression should be written against the dataset schema. The filter must be unbound.

This is an opportunistic pushdown filter. Filtering capabilities will vary between formats. If a format is not capable of applying the filter then it will ignore it.

Each fragment will do its best to filter the data based on the information (partitioning guarantees, statistics) available to it. If it is able to apply some filtering then it will indicate what filtering it was able to apply by attaching a guarantee to the batch.

For example, if a filter is x < 50 && y > 40 then a batch may be able to apply a guarantee x < 50. Post-scan filtering would then only need to consider y > 40 (for this specific batch). The next batch may not be able to attach any guarantee and both clauses would need to be applied to that batch.

A single guarantee-aware filtering operation should generally be applied to all resulting batches. The scan node is not responsible for this.

Fields that are referenced by the filter should be included in thecolumns

vector. The scan node will not automatically fetch fields referenced by the filter expression.

If the filter references fields that are not included in

columns this may or may not be an error, depending on the format.

See also

AddFieldsNeededForFilter

std::vector<FieldPath>columns#

The columns to scan.

This is not a simple list of top-level column indices but instead a set of paths allowing for partial selection of columns

These paths refer to the dataset schema

For example, consider the following dataset schema: schema({ field(“score”,int32()), “marker”, struct_({ field(“color”,utf8()), field(“location”, struct_({ field(“x”,float64()), field(“y”,float64()) }) }) })

Ifcolumns is {{0}, {1,1,0}} then the output schema is: schema({field(“score”,int32()), field(“x”,float64())})

Ifcolumns is {{1,1,1}, {1,1}} then the output schema is: schema({ field(“y”,float64()), field(“location”, struct_({ field(“x”,float64()), field(“y”,float64()) }) })

int32_ttarget_bytes_readahead=kDefaultBytesReadahead#

Target number of bytes to read ahead in a fragment.

This limit involves some amount of estimation. Formats typically only know batch boundaries in terms of rows (not decoded bytes) and so an estimation must be done to guess the average row size. Other formats like CSV and JSON must make even more generalized guesses.

This is a best-effort guide. Some formats may need to read ahead further, for example, if scanning a parquet file that has batches with 100MiB of data then the actual readahead will be at least 100MiB

Set to 0 to disable readahead. When disabled, the scanner will read the dataset one batch at a time

This limit applies across all fragments. If the limit is 32MiB and the fragment readahead allows for 20 fragments to be read at once then the total readahead will still be 32MiB and NOT 20 * 32MiB.

int32_tfragment_readahead=kDefaultFragmentReadahead#

Number of fragments to read ahead.

Higher readahead will potentially lead to more efficient I/O but will lead to the scan operation using more RAM. The default is fairly conservative and designed for fast local disks (or slow local spinning disks which cannot handle much parallelism anyways). When using a highly parallel remote filesystem you will likely want to increase these values.

Set to 0 to disable fragment readahead. When disabled the dataset will be scanned one fragment at a time.

constFragmentScanOptions*format_options=NULLPTR#

Options specific to the file format.

Public Static Functions

staticstd::vector<FieldPath>AllColumns(constSchema&dataset_schema)#

Utility method to get a selection representing all columns in a dataset.

staticStatusAddFieldsNeededForFilter(ScanV2Options*options)#

Utility method to add fields needed for the current filter.

This method adds any fields that are needed byfilter which are not already included in the list of columns. Any new fields added will be added to the end in no particular order.

structProjectionDescr#

#include <arrow/dataset/scanner.h>

Describes a projection.

Public Members

compute::Expressionexpression#

The projection expression itself This expression must be a call to make_struct.

std::shared_ptr<Schema>schema#

The output schema of the projection.

This can be calculated from the input schema and the expression but it is cached here for convenience.

Public Static Functions

staticResult<ProjectionDescr>FromStructExpression(constcompute::Expression&expression,constSchema&dataset_schema)#

Create aProjectionDescr by binding an expression to the dataset schema.

expression must return a struct type

staticResult<ProjectionDescr>FromExpressions(std::vector<compute::Expression>exprs,std::vector<std::string>names,constSchema&dataset_schema)#

Create aProjectionDescr from expressions/names for each field.

staticResult<ProjectionDescr>FromNames(std::vector<std::string>names,constSchema&dataset_schema,booladd_augmented_fields=true)#

Create a default projection referencing fields in the dataset schema.

staticResult<ProjectionDescr>Default(constSchema&dataset_schema,booladd_augmented_fields=true)#

Make a projection that projects every field in the dataset schema.

structTaggedRecordBatch#

#include <arrow/dataset/scanner.h>

Combines a record batch with the fragment that the record batch originated from.

Knowing the source fragment can be useful for debugging & understanding loaded data

structEnumeratedRecordBatch#

#include <arrow/dataset/scanner.h>

Combines a tagged batch with positional information.

This is returned when scanning batches in an unordered fashion. This information is needed if you ever want to reassemble the batches in order

classScanner#

#include <arrow/dataset/scanner.h>

A scanner glues together several dataset classes to load in data.

The dataset contains a collection of fragments and partitioning rules.

The fragments identify independently loadable units of data (i.e. each fragment has a potentially unique schema and possibly even format. It should be possible to read fragments in parallel if desired).

The fragment’s format contains the logic necessary to actually create a task to load the fragment into memory. That task may or may not support parallel execution of its own.

The scanner is then responsible for creating scan tasks from every fragment in the dataset and (potentially) sequencing the loaded record batches together.

The scanner should not buffer the entire dataset in memory (unless asked) instead yielding record batches as soon as they are ready to scan. Various readahead properties control how much data is allowed to be scanned before pausing to let a slow consumer catchup.

Today the scanner also handles projection & filtering although that may change in the future.

Public Functions

virtualStatusScan(std::function<Status(TaggedRecordBatch)>visitor)=0#

Apply a visitor to eachRecordBatch as it is scanned.

If multiple threads are used (via use_threads), the visitor will be invoked from those threads and is responsible for any synchronization.

virtualResult<std::shared_ptr<Table>>ToTable()=0#

Convert aScanner into aTable.

Use this convenience utility with care. This will serially materialize the Scan result in memory before creating theTable.

virtualResult<TaggedRecordBatchIterator>ScanBatches()=0#

Scan the dataset into a stream of record batches.

Each batch is tagged with the fragment it originated from. The batches will arrive in order. The order of fragments is determined by the dataset.

Note: The scanner will perform some readahead but will avoid materializing too much in memory (this is goverended by the readahead options and use_threads option). If the readahead queue fills up then I/O will pause until the calling thread catches up.

virtualResult<EnumeratedRecordBatchIterator>ScanBatchesUnordered()=0#

Scan the dataset into a stream of record batches.

Unlike ScanBatches this method may allow record batches to be returned out of order. This allows for more efficient scanning: some fragments may be accessed more quickly than others (e.g. may be cached in RAM or just happen to get scheduled earlier by the I/O)

To make up for the out-of-order iteration each batch is further tagged with positional information.

virtualResult<std::shared_ptr<Table>>TakeRows(constArray&indices)=0#

A convenience to synchronously load the given rows by index.

Will only consume as many batches as needed fromScanBatches().

virtualResult<std::shared_ptr<Table>>Head(int64_tnum_rows)=0#

Get the first N rows.

virtualResult<int64_t>CountRows()=0#

Count rows matching a predicate.

This method will push down the predicate and compute the result based on fragment metadata if possible.

virtualResult<std::shared_ptr<RecordBatchReader>>ToRecordBatchReader()=0#

Convert theScanner to aRecordBatchReader so it can be easily used with APIs that expect a reader.

inlineconststd::shared_ptr<ScanOptions>&options()const#

Get the options for this scan.

virtualconststd::shared_ptr<Dataset>&dataset()const=0#

Get the dataset that this scanner will scan.

classScannerBuilder#

#include <arrow/dataset/scanner.h>

ScannerBuilder is a factory class to construct aScanner.

It is used to pass information, notably a potential filter expression and a subset of columns to materialize.

Public Functions

StatusProject(std::vector<std::string>columns)#

Set the subset of columns to materialize.

Columns which are not referenced may not be read from fragments.

Parameters:

columns –[in] list of columns to project. Order and duplicates will be preserved.

Returns:

Failure if any column name does not exists in the dataset’sSchema.

StatusProject(std::vector<compute::Expression>exprs,std::vector<std::string>names)#

Set expressions which will be evaluated to produce the materialized columns.

Columns which are not referenced may not be read from fragments.

Parameters:

• exprs –[in] expressions to evaluate to produce columns.

• names –[in] list of names for the resulting columns.

Returns:

Failure if any referenced column does not exists in the dataset’sSchema.

StatusFilter(constcompute::Expression&filter)#

Set the filter expression to return only rows matching the filter.

The predicate will be passed down to Sources and corresponding Fragments to exploit predicate pushdown if possible using partition information orFragment internal metadata, e.g. Parquet statistics. Columns which are not referenced may not be read from fragments.

Parameters:

filter –[in] expression to filter rows with.

Returns:

Failure if any referenced columns does not exist in the dataset’sSchema.

StatusUseThreads(booluse_threads=true)#

Indicate if theScanner should make use of the available ThreadPool found inScanOptions;.

StatusCacheMetadata(boolcache_metadata=true)#

Indicate if metadata should be cached when scanning.

Fragments may typically cache metadata to speed up repeated accesses. However, in use cases where a single scan is done, or if memory use is more critical than CPU time, setting this option to false can lessen memory use.

StatusBatchSize(int64_tbatch_size)#

Set the maximum number of rows perRecordBatch.

This option provides a control limiting the memory owned by anyRecordBatch.

Parameters:

batch_size –[in] the maximum number of rows.

Returns:

An error if the number for batch is not greater than 0.

StatusBatchReadahead(int32_tbatch_readahead)#

Set the number of batches to read ahead within a fragment.

This option provides a control on the RAM vs I/O tradeoff. It might not be supported by all file formats, in which case it will simply be ignored.

Parameters:

batch_readahead –[in] How many batches to read ahead within a fragment

Returns:

an error if this number is less than 0.

StatusFragmentReadahead(int32_tfragment_readahead)#

Set the number of fragments to read ahead.

This option provides a control on the RAM vs I/O tradeoff.

Parameters:

fragment_readahead –[in] How many fragments to read ahead

Returns:

an error if this number is less than 0.

StatusPool(MemoryPool*pool)#

Set the pool from which materialized and scanned arrays will be allocated.

StatusFragmentScanOptions(std::shared_ptr<FragmentScanOptions>fragment_scan_options)#

Set fragment-specific scan options.

StatusBackpressure(acero::BackpressureOptionsbackpressure)#

Override default backpressure configuration.

Result<std::shared_ptr<ScanOptions>>GetScanOptions()#

Return the current scan options for the builder.

Result<std::shared_ptr<Scanner>>Finish()#

Return the constructed now-immutableScanner object.

Public Static Functions

staticstd::shared_ptr<ScannerBuilder>FromRecordBatchReader(std::shared_ptr<RecordBatchReader>reader)#

Make a scanner from a record batch reader.

The resulting scanner can be scanned only once. This is intended to support writing data from streaming sources or other sources that can be iterated only once.

classScanNodeOptions:publicarrow::acero::ExecNodeOptions#

#include <arrow/dataset/scanner.h>

Construct a source ExecNode which yields batches from a dataset scan.

Does not construct associated filter or project nodes.

Batches are yielded sequentially, like single-threaded, when require_sequenced_output=true.

Yielded batches will be augmented with fragment/batch indices when implicit_ordering=true to enable stable ordering for simple ExecPlans.

Concrete implementations#

classInMemoryFragment:publicarrow::dataset::Fragment#

#include <arrow/dataset/dataset.h>

A trivialFragment that yields ScanTask out of a fixed set ofRecordBatch.

Public Functions

virtualResult<RecordBatchGenerator>ScanBatchesAsync(conststd::shared_ptr<ScanOptions>&options)override#

An asynchronous version of Scan.

virtualFuture<std::optional<int64_t>>CountRows(compute::Expressionpredicate,conststd::shared_ptr<ScanOptions>&options)override#

Count the number of rows in this fragment matching the filter using metadata only.

That is, this method may perform I/O, but will not load data.

If this is not possible, resolve with an empty optional. The fragment can perform I/O (e.g. to read metadata) before it deciding whether it can satisfy the request.

virtualFuture<std::shared_ptr<InspectedFragment>>InspectFragment(constFragmentScanOptions*format_options,compute::ExecContext*exec_context)override#

Inspect a fragment to learn basic information.

This will be called before a scan and a fragment should attach whatever information will be needed to figure out an evolution strategy. This information will then be passed to the call to BeginScan

virtualFuture<std::shared_ptr<FragmentScanner>>BeginScan(constFragmentScanRequest&request,constInspectedFragment&inspected_fragment,constFragmentScanOptions*format_options,compute::ExecContext*exec_context)override#

Start a scan operation.

classInMemoryDataset:publicarrow::dataset::Dataset#

#include <arrow/dataset/dataset.h>

A Source which yields fragments wrapping a stream of record batches.

The record batches must match the schema provided to the source at construction.

Public Functions

inlineInMemoryDataset(std::shared_ptr<Schema>schema,std::shared_ptr<RecordBatchGenerator>get_batches)#

Construct a dataset from a schema and a factory of record batch iterators.

InMemoryDataset(std::shared_ptr<Schema>schema,RecordBatchVectorbatches)#

Convenience constructor taking a fixed list of batches.

explicitInMemoryDataset(std::shared_ptr<Table>table)#

Convenience constructor taking aTable.

inlinevirtualstd::stringtype_name()constoverride#

The name identifying the kind ofDataset.

virtualResult<std::shared_ptr<Dataset>>ReplaceSchema(std::shared_ptr<Schema>schema)constoverride#

Return a copy of thisDataset with a different schema.

The copy will view the same Fragments. If the new schema is not compatible with the original dataset’s schema then an error will be raised.

classRecordBatchGenerator#

#include <arrow/dataset/dataset.h>

classUnionDataset:publicarrow::dataset::Dataset#

#include <arrow/dataset/dataset.h>

ADataset wrapping child Datasets.

Public Functions

inlinevirtualstd::stringtype_name()constoverride#

The name identifying the kind ofDataset.

virtualResult<std::shared_ptr<Dataset>>ReplaceSchema(std::shared_ptr<Schema>schema)constoverride#

Return a copy of thisDataset with a different schema.

The copy will view the same Fragments. If the new schema is not compatible with the original dataset’s schema then an error will be raised.

Public Static Functions

staticResult<std::shared_ptr<UnionDataset>>Make(std::shared_ptr<Schema>schema,DatasetVectorchildren)#

Construct aUnionDataset wrapping child Datasets.

Parameters:

• schema –[in] the schema of the resulting dataset.

• children –[in] one or more child Datasets. Their schemas must be identical to schema.

classUnionDatasetFactory:publicarrow::dataset::DatasetFactory#

#include <arrow/dataset/discovery.h>

DatasetFactory provides a way to inspect/discover aDataset’s expected schema before materialization.

Public Functions

inlineconststd::vector<std::shared_ptr<DatasetFactory>>&factories()const#

Return the list of childDatasetFactory.

virtualResult<std::vector<std::shared_ptr<Schema>>>InspectSchemas(InspectOptionsoptions)override#

Get the schemas of the Datasets.

Instead of applying options globally, it applies at each child factory. This will not respectoptions.fragments exactly, but will respect the spirit of peeking the first fragments or all of them.

virtualResult<std::shared_ptr<Dataset>>Finish(FinishOptionsoptions)override#

Create aDataset.