Add support for lists to Arrow <-> csp.Struct conversion nodes

cpp/csp/adapters/arrow/ArrowFieldReader.cpp

@@ -40,20 +40,6 @@ void readColumn( const ArrowArrayT & typed, std::vector<StructPtr> & structs, in
                 fn( typed, i, structs[i].get() );
 }
 
-// Helper: compute nanosecond multiplier for a given arrow::TimeUnit
-int64_t timeUnitMultiplier( ::arrow::TimeUnit::type unit )
-{
-    switch( unit )
-    {
-        case ::arrow::TimeUnit::SECOND: return csp::NANOS_PER_SECOND;
-        case ::arrow::TimeUnit::MILLI:  return csp::NANOS_PER_MILLISECOND;
-        case ::arrow::TimeUnit::MICRO:  return csp::NANOS_PER_MICROSECOND;
-        case ::arrow::TimeUnit::NANO:   return 1LL;
-        default:
-            CSP_THROW( TypeError, "Unexpected arrow TimeUnit: " << static_cast<int>( unit ) );
-    }
-}
-
 // --- Generic lambda-based reader (covers Primitive, HalfFloat, StringLike, Nanos, Date) ---
 // ExtractFn signature: ValueT(const ArrowArrayT &, int64_t row)
 

cpp/csp/adapters/arrow/ArrowFieldReader.h

@@ -11,6 +11,8 @@
 #define _IN_CSP_ADAPTERS_ARROW_ArrowFieldReader_H
 
 #include <csp/engine/Struct.h>
+#include <csp/core/Exception.h>
+#include <csp/core/Time.h>
 #include <arrow/record_batch.h>
 #include <arrow/type.h>
 #include <functional>
