Test read multiple Parquet from HDFS, fix premature garbage collection error

Change-Id: I4a21d0f5d3b160a5edc950b42f862d34c7fee3ae

python/pyarrow/_parquet.pyx

@@ -341,6 +341,7 @@ cdef logical_type_name_from_enum(ParquetLogicalType type_):
 
 cdef class ParquetReader:
     cdef:
+        object source
         MemoryPool* allocator
         unique_ptr[FileReader] reader
         column_idx_map
@@ -360,6 +361,8 @@ cdef class ParquetReader:
         if metadata is not None:
             c_metadata = metadata.sp_metadata
 
+        self.source = source
+
         get_reader(source, &rd_handle)
         with nogil:
             check_status(OpenFile(rd_handle, self.allocator, properties,

python/pyarrow/parquet.py

@@ -72,7 +72,7 @@ class ParquetFile(object):
             return Table.from_arrays(arrays, names=columns)
 
 
-def read_table(source, columns=None):
+def read_table(source, columns=None, metadata=None):
     """
     Read a Table from Parquet format
 
@@ -83,13 +83,15 @@ def read_table(source, columns=None):
         pyarrow.io.PythonFileInterface or pyarrow.io.BufferReader.
     columns: list
         If not None, only these columns will be read from the file.
+    metadata : FileMetaData
+        If separately computed
 
     Returns
     -------
     pyarrow.Table
         Content of the file as a table (of columns)
     """
-    return ParquetFile(source).read(columns=columns)
+    return ParquetFile(source, metadata=metadata).read(columns=columns)
 
 
 def read_multiple_files(paths, columns=None, filesystem=None, metadata=None,
@@ -118,12 +120,11 @@ def read_multiple_files(paths, columns=None, filesystem=None, metadata=None,
         Content of the file as a table (of columns)
     """
     if filesystem is None:
-        def open_file(path, metadata=None):
-            return ParquetFile(path, metadata=metadata)
+        def open_file(path, meta=None):
+            return ParquetFile(path, metadata=meta)
     else:
-        def open_file(path, metadata=None):
-            return ParquetFile(filesystem.open(path, mode='rb'),
-                               metadata=metadata)
+        def open_file(path, meta=None):
+            return ParquetFile(filesystem.open(path, mode='rb'), metadata=meta)
 
     if len(paths) == 0:
         raise ValueError('Must pass at least one file path')
@@ -145,7 +146,7 @@ def read_multiple_files(paths, columns=None, filesystem=None, metadata=None,
     # Read the tables
     tables = []
     for path, path_metadata in zip(paths, all_file_metadata):
-        reader = open_file(path, metadata=path_metadata)
+        reader = open_file(path, meta=path_metadata)
         table = reader.read(columns=columns)
         tables.append(table)
 

python/pyarrow/tests/test_hdfs.py

@@ -21,9 +21,14 @@ import os
 import random
 import unittest
 
+import numpy as np
 import pytest
 
+from pyarrow.compat import guid
 import pyarrow.io as io
+import pyarrow as pa
+
+import pyarrow.tests.test_parquet as test_parquet
 
 # ----------------------------------------------------------------------
 # HDFS tests
@@ -138,6 +143,39 @@ class HdfsTestCases(object):
 
         assert result == data
 
+    @test_parquet.parquet
+    def test_hdfs_read_multiple_parquet_files(self):
+        import pyarrow.parquet as pq
+
+        nfiles = 10
+        size = 5
+
+        tmpdir = pjoin(self.tmp_path, 'multi-parquet')
+
+        self.hdfs.mkdir(tmpdir)
+
+        test_data = []
+        paths = []
+        for i in range(nfiles):
+            df = test_parquet._test_dataframe(size, seed=i)
+
+            # Hack so that we don't have a dtype cast in v1 files
+            df['uint32'] = df['uint32'].astype(np.int64)
+
+            path = pjoin(tmpdir, '{0}.parquet'.format(guid()))
+
+            table = pa.Table.from_pandas(df)
+            with self.hdfs.open(path, 'wb') as f:
+                pq.write_table(table, f)
+
+            test_data.append(table)
+            paths.append(path)
+
+        result = pq.read_multiple_files(paths, filesystem=self.hdfs)
+        expected = pa.concat_tables(test_data)
+
+        assert result.equals(expected)
+
 
 class TestLibHdfs(HdfsTestCases, unittest.TestCase):