MockFileSystem and use Results [skip ci]

cpp/src/arrow/filesystem/filesystem.h

@@ -155,6 +155,8 @@ class ARROW_EXPORT FileSystem {
  public:
   virtual ~FileSystem();
 
+  virtual std::string type() const = 0;
+
   /// Get statistics for the given target.
   ///
   /// Any symlink is automatically dereferenced, recursively.
@@ -243,6 +245,8 @@ class ARROW_EXPORT SubTreeFileSystem : public FileSystem {
                              std::shared_ptr<FileSystem> base_fs);
   ~SubTreeFileSystem() override;
 
+  std::string type() const override { return "subtree"; }
+
   /// \cond FALSE
   using FileSystem::GetTargetStats;
   /// \endcond
@@ -289,6 +293,8 @@ class ARROW_EXPORT SlowFileSystem : public FileSystem {
   SlowFileSystem(std::shared_ptr<FileSystem> base_fs, double average_latency,
                  int32_t seed);
 
+  std::string type() const override { return "slow"; }
+
   using FileSystem::GetTargetStats;
   Result<FileStats> GetTargetStats(const std::string& path) override;
   Result<std::vector<FileStats>> GetTargetStats(const Selector& select) override;

cpp/src/arrow/filesystem/hdfs.h

@@ -61,6 +61,8 @@ class ARROW_EXPORT HadoopFileSystem : public FileSystem {
  public:
   ~HadoopFileSystem() override;
 
+  std::string type() const override { return "hdfs"; }
+
   /// \cond FALSE
   using FileSystem::GetTargetStats;
   /// \endcond

cpp/src/arrow/filesystem/localfs.h

@@ -48,6 +48,8 @@ class ARROW_EXPORT LocalFileSystem : public FileSystem {
   explicit LocalFileSystem(const LocalFileSystemOptions&);
   ~LocalFileSystem() override;
 
+  std::string type() const override { return "local"; }
+
   /// \cond FALSE
   using FileSystem::GetTargetStats;
   /// \endcond

cpp/src/arrow/filesystem/mockfs.h

@@ -60,6 +60,8 @@ class ARROW_EXPORT MockFileSystem : public FileSystem {
   explicit MockFileSystem(TimePoint current_time);
   ~MockFileSystem() override;
 
+  std::string type() const override { return "mock"; }
+
   // XXX It's not very practical to have to explicitly declare inheritance
   // of default overrides.
   using FileSystem::GetTargetStats;

cpp/src/arrow/filesystem/s3fs.h

@@ -79,6 +79,8 @@ class ARROW_EXPORT S3FileSystem : public FileSystem {
  public:
   ~S3FileSystem() override;
 
+  std::string type() const override { return "s3"; }
+
   /// \cond FALSE
   using FileSystem::GetTargetStats;
   /// \endcond

cpp/src/arrow/python/common.h

@@ -66,7 +66,7 @@ inline Status CheckPyError(StatusCode code = StatusCode::UnknownError) {
 template <class T>
 T GetResultValue(Result<T> result) {
   if (ARROW_PREDICT_TRUE(result.ok())) {
-    return *std::move(result);
+    return std::move(*result);
   } else {
     int r = internal::check_status(result.status());  // takes the GIL
     assert(r == -1);                                  // should have errored out

python/pyarrow/_dataset.pyx

@@ -29,6 +29,7 @@ from pyarrow.lib cimport *
 # from pyarrow.lib import ArrowException
 # from pyarrow.lib import as_buffer
 # from pyarrow.lib cimport RecordBatch, MemoryPool, Schema, check_status
+from pyarrow._fs cimport FileSystem
 from pyarrow.includes.libarrow_dataset cimport *
 
 
@@ -90,7 +91,17 @@ cdef class ParquetWriterOptions(FileWriteOptions):
 
 
 cdef class FileFormat:
-    pass
+
+    cdef:
+        shared_ptr[CFileFormat] wrapped
+        CFileFormat* format
+
+    cdef void init(self, const shared_ptr[CFileFormat]& sp):
+        self.wrapped = sp
+        self.format = sp.get()
+
+    # @property
+    # def name(self):
 
 
 cdef class CsvFileFormat(FileFormat):
@@ -114,6 +125,30 @@ cdef class FileSource:
     cdef:
         shared_ptr[CFileSource] wrapped
 
+    def equals(self, FileSource other):
+        return self.wrapped.get() == other.wrapped.get()
+
+    def __eq__(self, other):
+        try:
+            return self.equals(other)
+        except TypeError:
+            return NotImplementedError
+
+    @property
+    def compression(self):
+        return self.wrapped.get().compression()
+
+    @property
+    def path(self):
+        return frombytes(self.wrapped.get().path())
+
+    # @property
+    # def fs(self):
+    #     FileSystem.wrap(self.wrapped.get().filesystem())
+
+    # def open(self):
+    #     CStatus Open(shared_ptr[CRandomAccessFile]* out)
+
 
 cdef class DataFragment:
     cdef:
@@ -132,15 +167,18 @@ cdef class DataFragment:
 
     def scan(self, ScanContext context=None):
         cdef:
-            CScanTaskIterator task_iterator
-            unique_ptr[CScanTask] task
+            CResult[CScanTaskIterator] iterator_result
+            CScanTaskIterator iterator
+            CScanTaskPtr task
 
         context = context or ScanContext()
-        check_status(self.fragment.Scan(context.wrapped, &task_iterator))
+
+        iterator_result = self.fragment.Scan(context.unwrap())
+        iterator = move(GetResultValue(iterator_result))
 
         while True:
-            task_iterator.Next(&task)
-            if task == nullptr:
+            iterator.Next(&task)
+            if task.get() == nullptr:
                 raise StopIteration()
             else:
                 yield ScanTask.wrap(task)
@@ -224,23 +262,19 @@ cdef class DataSource:
 
     def fragments(self, ScanOptions options=None):
         cdef:
-            CDataFragmentIterator fragment_iterator
-            shared_ptr[CDataFragment] fragment
+            CDataFragmentIterator it
+            CDataFragmentPtr fragment
 
         options = options or ScanOptions()
-
-        # NOTE(kszucs): shouldn't we return Status here?
-        fragment_iterator = self.source.GetFragments(options.wrapped)
+        it = self.source.GetFragments(options.wrapped)
 
         while True:
-            fragment_iterator.Next(&fragment)
-            if fragment == nullptr:
+            it.Next(&fragment)
+            if fragment.get() == nullptr:
                 raise StopIteration()
             else:
                 yield DataFragment.wrap(fragment)
 
-    get_fragments = fragments
-
 
 cdef class SimpleDataSource(DataSource):
 
@@ -321,18 +355,18 @@ cdef class Dataset:
     def __init__(self, data_sources, Schema schema=None):
         cdef:
             DataSource source
-            vector[shared_ptr[CDataSource]] sources
-            shared_ptr[CDataset] dataset
+            CDataSourceVector sources
+            CResult[CDatasetPtr] result
 
         for source in data_sources:
             sources.push_back(source.wrapped)
 
-        check_status(CDataset.Make(sources, schema.sp_schema, &dataset))
+        result = CDataset.Make(sources, schema.sp_schema)
         # if schema is None:
         #     dataset = make_shared[CDataset](sources)
         # else:
         #     dataset = make_shared[CDataset](sources, schema.sp_schema)
-        self.init(dataset)
+        self.init(GetResultValue(result))
 
     cdef void init(self, const shared_ptr[CDataset]& sp):
         self.wrapped = sp
@@ -340,12 +374,16 @@ cdef class Dataset:
 
     def new_scan(self):
         cdef:
-            unique_ptr[CScannerBuilder] builder
-            unique_ptr[CScanner] scanner
+            CResult[CScannerBuilderPtr] builder_result
+            CResult[CScannerPtr] scanner_result
+            CScannerBuilderPtr builder
+            CScannerPtr scanner
 
-        # TODO(kszucs): add other options like filter, schema etc.
-        check_status(self.dataset.NewScan(&builder))
-        check_status(builder.get().Finish(&scanner))
+        builder_result = self.dataset.NewScan()
+        builder = GetResultValue(builder_result)
+
+        scanner_result = builder.get().Finish()
+        scanner = GetResultValue(scanner_result)
 
         return Scanner.wrap(scanner)
 
@@ -386,55 +424,62 @@ cdef class ScanOptions:
 cdef class ScanContext:
 
     cdef:
-        shared_ptr[CScanContext] wrapped
+        CScanContextPtr wrapped
         CScanContext* context
 
     def __init__(self, MemoryPool memory_pool=None):
-        cdef shared_ptr[CScanContext] context
+        cdef CScanContextPtr context
 
         if memory_pool is not None:
             context.get().pool = memory_pool.pool
 
         self.init(context)
 
-    cdef init(self, const shared_ptr[CScanContext]& sp):
+    cdef init(self, CScanContextPtr& sp):
         self.wrapped = sp
         self.context = sp.get()
 
     @staticmethod
-    cdef ScanContext wrap(const shared_ptr[CScanContext]& sp):
+    cdef ScanContext wrap(CScanContextPtr& sp):
         cdef ScanContext self = ScanContext.__new__(ScanContext)
         self.init(sp)
         return self
 
-    cdef inline shared_ptr[CScanContext] unwrap(self):
+    cdef inline CScanContextPtr unwrap(self):
         return self.wrapped
 
 
 cdef class ScanTask:
 
     cdef:
-        unique_ptr[CScanTask] wrapped
+        CScanTaskPtr wrapped
         CScanTask* task
 
-    cdef init(self, unique_ptr[CScanTask]& up):
-        self.wrapped = move(up)
+    cdef init(self, CScanTaskPtr& sp):
+        self.wrapped = sp
         self.task = self.wrapped.get()
 
     @staticmethod
-    cdef wrap(unique_ptr[CScanTask]& up):
+    cdef wrap(CScanTaskPtr& sp):
         cdef SimpleScanTask self = SimpleScanTask.__new__(SimpleScanTask)
-        self.init(up)
+        self.init(sp)
         return self
 
+    cdef inline CScanTaskPtr unwrap(self):
+        return self.wrapped
+
     def scan(self):
         cdef:
-            CRecordBatchIterator it = self.task.Scan()
+            CResult[CRecordBatchIterator] iterator_result
+            CRecordBatchIterator iterator
             shared_ptr[CRecordBatch] record_batch
 
+        iterator_result = self.task.Scan()
+        iterator = move(GetResultValue(iterator_result))
+
         while True:
-            it.Next(&record_batch)
-            if record_batch == nullptr:
+            iterator.Next(&record_batch)
+            if record_batch.get() == nullptr:
                 raise StopIteration()
             else:
                 yield pyarrow_wrap_batch(record_batch)
@@ -445,69 +490,77 @@ cdef class SimpleScanTask(ScanTask):
     cdef:
         CSimpleScanTask* simple_task
 
-    cdef init(self, unique_ptr[CScanTask]& up):
-        ScanTask.init(self, up)
-        self.simple_task = <CSimpleScanTask*> up.get()
+    cdef init(self, CScanTaskPtr& sp):
+        ScanTask.init(self, sp)
+        self.simple_task = <CSimpleScanTask*> sp.get()
+
+
+# cdef class ScannerBuilder:
+
+#     cdef:
+#         CScannerBuilderPtr wrapped
+#         CScannerBuilder* builder
 
 
 cdef class Scanner:
 
     cdef:
-        unique_ptr[CScanner] wrapped
+        CScannerPtr wrapped
         CScanner* scanner
 
-    cdef void init(self, unique_ptr[CScanner]& up):
-        self.wrapped = move(up)
+    # def __init__(self, data_sources, ScanOptions options, ScanContext context):
+    #     cdef:
+    #         DataSource source
+    #         CDataSourceVector sources
+    #         CSimpleScanner* simple_scanner
+    #         unique_ptr[CScanner] scanner
+
+    #     for source in data_sources:
+    #         sources.push_back(source.unwrap())
+
+    #     simple_scanner = new CSimpleScanner(
+    #         sources, options.unwrap(), context.unwrap()
+    #     )
+    #     scanner.reset(simple_scanner)
+    #     self.init(scanner)
+
+    cdef void init(self, CScannerPtr& sp):
+        self.wrapped = sp
         self.scanner = self.wrapped.get()
 
     @staticmethod
-    cdef wrap(unique_ptr[CScanner]& up):
-        cdef Scanner self = SimpleScanner.__new__(SimpleScanner)
-        self.init(up)
+    cdef wrap(CScannerPtr& sp):
+        cdef Scanner self = Scanner.__new__(Scanner)
+        self.init(sp)
         return self
 
     def scan(self):
         cdef:
-            CScanTaskIterator task_iterator = self.scanner.Scan()
-            unique_ptr[CScanTask] task
+            CResult[CScanTaskIterator] iterator_result
+            CScanTaskIterator iterator
+            CScanTaskPtr task
+
+        iterator_result = self.scanner.Scan()
+        iterator = move(GetResultValue(iterator_result))
 
         while True:
-            task_iterator.Next(&task)
-            if task == nullptr:
+            iterator.Next(&task)
+            if task.get() == nullptr:
                 raise StopIteration()
             else:
                 yield ScanTask.wrap(task)
 
     def to_table(self):
-        cdef shared_ptr[CTable] table
+        cdef:
+            shared_ptr[CTable] table
+            CResult[shared_ptr[CTable]] result
 
         with nogil:
-            check_status(self.scanner.ToTable(&table))
+            result = self.scanner.ToTable()
+        table = GetResultValue(result)
 
         return pyarrow_wrap_table(table)
 
 
-cdef class SimpleScanner(Scanner):
-
-    cdef:
-        CSimpleScanner* simple_scanner
-
-    def __init__(self, data_sources, ScanOptions options, ScanContext context):
-        cdef:
-            DataSource source
-            CDataSourceVector sources
-            CSimpleScanner* simple_scanner
-            unique_ptr[CScanner] scanner
-
-        for source in data_sources:
-            sources.push_back(source.unwrap())
-
-        simple_scanner = new CSimpleScanner(
-            sources, options.unwrap(), context.unwrap()
-        )
-        scanner.reset(simple_scanner)
-        self.init(scanner)
-
-
 # cdef class ScannerBuilder:
 #     pass

python/pyarrow/_fs.pxd

@@ -39,7 +39,7 @@ cdef class FileStats:
         CFileStats stats
 
     @staticmethod
-    cdef FileStats wrap(CFileStats stats)
+    cdef wrap(CFileStats stats)
 
 
 cdef class Selector:
@@ -54,6 +54,9 @@ cdef class FileSystem:
 
     cdef init(self, const shared_ptr[CFileSystem]& wrapped)
 
+    @staticmethod
+    cdef wrap(shared_ptr[CFileSystem]& sp)
+
 
 cdef class LocalFileSystem(FileSystem):
     cdef:

python/pyarrow/_fs.pyx

@@ -48,7 +48,7 @@ cdef class FileStats:
                         "FileSystem.get_target_stats method instead.")
 
     @staticmethod
-    cdef FileStats wrap(CFileStats stats):
+    cdef wrap(CFileStats stats):
         cdef FileStats self = FileStats.__new__(FileStats)
         self.stats = stats
         return self
@@ -184,6 +184,22 @@ cdef class FileSystem:
         self.wrapped = wrapped
         self.fs = wrapped.get()
 
+    @staticmethod
+    cdef wrap(shared_ptr[CFileSystem]& sp):
+        cdef FileSystem self
+
+        typ = tobytes(sp.get().type())
+        print(FileSystem.__subclasses__)
+        for klass in FileSystem.__subclasses__:
+            if klass.type_id == typ:
+                self = klass.__new__(klass)
+                break
+        else:
+            raise TypeError('Cannot wrap FileSystem pointer')
+
+        self.init(sp)
+        return self
+
     def get_target_stats(self, paths_or_selector):
         """Get statistics for the given target.
 

python/pyarrow/dataset.py

@@ -30,7 +30,6 @@ from pyarrow._dataset import (
     ScanOptions,
     ScanTask,
     Scanner,
-    SimpleScanner,
     ScanOptions,
     ScanContext,
     SimpleDataSource,

python/pyarrow/fs.py

@@ -24,7 +24,8 @@ from pyarrow._fs import (  # noqa
     FileSystem,
     LocalFileSystem,
     LocalFileSystemOptions,
-    SubTreeFileSystem
+    SubTreeFileSystem,
+    _MockFileSystem
 )
 
 try:

python/pyarrow/includes/libarrow_dataset.pxd

@@ -36,12 +36,6 @@ cdef extern from "arrow/dataset/api.h" namespace "arrow::dataset" nogil:
     #     CFilter_EXPRESSION" arrow::dataset::Filter::type::EXPRESSION"
     #     CFilter_GENERIC" arrow::dataset::Filter::GENERIC"
 
-    cdef cppclass CDataFragmentIterator "arrow::dataset::DataFragmentIterator"(
-            CIterator[shared_ptr[CDataFragment]]):
-        pass
-
-    cdef cppclass CDataFragmentVector "arrow::dataset::DataFragmentVector":
-        pass
 
     cdef cppclass CDataSourceVector "arrow::dataset::DataSourceVector"(
             vector[shared_ptr[CDataSource]]):
@@ -60,66 +54,92 @@ cdef extern from "arrow/dataset/api.h" namespace "arrow::dataset" nogil:
     cdef cppclass CFilter "arrow::dataset::Filter":
         pass
 
-    cdef cppclass CScanOptions" arrow::dataset::ScanOptions":
+    cdef cppclass CScanOptions "arrow::dataset::ScanOptions":
         # const shared_ptr[DataSelector]& selector()
         pass
 
-    cdef cppclass CScanContext" arrow::dataset::ScanContext":
+    cdef cppclass CScanOptionsPtr "arrow::dataset::ScanOptionsPtr"(
+            shared_ptr[CScanOptions]):
+        pass
+
+    cdef cppclass CScanContext "arrow::dataset::ScanContext":
         CMemoryPool* pool
 
-    cdef cppclass CScanTask" arrow::dataset::ScanTask":
-        CRecordBatchIterator Scan()
+    cdef cppclass CScanContextPtr "arrow::dataset::ScanContextPtr"(
+            shared_ptr[CScanContext]):
+        pass
 
-    cdef cppclass CSimpleScanTask "arrow::dataset::SimpleScanTask":
+    cdef cppclass CScanTask" arrow::dataset::ScanTask":
+        CResult[CRecordBatchIterator] Scan()
+
+    cdef cppclass CScanTaskPtr "arrow::dataset::ScanTaskPtr"(
+            shared_ptr[CScanTask]):
         pass
 
     cdef cppclass CScanTaskIterator "arrow::dataset::ScanTaskIterator"(
-            CIterator[unique_ptr[CScanTask]]):
+            CIterator[CScanTaskPtr]):
+        pass
+
+    cdef cppclass CSimpleScanTask "arrow::dataset::SimpleScanTask"(CScanTask):
         pass
 
     cdef cppclass CScannerBuilder "arrow::dataset::ScannerBuilder":
         CScannerBuilder(shared_ptr[CDataset] dataset,
                         shared_ptr[CScanContext] scan_context)
-        CStatus Project(const vector[c_string]& columns)
-        CStatus Filter(const CExpression& filter)
-        CStatus Filter(shared_ptr[CExpression] filter)
-        CStatus UseThreads(c_bool use_threads)
-        CStatus Finish(unique_ptr[CScanner]* out) const
+        # CStatus Project(const vector[c_string]& columns)
+        # CStatus Filter(const CExpression& filter)
+        # CStatus Filter(shared_ptr[CExpression] filter)
+        # CStatus UseThreads(c_bool use_threads)
+        CResult[CScannerPtr] Finish()
         shared_ptr[CSchema] schema() const
 
-    cdef cppclass CScanTaskIterator "arrow::dataset::ScanTaskIterator":
+    cdef cppclass CScannerBuilderPtr "arrow::dataset::ScannerBuilderPtr"(
+            shared_ptr[CScannerBuilder]):
         pass
 
     cdef cppclass CScanner "arrow::dataset::Scanner":
-        c_bool use_threads
-        shared_ptr[CExpression] filter
-        # shared_ptr[CExpressionEvaluator] evaluator
-        shared_ptr[CSchema] schema
-        # shared_ptr[CRecordBatchProjector] projector
-        CScanTaskIterator Scan()
-        CStatus ToTable(shared_ptr[CTable]* out)
+        CResult[CScanTaskIterator] Scan()
+        CResult[shared_ptr[CTable]] ToTable()
 
-    cdef cppclass CSimpleScanner "arrow::dataset::SimpleScanner"(CScanner):
-        CSimpleScanner(CDataSourceVector sources,
-                       shared_ptr[CScanOptions] options,
-                       shared_ptr[CScanContext] context)
+    cdef cppclass CScannerPtr "arrow::dataset::ScannerPtr"(
+            shared_ptr[CScanner]):
+        pass
 
     cdef cppclass CDataFragment "arrow::dataset::DataFragment":
-        CStatus Scan(shared_ptr[CScanContext] scan_context,
-                     CScanTaskIterator* out)
+        CResult[CScanTaskIterator] Scan(CScanContextPtr context)
         c_bool splittable()
-        shared_ptr[CScanOptions] scan_options()
+        CScanOptionsPtr scan_options()
+
+    cdef cppclass CDataFragmentPtr "arrow::dataset::DataFragmentPtr"(
+            shared_ptr[CDataFragment]):
+        pass
+
+    cdef cppclass CDataFragmentIterator "arrow::dataset::DataFragmentIterator"(
+            CIterator[CDataFragmentPtr]):
+        pass
+
+    cdef cppclass CDataFragmentVector "arrow::dataset::DataFragmentVector"(
+            vector[CDataFragmentPtr]):
+        pass
 
     cdef cppclass CSimpleDataFragment "arrow::dataset::SimpleDataFragment"(
             CDataFragment):
         CSimpleDataFragment(vector[shared_ptr[CRecordBatch]] record_batches,
                             shared_ptr[CScanOptions] scan_options)
 
-    cdef cppclass CDataSource" arrow::dataset::DataSource":
+    cdef cppclass CDataSource "arrow::dataset::DataSource":
         CDataFragmentIterator GetFragments(shared_ptr[CScanOptions] options)
         # const shared_ptr[CExpression]& partition_expression()
         c_string type()
 
+    cdef cppclass CDataSourcePtr "arrow::dataset::DataSourcePtr"(
+            shared_ptr[CDataSource]):
+        pass
+
+    cdef cppclass CDataSourceVector "arrow::dataset::DataSourceVector"(
+            vector[shared_ptr[CDataSource]]):
+        pass
+
     cdef cppclass CSimpleDataSource "arrow::dataset::SimpleDataSource"(
             CDataSource):
         pass
@@ -128,14 +148,18 @@ cdef extern from "arrow/dataset/api.h" namespace "arrow::dataset" nogil:
             CDataSource):
         pass
 
-    cdef cppclass CDataset" arrow::dataset::Dataset":
+    cdef cppclass CDataset "arrow::dataset::Dataset":
         @staticmethod
-        CStatus Make(vector[shared_ptr[CDataSource]] sources,
-                     shared_ptr[CSchema] schema, shared_ptr[CDataset] *out)
-        CStatus NewScan(unique_ptr[CScannerBuilder]* out)
-        const vector[shared_ptr[CDataSource]]& sources()
+        CResult[CDatasetPtr] Make(CDataSourceVector sources,
+                                  shared_ptr[CSchema] schema)
+        CResult[CScannerBuilderPtr] NewScan()
+        const CDataSourceVector& sources()
         shared_ptr[CSchema] schema()
 
+    cdef cppclass CDatasetPtr "arrow::dataset::DatasetPtr"(
+            shared_ptr[CDataset]):
+        pass
+
     ############################### File ######################################
 
     cdef cppclass CFileScanOptions "arrow::dataset::FileScanOptions"(
@@ -167,7 +191,7 @@ cdef extern from "arrow/dataset/api.h" namespace "arrow::dataset" nogil:
                          CScanTaskIterator* out) const
         CStatus MakeFragment(const CFileSource& location,
                              shared_ptr[CScanOptions] opts,
-                             unique_ptr[CDataFragment]* out)
+                             shared_ptr[CDataFragment]* out)
 
     cdef cppclass CFileBasedDataFragment \
             "arrow::dataset::FileBasedDataFragment"(CDataFragment):
@@ -175,22 +199,21 @@ cdef extern from "arrow/dataset/api.h" namespace "arrow::dataset" nogil:
                                shared_ptr[CFileFormat] format,
                                shared_ptr[CScanOptions] scan_options)
         CStatus Scan(shared_ptr[CScanContext] scan_context,
-                     unique_ptr[CScanTaskIterator]* out)
+                     shared_ptr[CScanTaskIterator]* out)
         const CFileSource& source()
         shared_ptr[CFileFormat] format()
         shared_ptr[CScanOptions] scan_options()
 
-    # NOTE(kszucs): removed Based from the class name
     cdef cppclass CFileSystemDataSource \
-            "arrow::dataset::FileSystemBasedDataSource"(CDataSource):
+            "arrow::dataset::FileSystemDataSource"(CDataSource):
         @staticmethod
         CStatus Make(CFileSystem* filesystem,
                      const CSelector& selector,
                      shared_ptr[CFileFormat] format,
                      shared_ptr[CScanOptions] scan_options,
-                     unique_ptr[CFileSystemDataSource]* out)
+                     shared_ptr[CFileSystemDataSource]* out)
         c_string type()
-        unique_ptr[CDataFragmentIterator] GetFragments(
+        shared_ptr[CDataFragmentIterator] GetFragments(
             shared_ptr[CScanOptions] options)
 
     ############################### File CSV ##################################
@@ -207,9 +230,9 @@ cdef extern from "arrow/dataset/api.h" namespace "arrow::dataset" nogil:
         c_string name()
         c_bool IsKnownExtension(const c_string& ext)
         CStatus ScanFile(const CFileSource& source,
-                         shared_ptr[CScanOptions] scan_options,
-                         shared_ptr[CScanContext] scan_context,
-                         unique_ptr[CScanTaskIterator]* out)
+                         CScanOptionsPtr scan_options,
+                         CScanContextPtr scan_context,
+                         shared_ptr[CScanTaskIterator]* out)
 
      ############################### File JSON ################################
 
@@ -228,7 +251,7 @@ cdef extern from "arrow/dataset/api.h" namespace "arrow::dataset" nogil:
         CStatus ScanFile(const CFileSource& source,
                          shared_ptr[CScanOptions] scan_options,
                          shared_ptr[CScanContext] scan_context,
-                         unique_ptr[CScanTaskIterator]* out)
+                         shared_ptr[CScanTaskIterator]* out)
 
     ############################### File Feather #############################
 
@@ -247,7 +270,7 @@ cdef extern from "arrow/dataset/api.h" namespace "arrow::dataset" nogil:
         CStatus ScanFile(const CFileSource& source,
                          shared_ptr[CScanOptions] scan_options,
                          shared_ptr[CScanContext] scan_context,
-                         unique_ptr[CScanTaskIterator]* out)
+                         shared_ptr[CScanTaskIterator]* out)
 
     ############################### File Parquet #############################
 
@@ -266,10 +289,10 @@ cdef extern from "arrow/dataset/api.h" namespace "arrow::dataset" nogil:
         CStatus ScanFile(const CFileSource& source,
                          shared_ptr[CScanOptions] scan_options,
                          shared_ptr[CScanContext] scan_context,
-                         unique_ptr[CScanTaskIterator]* out)
+                         shared_ptr[CScanTaskIterator]* out)
         CStatus MakeFragment(const CFileSource& source,
                              shared_ptr[CScanOptions] opts,
-                             unique_ptr[CDataFragment]* out)
+                             shared_ptr[CDataFragment]* out)
 
     cdef cppclass CParquetFragment "arrow::dataset::ParquetFragment"(
             CFileBasedDataFragment):

python/pyarrow/tests/test_dataset.py

@@ -20,6 +20,7 @@ from collections.abc import Generator
 import pytest
 
 import pyarrow as pa
+from pyarrow.fs import _MockFileSystem
 
 try:
     from pyarrow.dataset import (
@@ -32,7 +33,6 @@ try:
         Scanner,
         ScanOptions,
         ScanContext,
-        SimpleScanner,
         SimpleDataSource,
         TreeDataSource
     )
@@ -45,6 +45,11 @@ except ImportError as e:
 pytestmark = pytest.mark.dataset
 
 
+@pytest.fixture
+def fs():
+    return _MockFileSystem()
+
+
 @pytest.fixture
 def schema():
     return pa.schema([
@@ -139,7 +144,6 @@ def test_dataset(simple_data_source, tree_data_source):
 
     scanner = dataset.new_scan()
     assert isinstance(scanner, Scanner)
-    assert isinstance(scanner, SimpleScanner)
 
     for task in scanner.scan():
         assert isinstance(task, ScanTask)
@@ -159,3 +163,19 @@ def test_dataset(simple_data_source, tree_data_source):
 
 def test_projector():
     pass
+
+
+# def test_file_source(fs):
+#     source1 = FileSource('/path/to/file.ext', fs, compression=None)
+#     source2 = FileSource('/path/to/file.ext.gz', fs, compression='gzip')
+
+#     assert source1.path == '/path/to/file.ext'
+#     assert source1.fs == fs
+#     assert source1.compression is None
+#     assert source2.path == '/path/to/file.ext.gz'
+#     assert source2.fs == fs
+#     assert source2.compression == 'gzip'
+
+
+def test_file_system_data_source():
+    pass