@@ -22,6 +24,20 @@
 namespace csp::adapters::arrow
 {
 
+// Helper: compute nanosecond multiplier for a given arrow::TimeUnit
+inline int64_t timeUnitMultiplier( ::arrow::TimeUnit::type unit )
+{
+    switch( unit )
+    {
+        case ::arrow::TimeUnit::SECOND: return csp::NANOS_PER_SECOND;
+        case ::arrow::TimeUnit::MILLI:  return csp::NANOS_PER_MILLISECOND;
+        case ::arrow::TimeUnit::MICRO:  return csp::NANOS_PER_MICROSECOND;
+        case ::arrow::TimeUnit::NANO:   return 1LL;
+        default:
+            CSP_THROW( TypeError, "Unexpected arrow TimeUnit: " << static_cast<int>( unit ) );
+    }
+}
+
 class FieldReader
 {
 public:

cpp/csp/adapters/arrow/RecordBatchToStruct.cpp

@@ -10,25 +10,6 @@
 namespace csp::adapters::arrow
 {
 
-namespace
-{
-
-// Helper to resolve column name -> struct field name mapping
-// If fieldMap is null, column name = field name (identity mapping)
-std::string resolveFieldName( const DictionaryPtr & fieldMap, const std::string & columnName )
-{
-    if( !fieldMap )
-        return columnName;
-
-    std::string fieldName;
-    if( fieldMap -> tryGet<std::string>( columnName, fieldName ) )
-        return fieldName;
-
-    return columnName;
-}
-
-} // anonymous namespace
-
 RecordBatchToStructConverter::RecordBatchToStructConverter(
     const std::shared_ptr<::arrow::Schema> & schema,
     const std::shared_ptr<StructMeta> & structMeta,
@@ -51,12 +32,16 @@ RecordBatchToStructConverter::RecordBatchToStructConverter(
         if( customColumnNames.count( arrowField -> name() ) )
             continue;
 
-        // Skip columns that don't have a matching struct field
-        std::string fieldName = resolveFieldName( fieldMap, arrowField -> name() );
-        auto structField = structMeta -> field( fieldName );
-        if( !structField )
+        // Look up field name from the mapping; skip columns not in the map
+        std::string fieldName;
+        if( !fieldMap -> tryGet<std::string>( arrowField -> name(), fieldName ) )
             continue;
 
+        auto structField = structMeta -> field( fieldName );
+        CSP_TRUE_OR_THROW_RUNTIME( structField != nullptr,
+            "Struct field '" << fieldName << "' (mapped from column '" << arrowField -> name()
+            << "') not found on struct type '" << structMeta -> name() << "'" );
+
         m_scalarReaders.push_back( { createFieldReader( arrowField, structField ), i } );
     }
 

cpp/csp/adapters/arrow/StructToRecordBatch.cpp

@@ -17,42 +17,22 @@ StructToRecordBatchConverter::StructToRecordBatchConverter(
 {
     std::vector<std::shared_ptr<::arrow::Field>> arrowFields;
 
-    if( fieldMap )
+    for( auto it = fieldMap -> begin(); it != fieldMap -> end(); ++it )
     {
-        // When fieldMap is provided, only include fields listed in it
-        for( auto it = fieldMap -> begin(); it != fieldMap -> end(); ++it )
-        {
-            auto fieldName = it.key();
-            auto colName   = it.value<std::string>();
-            auto structField = structMeta -> field( fieldName );
-            if( !structField )
-                continue;
-
-            // Skip DIALECT_GENERIC fields (handled by custom writers)
-            if( structField -> type() -> type() == CspType::Type::DIALECT_GENERIC )
-                continue;
-
-            auto created = createFieldWriter( colName, structField );
-            for( auto & dt : created.writer -> dataTypes() )
-                arrowFields.push_back( std::make_shared<::arrow::Field>( colName, dt ) );
-            m_writers.push_back( std::move( created.writer ) );
-        }
-    }
-    else
-    {
-        // No fieldMap: include all non-DIALECT_GENERIC fields using fieldNames() for stable insertion order
-        // (fields() is sorted by type/size for memory layout optimization, not insertion order)
-        for( auto & fieldName : structMeta -> fieldNames() )
-        {
-            auto structField = structMeta -> field( fieldName );
-            if( !structField || structField -> type() -> type() == CspType::Type::DIALECT_GENERIC )
-                continue;
-
-            auto created = createFieldWriter( fieldName, structField );
-            for( auto & dt : created.writer -> dataTypes() )
-                arrowFields.push_back( std::make_shared<::arrow::Field>( fieldName, dt ) );
-            m_writers.push_back( std::move( created.writer ) );
-        }
+        auto fieldName = it.key();
+        auto colName   = it.value<std::string>();
+        auto structField = structMeta -> field( fieldName );
+        CSP_TRUE_OR_THROW_RUNTIME( structField != nullptr,
+            "Struct field '" << fieldName << "' not found on struct type '" << structMeta -> name() << "'" );
+
+        // Skip DIALECT_GENERIC fields (handled by custom writers)
+        if( structField -> type() -> type() == CspType::Type::DIALECT_GENERIC )
+            continue;
+
+        auto created = createFieldWriter( colName, structField );
+        for( auto & dt : created.writer -> dataTypes() )
+            arrowFields.push_back( std::make_shared<::arrow::Field>( colName, dt ) );
+        m_writers.push_back( std::move( created.writer ) );
     }
 
     // Append custom writers and their columns to schema

cpp/csp/python/NumpyConversions.cpp

@@ -202,5 +202,66 @@ void validateNumpyTypeVsCspType( const CspTypePtr & type, char numpy_type_char )
     }
 }
 
+DialectGenericType numpyReshape( DialectGenericType flatData, const std::vector<int64_t> & dims )
+{
+    // PyArray_Newshape accepts a PyArray_Dims struct directly, avoiding a numpy array allocation for dims.
+    // Stack-allocate the dims for typical 2-8 dimensional arrays.
+    npy_intp dimsBuf[8];
+    npy_intp ndims = static_cast<npy_intp>( dims.size() );
+    npy_intp * dimsPtr;
+
+    std::vector<npy_intp> heapDims;
+    if( ndims <= 8 )
+    {
+        for( npy_intp i = 0; i < ndims; ++i )
+            dimsBuf[i] = static_cast<npy_intp>( dims[i] );
+        dimsPtr = dimsBuf;
+    }
+    else
+    {
+        heapDims.resize( ndims );
+        for( npy_intp i = 0; i < ndims; ++i )
+            heapDims[i] = static_cast<npy_intp>( dims[i] );
+        dimsPtr = heapDims.data();
+    }
+
+    PyArray_Dims newshape{ dimsPtr, static_cast<int>( ndims ) };
+    auto * flatPyArr = reinterpret_cast<PyArrayObject *>( toPythonBorrowed( flatData ) );
+    PyObjectPtr reshaped{ PyObjectPtr::own(
+        reinterpret_cast<PyObject *>( PyArray_Newshape( flatPyArr, &newshape, NPY_CORDER ) ) ) };
+    if( !reshaped.get() )
+        CSP_THROW( PythonPassthrough, "" );
+
+    return fromPython<DialectGenericType>( reshaped.get() );
+}
+
+std::vector<int64_t> numpyShape( DialectGenericType ndarray )
+{
+    auto * pyArr = reinterpret_cast<PyArrayObject *>( toPythonBorrowed( ndarray ) );
+    int ndim = PyArray_NDIM( pyArr );
+    npy_intp * shape = PyArray_SHAPE( pyArr );
+    std::vector<int64_t> result( ndim );
+    for( int i = 0; i < ndim; ++i )
+        result[i] = shape[i];
+    return result;
+}
+
+int npyTypeFromPyType( DialectGenericType pyTypeObj )
+{
+    auto & cspType = pyTypeAsCspType( toPythonBorrowed( pyTypeObj ) );
+
+    return csp::PartialSwitchCspType<csp::CspType::Type::DOUBLE, csp::CspType::Type::INT64,
+                                     csp::CspType::Type::BOOL, csp::CspType::Type::STRING,
+                                     csp::CspType::Type::DATETIME, csp::CspType::Type::TIMEDELTA,
+                                     csp::CspType::Type::DATE>::invoke(
+        cspType.get(),
+        []( auto tag ) -> int
+        {
+            using CValueType = typename decltype( tag )::type;
+            return NPY_TYPE<CValueType>::value;
+        }
+    );
+}
+
 
 }

cpp/csp/python/NumpyConversions.h

@@ -24,6 +24,9 @@ template<> struct NPY_TYPE<int64_t>  { static const int value = NPY_LONGLONG; };
 template<> struct NPY_TYPE<uint64_t> { static const int value = NPY_ULONGLONG; };
 template<> struct NPY_TYPE<double>   { static const int value = NPY_DOUBLE; };
 template<> struct NPY_TYPE<std::string>   { static const int value = NPY_UNICODE; };
+template<> struct NPY_TYPE<DateTime>      { static const int value = NPY_DATETIME; };
+template<> struct NPY_TYPE<TimeDelta>     { static const int value = NPY_TIMEDELTA; };
+template<> struct NPY_TYPE<Date>          { static const int value = NPY_DATETIME; };
 
 template<typename T> struct is_native  { static constexpr bool value = false; };
 template<> struct is_native<bool>      { static constexpr bool value = true; };
@@ -232,14 +235,23 @@ PyObject * valuesAtIndexToNumpy( ValueType valueType, const csp::TimeSeriesProvi
                                  autogen::TimeIndexPolicy startPolicy, autogen::TimeIndexPolicy endPolicy,
                                  DateTime startDt = DateTime::NONE(), DateTime endDt = DateTime::NONE() );
 
-int64_t scalingFromNumpyDtUnit( NPY_DATETIMEUNIT base );
-NPY_DATETIMEUNIT datetimeUnitFromDescr( PyArray_Descr* descr );
+CSPIMPL_EXPORT int64_t scalingFromNumpyDtUnit( NPY_DATETIMEUNIT base );
+CSPIMPL_EXPORT NPY_DATETIMEUNIT datetimeUnitFromDescr( PyArray_Descr* descr );
 
 // for getting strings from elems of numpy arrays of strings
 void stringFromNumpyStr( void* data, std::string& out, char numpy_type, int elem_size_bytes );
 
 void validateNumpyTypeVsCspType( const CspTypePtr & type, char numpy_type_char );
 
+// Reshape a flat 1D numpy array into an NDArray using the given dimensions.
+CSPIMPL_EXPORT DialectGenericType numpyReshape( DialectGenericType flatData, const std::vector<int64_t> & dims );
+
+// Extract shape from a numpy NDArray as a vector of dimension sizes.
+CSPIMPL_EXPORT std::vector<int64_t> numpyShape( DialectGenericType ndarray );
+
+// Convert a Python type object (passed as DialectGenericType) to an NPY type constant.
+CSPIMPL_EXPORT int npyTypeFromPyType( DialectGenericType pyTypeObj );
+
 }
 
 #endif

cpp/csp/python/adapters/ArrowCppNodes.cpp

@@ -4,6 +4,8 @@
 // RecordBatches are transported across the Python/C++ boundary as PyCapsules
 // using the Arrow C Data Interface.
 
+// Must include numpy first without NO_IMPORT_ARRAY to define PyArray_API in this TU
+#include <numpy/ndarrayobject.h>
 #include <csp/adapters/arrow/RecordBatchToStruct.h>
 #include <csp/adapters/arrow/StructToRecordBatch.h>
 #include <csp/engine/CppNode.h>
@@ -13,6 +15,8 @@
 #include <csp/python/InitHelper.h>
 #include <csp/python/PyCppNode.h>
 #include <csp/python/PyNodeWrapper.h>
+#include <csp/python/adapters/ArrowNumpyListReader.h>
+#include <csp/python/adapters/ArrowNumpyListWriter.h>
 #include <arrow/c/abi.h>
 #include <arrow/c/bridge.h>
 #include <arrow/record_batch.h>
@@ -67,12 +71,45 @@ DECLARE_CPPNODE( record_batches_to_struct )
 
         // Parse properties
         auto & props = properties.value();
-        DictionaryPtr fieldMap;
-        if( props -> exists( "field_map" ) )
-            fieldMap = props -> get<DictionaryPtr>( "field_map" );
+        auto fieldMap = props -> get<DictionaryPtr>( "field_map" );
+
+        // Optional: mapping of arrow_col_name -> dims_col_name for NDArray columns
+        DictionaryPtr numpyDimensionNames;
+        if( props -> exists( "numpy_dimension_names" ) )
+            numpyDimensionNames = props -> get<DictionaryPtr>( "numpy_dimension_names" );
 
         auto structMeta = cls.value();
-        s_converter = std::make_unique<RecordBatchToStructConverter>( s_schema, structMeta, fieldMap );
+
+        // Build FieldReader entries for numpy array fields
+        std::vector<std::unique_ptr<FieldReader>> customReaders;
+
+        if( props -> exists( "numpy_fields" ) )
+        {
+            auto numpyFields = props -> get<DictionaryPtr>( "numpy_fields" );
+
+            for( auto it = numpyFields -> begin(); it != numpyFields -> end(); ++it )
+            {
+                auto colName   = it.key();
+                auto fieldName = it.value<std::string>();
+                auto structField = structMeta -> field( fieldName );
+                CSP_TRUE_OR_THROW_RUNTIME( structField != nullptr,
+                                           "Struct field '" << fieldName << "' not found on struct type '" << structMeta -> name() << "'" );
+
+                if( numpyDimensionNames && numpyDimensionNames -> exists( colName ) )
+                {
+                    auto dimsColName = numpyDimensionNames -> get<std::string>( colName );
+                    customReaders.push_back(
+                        csp::python::createNumpyNDArrayReader( s_schema, colName, dimsColName, structField, csp::python::numpyReshape ) );
+                }
+                else
+                {
+                    customReaders.push_back(
+                        csp::python::createNumpyArrayReader( s_schema, colName, structField ) );
+                }
+            }
+        }
+
+        s_converter = std::make_unique<RecordBatchToStructConverter>( s_schema, structMeta, fieldMap, std::move( customReaders ) );
     }
 
     INVOKE()
@@ -150,11 +187,46 @@ DECLARE_CPPNODE( struct_to_record_batches )
 
         s_maxBatchSize = props -> get<int64_t>( "max_batch_size", 0 );
 
-        DictionaryPtr fieldMap;
-        if( props -> exists( "field_map" ) )
-            fieldMap = props -> get<DictionaryPtr>( "field_map" );
+        auto fieldMap = props -> get<DictionaryPtr>( "field_map" );
+
+        DictionaryPtr numpyDimensionNames;
+        if( props -> exists( "numpy_dimension_names" ) )
+            numpyDimensionNames = props -> get<DictionaryPtr>( "numpy_dimension_names" );
+
+        // Build FieldWriter entries for numpy array fields
+        std::vector<std::unique_ptr<FieldWriter>> customWriters;
 
-        s_converter = std::make_unique<StructToRecordBatchConverter>( structMeta, fieldMap );
+        if( props -> exists( "numpy_fields" ) )
+        {
+            auto numpyFields = props -> get<DictionaryPtr>( "numpy_fields" );
+            auto numpyElementTypes = props -> get<DictionaryPtr>( "numpy_element_types" );
+
+            for( auto it = numpyFields -> begin(); it != numpyFields -> end(); ++it )
+            {
+                auto colName   = it.key();
+                auto fieldName = it.value<std::string>();
+                auto structField = structMeta -> field( fieldName );
+                CSP_TRUE_OR_THROW_RUNTIME( structField != nullptr,
+                                           "Struct field '" << fieldName << "' not found on struct type '" << structMeta -> name() << "'" );
+
+                auto pyTypeObj = numpyElementTypes -> get<DialectGenericType>( fieldName );
+                int npyType = csp::python::npyTypeFromPyType( pyTypeObj );
+
+                if( numpyDimensionNames && numpyDimensionNames -> exists( colName ) )
+                {
+                    auto dimsColName = numpyDimensionNames -> get<std::string>( colName );
+                    customWriters.push_back(
+                        csp::python::createNumpyNDArrayWriter( colName, dimsColName, structField, npyType, csp::python::numpyShape ) );
+                }
+                else
+                {
+                    customWriters.push_back(
+                        csp::python::createNumpyArrayWriter( colName, structField, npyType ) );
+                }
+            }
+        }
+
+        s_converter = std::make_unique<StructToRecordBatchConverter>( structMeta, fieldMap, std::move( customWriters ) );
     }
 
     INVOKE()
@@ -196,3 +268,18 @@ EXPORT_CPPNODE( struct_to_record_batches );
 
 REGISTER_CPPNODE( csp::cppnodes, record_batches_to_struct );
 REGISTER_CPPNODE( csp::cppnodes, struct_to_record_batches );
+
+// Initialize numpy array API for this translation unit
+static bool _init_numpy = []()
+{
+    csp::python::InitHelper::instance().registerCallback(
+        []( PyObject * module ) -> bool
+        {
+            import_array1( false );
+            csp::python::registerNumpyListFieldReaderFactory();
+            csp::python::registerNumpyListFieldWriterFactory();
+            return true;
+        }
+    );
+    return true;
+}();