Merge pull request #701 from Point72/ac/arrow_nodes_numpy_support

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

Author: arhamchopra

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; };
@@ -93,7 +96,7 @@ inline PyObject * as_nparray( const csp::TimeSeriesProvider * ts, const TickBuff
 {
     int32_t len = startIndex - endIndex + 1;
     if( len <= 0 || !ts -> valid() || ( !tickBuffer && endIndex != 0 ) )
-        return empty_array( NPY_TYPE<DateTime>::value );
+        return empty_array( NPY_OBJECT );
 
     DateTime * values = getValues( tickBuffer, lastValue, startIndex, endIndex, &len, tailPadding );
     npy_intp dims[]{ ( npy_intp ) len };
@@ -104,6 +107,7 @@ inline PyObject * as_nparray( const csp::TimeSeriesProvider * ts, const TickBuff
         auto date_type = PyPtr<PyObject>::own( PyUnicode_FromString( "<M8[ns]" ) );
         PyArray_DescrConverter( date_type.get(), &datetime_descr );
     }
+
     // PyArray_NewFromDescr steals a reference
     Py_INCREF( datetime_descr );
     auto * array = PyArray_NewFromDescr( &PyArray_Type, datetime_descr, 1, dims, nullptr, values, 0, nullptr );
@@ -116,7 +120,7 @@ inline PyObject * as_nparray( const csp::TimeSeriesProvider * ts, const TickBuff
 {
     int32_t len = startIndex - endIndex + 1;
     if( len <= 0 || !ts -> valid()|| ( !tickBuffer && endIndex != 0 ) )
-        return empty_array( NPY_TYPE<TimeDelta>::value );
+        return empty_array( NPY_OBJECT );
 
     TimeDelta * values = getValues( tickBuffer, lastValue, startIndex, endIndex, &len, tailPadding );
     npy_intp dims[]{ ( npy_intp ) len };
@@ -127,6 +131,7 @@ inline PyObject * as_nparray( const csp::TimeSeriesProvider * ts, const TickBuff
         auto timedelta_type = PyPtr<PyObject>::own( PyUnicode_FromString( "<m8[ns]" ) );
         PyArray_DescrConverter( timedelta_type.get(), &timedelta_descr );
     }
+
     Py_INCREF( timedelta_descr );
     auto * array = PyArray_NewFromDescr( &PyArray_Type, timedelta_descr, 1, dims, nullptr, values, 0, nullptr );
     PyArray_ENABLEFLAGS( ( PyArrayObject * ) array, NPY_ARRAY_OWNDATA);
@@ -232,14 +237,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