Fix API

Author: alamb

datafusion-examples/examples/custom_data_source/csv_json_opener.rs

@@ -27,7 +27,9 @@ use datafusion::{
         file_format::file_compression_type::FileCompressionType,
         listing::PartitionedFile,
         object_store::ObjectStoreUrl,
-        physical_plan::{CsvSource, FileSource, FileStream, JsonOpener, JsonSource},
+        physical_plan::{
+            CsvSource, FileSource, FileStreamBuilder, JsonOpener, JsonSource,
+        },
     },
     error::Result,
     physical_plan::metrics::ExecutionPlanMetricsSet,
@@ -81,7 +83,8 @@ async fn csv_opener() -> Result<()> {
 
     let mut result = vec![];
     let mut stream =
-        FileStream::new(&scan_config, 0, opener, &ExecutionPlanMetricsSet::new())?;
+        FileStreamBuilder::new(&scan_config, 0, opener, &ExecutionPlanMetricsSet::new())
+            .build()?;
     while let Some(batch) = stream.next().await.transpose()? {
         result.push(batch);
     }
@@ -137,12 +140,13 @@ async fn json_opener() -> Result<()> {
     .with_file(PartitionedFile::new(path.to_string(), 10))
     .build();
 
-    let mut stream = FileStream::new(
+    let mut stream = FileStreamBuilder::new(
         &scan_config,
         0,
         Arc::new(opener),
         &ExecutionPlanMetricsSet::new(),
-    )?;
+    )
+    .build()?;
     let mut result = vec![];
     while let Some(batch) = stream.next().await.transpose()? {
         result.push(batch);

datafusion/core/src/datasource/physical_plan/mod.rs

@@ -49,5 +49,5 @@ pub use datafusion_datasource::file_scan_config::{
 pub use datafusion_datasource::file_sink_config::*;
 
 pub use datafusion_datasource::file_stream::{
-    FileOpenFuture, FileOpener, FileStream, OnError,
+    FileOpenFuture, FileOpener, FileStream, FileStreamBuilder, OnError,
 };

datafusion/datasource/src/file_stream/mod.rs

@@ -306,6 +306,19 @@ impl<'a> FileStreamBuilder<'a> {
 }
 
 impl FileStream {
+    /// Create a new [`FileStream`] using a legacy [`FileOpener`].
+    ///
+    /// Prefer [`FileStreamBuilder`] for new code.
+    #[deprecated(since = "52.3.0", note = "use FileStreamBuilder instead")]
+    pub fn new(
+        config: &FileScanConfig,
+        partition: usize,
+        file_opener: Arc<dyn FileOpener>,
+        metrics: &ExecutionPlanMetricsSet,
+    ) -> Result<Self> {
+        FileStreamBuilder::new(config, partition, file_opener, metrics).build()
+    }
+
     /// Return this stream's registered shared-state id.
     fn stream_id(&self) -> Result<FileStreamId> {
         self.stream_id.ok_or_else(|| {