ARROW-369: [Python] Convert multiple record batches at once to Pandas

Modified Pandas adapter to handle columns with multiple chunks with `ConvertColumnToPandas`.  This modifies the pyarrow public API by adding a class `RecordBatchList` and static method `toPandas` which takes a list of Arrow RecordBatches and outputs a Pandas DataFrame.

Adds unit test in test_table.py to do the conversion for each column with typed specialization.

Author: Bryan Cutler <cutlerb@gmail.com>

Closes #216 from BryanCutler/multi-batch-toPandas-ARROW-369 and squashes the following commits:

b6c9986 [Bryan Cutler] fixed formatting
edf056e [Bryan Cutler] simplified with pyarrow.schema.Schema.equals
068bc1b [Bryan Cutler] Merge remote-tracking branch 'upstream/master' into multi-batch-toPandas-ARROW-369
da65345 [Bryan Cutler] fixed test case for schema checking
9edb0ba [Bryan Cutler] used auto keyword where some typecasting was done in ConvertValues
bd2a720 [Bryan Cutler] added testcase for schema not equal, disabled now
c3d7e8f [Bryan Cutler] Changed conversion to make Table from columns first, now conversion is now just a free function
3ee51e6 [Bryan Cutler] cleanup
398b18d [Bryan Cutler] Fixed case for Integer specialization without nulls
7b29a55 [Bryan Cutler] Initial working version of RecordBatch list to_pandas, need more tests and cleanup

Author: BryanCutler

python/pyarrow/__init__.py

@@ -41,5 +41,7 @@
                             list_, struct, field,
                             DataType, Field, Schema, schema)
 
-from pyarrow.table import Column, RecordBatch, Table, from_pandas_dataframe
+from pyarrow.table import (Column, RecordBatch, dataframe_from_batches, Table,
+                           from_pandas_dataframe)
+
 from pyarrow.version import version as __version__

python/pyarrow/includes/libarrow.pxd

@@ -158,6 +158,9 @@ cdef extern from "arrow/api.h" namespace "arrow" nogil:
         CColumn(const shared_ptr[CField]& field,
                 const shared_ptr[CArray]& data)
 
+        CColumn(const shared_ptr[CField]& field,
+                const vector[shared_ptr[CArray]]& chunks)
+
         int64_t length()
         int64_t null_count()
         const c_string& name()

python/pyarrow/table.pyx

@@ -28,6 +28,7 @@ cimport pyarrow.includes.pyarrow as pyarrow
 import pyarrow.config
 
 from pyarrow.array cimport Array, box_arrow_array
+from pyarrow.error import ArrowException
 from pyarrow.error cimport check_status
 from pyarrow.schema cimport box_data_type, box_schema
 
@@ -414,6 +415,52 @@ cdef class RecordBatch:
         return result
 
 
+def dataframe_from_batches(batches):
+    """
+    Convert a list of Arrow RecordBatches to a pandas.DataFrame
+
+    Parameters
+    ----------
+
+    batches: list of RecordBatch
+        RecordBatch list to be converted, schemas must be equal
+    """
+
+    cdef:
+        vector[shared_ptr[CArray]] c_array_chunks
+        vector[shared_ptr[CColumn]] c_columns
+        shared_ptr[CTable] c_table
+        Array arr
+        Schema schema
+
+    import pandas as pd
+
+    schema = batches[0].schema
+
+    # check schemas are equal
+    if any((not schema.equals(other.schema) for other in batches[1:])):
+        raise ArrowException("Error converting list of RecordBatches to "
+                "DataFrame, not all schemas are equal")
+
+    cdef int K = batches[0].num_columns
+
+    # create chunked columns from the batches
+    c_columns.resize(K)
+    for i in range(K):
+        for batch in batches:
+            arr = batch[i]
+            c_array_chunks.push_back(arr.sp_array)
+        c_columns[i].reset(new CColumn(schema.sp_schema.get().field(i),
+                           c_array_chunks))
+        c_array_chunks.clear()
+
+    # create a Table from columns and convert to DataFrame
+    c_table.reset(new CTable('', schema.sp_schema, c_columns))
+    table = Table()
+    table.init(c_table)
+    return table.to_pandas()
+
+
 cdef class Table:
     """
     A collection of top-level named, equal length Arrow arrays.

python/pyarrow/tests/test_table.py

@@ -19,6 +19,7 @@
 
 from pandas.util.testing import assert_frame_equal
 import pandas as pd
+import pytest
 
 import pyarrow as pa
 
@@ -50,6 +51,40 @@ def test_recordbatch_from_to_pandas():
     assert_frame_equal(data, result)
 
 
+def test_recordbatchlist_to_pandas():
+    data1 = pd.DataFrame({
+        'c1': np.array([1, 1, 2], dtype='uint32'),
+        'c2': np.array([1.0, 2.0, 3.0], dtype='float64'),
+        'c3': [True, None, False],
+        'c4': ['foo', 'bar', None]
+    })
+
+    data2 = pd.DataFrame({
+        'c1': np.array([3, 5], dtype='uint32'),
+        'c2': np.array([4.0, 5.0], dtype='float64'),
+        'c3': [True, True],
+        'c4': ['baz', 'qux']
+    })
+
+    batch1 = pa.RecordBatch.from_pandas(data1)
+    batch2 = pa.RecordBatch.from_pandas(data2)
+
+    result = pa.dataframe_from_batches([batch1, batch2])
+    data = pd.concat([data1, data2], ignore_index=True)
+    assert_frame_equal(data, result)
+
+
+def test_recordbatchlist_schema_equals():
+    data1 = pd.DataFrame({'c1': np.array([1], dtype='uint32')})
+    data2 = pd.DataFrame({'c1': np.array([4.0, 5.0], dtype='float64')})
+
+    batch1 = pa.RecordBatch.from_pandas(data1)
+    batch2 = pa.RecordBatch.from_pandas(data2)
+
+    with pytest.raises(pa.ArrowException):
+        pa.dataframe_from_batches([batch1, batch2])
+
+
 def test_table_basics():
     data = [
         pa.from_pylist(range(5)),