Address review: private SourceEntry, const fields, deduplicate batch logic, hoist columns

- Make SourceEntry private in RecordBatchRowProcessor
- Make m_arrowTypeId and m_columnName const
- Replace duplicated first-batch loop with fetchNextBatch call
- Hoist columns() out of rebindSource loop

Signed-off-by: Arham Chopra <arham.chopra@cubistsystematic.com>

Author: arhamchopra

cpp/csp/adapters/arrow/ColumnDispatcher.h

@@ -57,8 +57,8 @@ class ColumnDispatcher
 
     virtual void * getCurValueUntyped() = 0;
 
-    ::arrow::Type::type m_arrowTypeId;
-    std::string         m_columnName;
+    const ::arrow::Type::type m_arrowTypeId;
+    const std::string         m_columnName;
 };
 
 // Factory: create a typed ColumnDispatcher for an arrow field.

cpp/csp/adapters/arrow/RecordBatchRowProcessor.cpp

@@ -102,22 +102,7 @@ void RecordBatchRowProcessor::bindSources(
         }
 
         // Pull first non-empty batch
-        for( ;; )
-        {
-            auto status = entry.source -> ReadNext( &entry.currentBatch );
-            CSP_TRUE_OR_THROW_RUNTIME( status.ok(),
-                "bindSources: failed to read first batch: " << status.ToString() );
-
-            if( !entry.currentBatch )
-                break;
-
-            entry.numRows    = entry.currentBatch -> num_rows();
-            entry.currentRow = 0;
-            rebindSource( entry );
-
-            if( entry.numRows > 0 )
-                break;
-        }
+        fetchNextBatch( entry );
 
         m_sources.push_back( std::move( entry ) );
     }
@@ -143,8 +128,9 @@ bool RecordBatchRowProcessor::fetchNextBatch( SourceEntry & entry )
 
 void RecordBatchRowProcessor::rebindSource( SourceEntry & entry )
 {
+    auto & cols = entry.currentBatch -> columns();
     for( size_t i = 0; i < entry.dispatchers.size(); ++i )
-        entry.dispatchers[i] -> bindColumn( entry.currentBatch -> column( entry.colIndices[i] ).get() );
+        entry.dispatchers[i] -> bindColumn( cols[entry.colIndices[i]].get() );
 }
 
 bool RecordBatchRowProcessor::skipRow()

cpp/csp/adapters/arrow/RecordBatchRowProcessor.h

@@ -31,16 +31,6 @@ class RecordBatchRowProcessor
         int         colIndex;
     };
 
-    struct SourceEntry
-    {
-        ::arrow::RecordBatchReader *                  source = nullptr;
-        std::shared_ptr<::arrow::RecordBatch>        currentBatch;
-        int64_t                                      numRows    = 0;
-        int64_t                                      currentRow = 0;
-        std::vector<int>                             colIndices;
-        std::vector<ColumnDispatcher *>              dispatchers;
-    };
-
     RecordBatchRowProcessor() = default;
 
     // (Re)create dispatchers for the given schema and requested columns.
@@ -74,6 +64,16 @@ class RecordBatchRowProcessor
     bool readRowAndAdvance();
 
 private:
+    struct SourceEntry
+    {
+        ::arrow::RecordBatchReader *                  source = nullptr;
+        std::shared_ptr<::arrow::RecordBatch>        currentBatch;
+        int64_t                                      numRows    = 0;
+        int64_t                                      currentRow = 0;
+        std::vector<int>                             colIndices;
+        std::vector<ColumnDispatcher *>              dispatchers;
+    };
+
     std::vector<std::unique_ptr<ColumnDispatcher>>      m_dispatchers;
     std::unordered_map<std::string, ColumnDispatcher *>  m_nameToDispatcher;
     std::vector<SourceEntry>                             m_sources;