Skip to content

Decouple the FileSource trait from the concrete FileScanConfig DataSource #23467

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
mitchsw wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Decouple the FileSource trait from the concrete FileScanConfig DataSource #23467

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 5 additions & 8 deletions datafusion-examples/examples/custom_data_source/csv_json_opener.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ use datafusion::{
listing::PartitionedFile,
object_store::ObjectStoreUrl,
physical_plan::{
CsvSource, FileSource, FileStreamBuilder, JsonOpener, JsonSource,
CsvSource, FileSourceArgs, FileStreamBuilder, JsonOpener, JsonSource,
},
},
error::Result,
Expand Down Expand Up @@ -66,20 +66,17 @@ async fn csv_opener() -> Result<()> {

let source = CsvSource::new(Arc::clone(&schema))
.with_csv_options(options)
.with_comment(Some(b'#'))
.with_batch_size(8192);
.with_comment(Some(b'#'));

let scan_config =
FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), source)
FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), source.into())
.with_projection_indices(Some(vec![0, 1]))?
.with_limit(Some(5))
.with_file(PartitionedFile::new(csv_path.display().to_string(), 10))
.build();

let opener =
scan_config
.file_source()
.create_file_opener(object_store, &scan_config, 0)?;
let args = FileSourceArgs::new(object_store).with_batch_size(8192);
let opener = scan_config.file_source().create_file_opener(&args, 0)?;

let mut result = vec![];
let metrics = ExecutionPlanMetricsSet::new();
Expand Down
2 changes: 1 addition & 1 deletion datafusion/core/src/datasource/physical_plan/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ pub use json::{JsonOpener, JsonSource};

