Address review feedback on CppNodes and arrow.py

- Remove unnecessary ticked() checks on single-input nodes
- Remove extra blank line in START()
- Use PyList_GET_ITEM for faster unchecked access in loop
- Use typed TS_INPUT(vector<StructPtr>) instead of Generic for struct_to_record_batches
- Use standard @csp.node decorator instead of _node_internal_use

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

Author: arhamchopra

cpp/csp/python/adapters/ArrowCppNodes.cpp

@@ -72,15 +72,11 @@ DECLARE_CPPNODE( record_batches_to_struct )
             fieldMap = props -> get<DictionaryPtr>( "field_map" );
 
         auto structMeta = cls.value();
-
         s_converter = std::make_unique<RecordBatchToStructConverter>( s_schema, structMeta, fieldMap );
     }
 
     INVOKE()
     {
-        if( !data.ticked() )
-            return;
-
         // data is a list of (schema_capsule, array_capsule) tuples
         PyObject * pyList = csp::python::toPythonBorrowed( data.lastValue<DialectGenericType>() );
         if( !PyList_Check( pyList ) )
@@ -95,7 +91,7 @@ DECLARE_CPPNODE( record_batches_to_struct )
 
         for( Py_ssize_t i = 0; i < numBatches; ++i )
         {
-            PyObject * pyTuple = PyList_GetItem( pyList, i );
+            PyObject * pyTuple = PyList_GET_ITEM( pyList, i );
             if( !PyTuple_Check( pyTuple ) || PyTuple_Size( pyTuple ) != 2 )
                 CSP_THROW( csp::TypeError, "Expected tuple of 2 PyCapsules for record batch " << i );
 
@@ -137,9 +133,9 @@ EXPORT_CPPNODE( record_batches_to_struct );
 
 DECLARE_CPPNODE( struct_to_record_batches )
 {
-    SCALAR_INPUT( StructMetaPtr,  cls );         // source struct type
-    SCALAR_INPUT( DictionaryPtr,  properties );  // field_map, numpy config
-    TS_INPUT( Generic,            data );         // vector<StructPtr>
+    SCALAR_INPUT( StructMetaPtr,      cls );         // source struct type
+    SCALAR_INPUT( DictionaryPtr,      properties );  // field_map, numpy config
+    TS_INPUT( std::vector<StructPtr>, data );
     TS_OUTPUT( Generic );                        // DialectGenericType (Python list of capsule tuples)
 
     STATE_VAR( std::unique_ptr<StructToRecordBatchConverter>, s_converter );
@@ -163,10 +159,7 @@ DECLARE_CPPNODE( struct_to_record_batches )
 
     INVOKE()
     {
-        if( !data.ticked() )
-            return;
-
-        auto & structs = data.lastValue<std::vector<StructPtr>>();
+        auto & structs = data.lastValue();
         auto batches = s_converter -> convert( structs, s_maxBatchSize );
 
         auto py_list = csp::python::PyObjectPtr::own( PyList_New( static_cast<Py_ssize_t>( batches.size() ) ) );

csp/adapters/arrow.py

@@ -7,7 +7,6 @@
 import csp
 from csp.impl.types.tstype import ts
 from csp.impl.wiring import input_adapter_def
-from csp.impl.wiring.node import _node_internal_use
 from csp.lib import _arrowadapterimpl
 
 __all__ = [
@@ -153,7 +152,7 @@ def write_record_batches(
                 s_prev_batch_size += len(batch)
 
 
-@_node_internal_use(cppimpl=_arrowadapterimpl.record_batches_to_struct)
+@csp.node(cppimpl=_arrowadapterimpl.record_batches_to_struct)
 def _record_batches_to_struct(
     schema_ptr: object,
     cls: "T",
@@ -202,7 +201,7 @@ def record_batches_to_struct(
     return _record_batches_to_struct(schema_capsule, cls, properties, c_data)
 
 
-@_node_internal_use(cppimpl=_arrowadapterimpl.struct_to_record_batches)
+@csp.node(cppimpl=_arrowadapterimpl.struct_to_record_batches)
 def _struct_to_record_batches(
     cls: "T",
     properties: dict,