ARROW-8201: [Python] Add FileFragment.open() method

python/pyarrow/_dataset.pyx

@@ -1053,6 +1053,28 @@ cdef class FileFragment(Fragment):
             self.partition_expression
         )
 
+    def open(self):
+        """
+        Open a NativeFile of the buffer or file viewed by this fragment.
+        """
+        cdef:
+            shared_ptr[CFileSystem] c_filesystem
+            shared_ptr[CRandomAccessFile] opened
+            c_string c_path
+            NativeFile out = NativeFile()
+
+        if self.buffer is not None:
+            return pa.BufferReader(self.buffer)
+
+        c_path = tobytes(self.file_fragment.source().path())
+        with nogil:
+            c_filesystem = self.file_fragment.source().filesystem()
+            opened = GetResultValue(c_filesystem.get().OpenInputFile(c_path))
+
+        out.set_random_access_file(opened)
+        out.is_readable = True
+        return out
+
     @property
     def path(self):
         """

python/pyarrow/tests/test_dataset.py

@@ -86,6 +86,19 @@ def _table_from_pandas(df):
     return table.replace_schema_metadata()
 
 
+def assert_dataset_fragment_convenience_methods(dataset):
+    # FileFragment convenience methods
+    for fragment in dataset.get_fragments():
+        if pq is not None and isinstance(dataset.format, ds.ParquetFileFormat):
+            assert isinstance(fragment.metadata, pq.FileMetaData)
+        with fragment.open() as nf:
+            assert isinstance(nf, pa.NativeFile)
+            assert not nf.closed
+            assert nf.seekable()
+            assert nf.readable()
+            assert not nf.writable()
+
+
 @pytest.fixture
 @pytest.mark.parquet
 def mockfs():
@@ -421,6 +434,7 @@ def test_dataset(dataset, dataset_reader):
                              2.0, 2.0, 3.0, 3.0, 4.0, 4.0]
     assert result['new'] == [False, False, True, True, False, False,
                              False, False, True, True]
+    assert_dataset_fragment_convenience_methods(dataset)
 
 
 @pytest.mark.parquet
@@ -951,6 +965,9 @@ def test_make_csv_fragment_from_buffer(dataset_reader):
     csv_format = ds.CsvFileFormat()
     fragment = csv_format.make_fragment(buffer)
 
+    # When buffer, fragment open returns a BufferReader, not NativeFile
+    assert isinstance(fragment.open(), pa.BufferReader)
+
     expected = pa.table([['a', 'b', 'c'],
                          [12, 11, 10],
                          ['dog', 'cat', 'rabbit']],
@@ -2955,6 +2972,8 @@ def test_ipc_format(tempdir, dataset_reader):
     result = dataset_reader.to_table(dataset)
     assert result.equals(table)
 
+    assert_dataset_fragment_convenience_methods(dataset)
+
     for format_str in ["ipc", "arrow"]:
         dataset = ds.dataset(path, format=format_str)
         result = dataset_reader.to_table(dataset)
@@ -2977,6 +2996,8 @@ def test_orc_format(tempdir, dataset_reader):
     result.validate(full=True)
     assert result.equals(table)
 
+    assert_dataset_fragment_convenience_methods(dataset)
+
     dataset = ds.dataset(path, format="orc")
     result = dataset_reader.to_table(dataset)
     result.validate(full=True)
@@ -3044,6 +3065,8 @@ def test_csv_format(tempdir, dataset_reader):
     result = dataset_reader.to_table(dataset)
     assert result.equals(table)
 
+    assert_dataset_fragment_convenience_methods(dataset)
+
     dataset = ds.dataset(path, format='csv')
     result = dataset_reader.to_table(dataset)
     assert result.equals(table)
@@ -3200,6 +3223,8 @@ def test_feather_format(tempdir, dataset_reader):
     result = dataset_reader.to_table(dataset)
     assert result.equals(table)
 
+    assert_dataset_fragment_convenience_methods(dataset)
+
     dataset = ds.dataset(basedir, format="feather")
     result = dataset_reader.to_table(dataset)
     assert result.equals(table)