pub use arrow::{ArrowOpener, ArrowSource};
pub use csv::{CsvOpener, CsvSource};
pub use datafusion_datasource::file::FileSource;
pub use datafusion_datasource::file::{FileSource, FileSourceArgs};
pub use datafusion_datasource::file_groups::FileGroup;
pub use datafusion_datasource::file_groups::FileGroupPartitioner;
pub use datafusion_datasource::file_scan_config::{
Expand Down
24 changes: 8 additions & 16 deletions datafusion/core/tests/physical_optimizer/pushdown_utils.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,12 @@ use arrow::{array::RecordBatch, compute::concat_batches};
use datafusion::{datasource::object_store::ObjectStoreUrl, physical_plan::PhysicalExpr};
use datafusion_common::{Result, config::ConfigOptions, internal_err};
use datafusion_datasource::{
PartitionedFile, file::FileSource, file_scan_config::FileScanConfig,
file_scan_config::FileScanConfigBuilder, file_stream::FileOpenFuture,
file_stream::FileOpener, source::DataSourceExec,
PartitionedFile,
file::{FileSource, FileSourceArgs},
file_scan_config::FileScanConfigBuilder,
file_stream::FileOpenFuture,
file_stream::FileOpener,
source::DataSourceExec,
};
use datafusion_physical_expr::projection::ProjectionExprs;
use datafusion_physical_expr_common::physical_expr::fmt_sql;
Expand All @@ -40,7 +43,6 @@ use datafusion_physical_plan::{
};
use futures::StreamExt;
use futures::{FutureExt, Stream};
use object_store::ObjectStore;
use std::{
fmt::{Display, Formatter},
pin::Pin,
Expand Down Expand Up @@ -102,7 +104,6 @@ impl FileOpener for TestOpener {
pub struct TestSource {
support: bool,
predicate: Option<Arc<dyn PhysicalExpr>>,
batch_size: Option<usize>,
batches: Vec<RecordBatch>,
metrics: ExecutionPlanMetricsSet,
projection: Option<ProjectionExprs>,
Expand All @@ -117,7 +118,6 @@ impl TestSource {
metrics: ExecutionPlanMetricsSet::new(),
batches,
predicate: None,
batch_size: None,
projection: None,
table_schema,
}
Expand All @@ -127,13 +127,12 @@ impl TestSource {
impl FileSource for TestSource {
fn create_file_opener(
&self,
_object_store: Arc<dyn ObjectStore>,
_base_config: &FileScanConfig,
args: &FileSourceArgs,
_partition: usize,
) -> Result<Arc<dyn FileOpener>> {
Ok(Arc::new(TestOpener {
batches: self.batches.clone(),
batch_size: self.batch_size,
batch_size: Some(args.batch_size),
projection: self.projection.clone(),
predicate: self.predicate.clone(),
}))
Expand All @@ -143,13 +142,6 @@ impl FileSource for TestSource {
self.predicate.clone()
}

fn with_batch_size(&self, batch_size: usize) -> Arc<dyn FileSource> {
Arc::new(TestSource {
batch_size: Some(batch_size),
..self.clone()
})
}

fn metrics(&self) -> &ExecutionPlanMetricsSet {
&self.metrics
}
Expand Down
37 changes: 9 additions & 28 deletions datafusion/datasource-arrow/src/source.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ use arrow::ipc::reader::{FileDecoder, FileReader, StreamReader};
use datafusion_common::error::Result;
use datafusion_common::exec_datafusion_err;
use datafusion_datasource::PartitionedFile;
use datafusion_datasource::file::FileSource;
use datafusion_datasource::file::{FileSource, FileSourceArgs};
use datafusion_datasource::file_scan_config::FileScanConfig;
use datafusion_datasource::projection::{ProjectionOpener, SplitProjection};
use datafusion_physical_expr_common::sort_expr::LexOrdering;
Expand Down Expand Up @@ -290,10 +290,10 @@ impl ArrowSource {
impl FileSource for ArrowSource {
fn create_file_opener(
&self,
object_store: Arc<dyn ObjectStore>,
_base_config: &FileScanConfig,
args: &FileSourceArgs,
_partition: usize,
) -> Result<Arc<dyn FileOpener>> {
let object_store = Arc::clone(&args.object_store);
let split_projection = self.projection.clone();

let opener: Arc<dyn FileOpener> = match self.format {
Expand All @@ -313,10 +313,6 @@ impl FileSource for ArrowSource {
)
}

fn with_batch_size(&self, _batch_size: usize) -> Arc<dyn FileSource> {
Arc::new(Self { ..self.clone() })
}

fn metrics(&self) -> &ExecutionPlanMetricsSet {
&self.metrics
}
Expand Down Expand Up @@ -484,13 +480,8 @@ mod tests {
Arc::new(ArrowSource::new_file_source(schema))
};

let scan_config = FileScanConfigBuilder::new(
ObjectStoreUrl::local_filesystem(),
source.clone(),
)
.build();

let file_opener = source.create_file_opener(object_store, &scan_config, 0)?;
let file_opener =
source.create_file_opener(&FileSourceArgs::new(object_store), 0)?;
let mut stream = file_opener.open(partitioned_file)?.await?;

assert!(stream.next().await.is_some());
Expand Down Expand Up @@ -526,13 +517,8 @@ mod tests {

let source = Arc::new(ArrowSource::new_file_source(schema));

let scan_config = FileScanConfigBuilder::new(
ObjectStoreUrl::local_filesystem(),
source.clone(),
)
.build();

let file_opener = source.create_file_opener(object_store, &scan_config, 0)?;
let file_opener =
source.create_file_opener(&FileSourceArgs::new(object_store), 0)?;
let mut stream = file_opener.open(partitioned_file)?.await?;

assert!(stream.next().await.is_some());
Expand Down Expand Up @@ -567,13 +553,8 @@ mod tests {

let source = Arc::new(ArrowSource::new_stream_file_source(schema));

let scan_config = FileScanConfigBuilder::new(
ObjectStoreUrl::local_filesystem(),
source.clone(),
)
.build();

let file_opener = source.create_file_opener(object_store, &scan_config, 0)?;
let file_opener =
source.create_file_opener(&FileSourceArgs::new(object_store), 0)?;
let result = file_opener.open(partitioned_file);
assert!(result.is_err());

Expand Down
53 changes: 28 additions & 25 deletions datafusion/datasource-avro/src/source.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,21 +23,17 @@ use arrow::datatypes::{Schema, SchemaRef};
use arrow_avro::reader::{Reader, ReaderBuilder};
use datafusion_common::error::Result;
use datafusion_datasource::TableSchema;
use datafusion_datasource::file::FileSource;
use datafusion_datasource::file_scan_config::FileScanConfig;
use datafusion_datasource::file::{FileSource, FileSourceArgs};
use datafusion_datasource::file_stream::FileOpener;
use datafusion_datasource::projection::{ProjectionOpener, SplitProjection};
use datafusion_physical_expr_adapter::BatchAdapterFactory;
use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
use datafusion_physical_plan::projection::ProjectionExprs;

use object_store::ObjectStore;

/// AvroSource holds the extra configuration that is necessary for opening avro files
#[derive(Clone)]
pub struct AvroSource {
table_schema: TableSchema,
batch_size: Option<usize>,
projection: SplitProjection,
metrics: ExecutionPlanMetricsSet,
}
Expand All @@ -49,7 +45,6 @@ impl AvroSource {
Self {
projection: SplitProjection::unprojected(&table_schema),
table_schema,
batch_size: None,
metrics: ExecutionPlanMetricsSet::new(),
}
}
Expand All @@ -58,9 +53,9 @@ impl AvroSource {
&self,
reader: R,
projection: Option<Vec<usize>>,
batch_size: usize,
) -> Result<Reader<R>> {
let mut builder = ReaderBuilder::new()
.with_batch_size(self.batch_size.expect("Batch size must set before open"));
let mut builder = ReaderBuilder::new().with_batch_size(batch_size);
if let Some(projection) = projection {
builder = builder.with_projection(projection);
}
Expand Down Expand Up @@ -114,13 +109,13 @@ impl AvroSource {
impl FileSource for AvroSource {
fn create_file_opener(
&self,
object_store: Arc<dyn ObjectStore>,
_base_config: &FileScanConfig,
args: &FileSourceArgs,
_partition: usize,
) -> Result<Arc<dyn FileOpener>> {
let mut opener = Arc::new(private::AvroOpener {
config: Arc::new(self.clone()),
object_store,
object_store: Arc::clone(&args.object_store),
batch_size: args.batch_size,
}) as Arc<dyn FileOpener>;
opener = ProjectionOpener::try_new(
self.projection.clone(),
Expand All @@ -134,12 +129,6 @@ impl FileSource for AvroSource {
&self.table_schema
}

fn with_batch_size(&self, batch_size: usize) -> Arc<dyn FileSource> {
let mut conf = self.clone();
conf.batch_size = Some(batch_size);
Arc::new(conf)
}

fn try_pushdown_projection(
&self,
projection: &ProjectionExprs,
Expand Down Expand Up @@ -183,12 +172,14 @@ mod private {
pub struct AvroOpener {
pub config: Arc<AvroSource>,
pub object_store: Arc<dyn ObjectStore>,
pub batch_size: usize,
}

impl FileOpener for AvroOpener {
fn open(&self, partitioned_file: PartitionedFile) -> Result<FileOpenFuture> {
let object_store = Arc::clone(&self.object_store);
let config = Arc::clone(&self.config);
let batch_size = self.batch_size;
let projected_file_schema = config.projected_file_schema();

Ok(Box::pin(async move {
Expand All @@ -199,15 +190,21 @@ mod private {
GetResultPayload::File(mut file, _) => {
// Probe the writer schema first so logical projected columns can be
// translated to the writer-schema ordinals expected by `arrow-avro`.
let probe_reader =
config.open(BufReader::new(file.try_clone()?), None)?;
let probe_reader = config.open(
BufReader::new(file.try_clone()?),
None,
batch_size,
)?;
let writer_projection = config.writer_projection_for_schema(
probe_reader.schema().as_ref(),
projected_file_schema.as_ref(),
);
file.rewind()?;
let reader =
config.open(BufReader::new(file), writer_projection)?;
let reader = config.open(
BufReader::new(file),
writer_projection,
batch_size,
)?;
let batch_adapter =
BatchAdapterFactory::new(Arc::clone(&projected_file_schema))
.make_adapter(&reader.schema())?;
Expand All @@ -222,14 +219,20 @@ mod private {
let bytes = r.bytes().await?;
// As above, inspect the writer schema before constructing the real
// reader so `with_projection` can use writer-schema ordinals.
let probe_reader =
config.open(BufReader::new(bytes.clone().reader()), None)?;
let probe_reader = config.open(
BufReader::new(bytes.clone().reader()),
None,
batch_size,
)?;
let writer_projection = config.writer_projection_for_schema(
probe_reader.schema().as_ref(),
projected_file_schema.as_ref(),
);
let reader = config
.open(BufReader::new(bytes.reader()), writer_projection)?;
let reader = config.open(
BufReader::new(bytes.reader()),
writer_projection,
batch_size,
)?;
let batch_adapter =
BatchAdapterFactory::new(Arc::clone(&projected_file_schema))
.make_adapter(&reader.schema())?;
Expand Down
Loading