Support NumpyNDArray, Numpy1DArray in record_batches_to_struct/struct_to_record_batches

Signed-off-by: Arham Chopra <[email protected]>

Author: arhamchopra

cpp/csp/adapters/parquet/ParquetOutputAdapter.cpp

@@ -127,10 +127,10 @@ ListColumnParquetOutputHandler::ListColumnParquetOutputHandler( Engine *engine,
                                                             listWriterInterface ) )
 {
     m_valueHandler = std::make_unique<ValueHandler>(
-            [ this ]( const TimeSeriesProvider *input )
+            [ this ]( const DialectGenericType& input )
             {
                 static_cast<ListColumnArrayBuilder *>(this -> m_columnArrayBuilder.get())
-                        -> setValue( input -> lastValueTyped<DialectGenericType>() );
+                        -> setValue( input );
             } );
 }
 
@@ -176,7 +176,7 @@ std::shared_ptr<::arrow::ArrayBuilder> ListColumnParquetOutputHandler::createVal
 
 void ListColumnParquetOutputAdapter::executeImpl()
 {
-    ( *m_valueHandler )( input() );
+    ( *m_valueHandler )( input() -> lastValueTyped<DialectGenericType>() );
     m_parquetWriter.scheduleEndCycleEvent();
 }
 

cpp/csp/adapters/parquet/ParquetOutputAdapter.h

@@ -93,18 +93,23 @@ class ListColumnParquetOutputHandler : public ParquetOutputHandler
         static_cast<ColumnBuilderType *>(this -> m_columnArrayBuilder.get()) -> setValue( value );
     }
 
-    void writeValueFromTs( const TimeSeriesProvider *input ) override final
+    void writeValueFromArgs( const DialectGenericType& input )
     {
         ( *m_valueHandler )( input );
     }
 
+    void writeValueFromTs( const TimeSeriesProvider *input ) override final
+    {
+        ( *m_valueHandler )( input -> lastValueTyped<DialectGenericType>() );
+    }
+
 private:
     std::shared_ptr<arrow::ArrayBuilder> createValueBuilder( const CspTypePtr &elemType,
                                                              DialectGenericListWriterInterface::Ptr &listWriterInterface );
 
 
 protected :
-    using ValueHandler = std::function<void( const TimeSeriesProvider * )>;
+    using ValueHandler = std::function<void( const DialectGenericType& )>;
 
     std::unique_ptr<ValueHandler>           m_valueHandler;
     std::shared_ptr<ListColumnArrayBuilder> m_columnArrayBuilder;

cpp/csp/adapters/parquet/ParquetReaderColumnAdapter.cpp

@@ -749,6 +749,24 @@ void BaseListColumnAdapter<ArrowListArrayType, ValueArrayType, ValueType>::addSu
                                                                << " in file " << m_parquetReader.getCurFileOrTableName() );
 }
 
+template< typename ArrowListArrayType, typename ValueArrayType, typename ValueType>
+void BaseListColumnAdapter<ArrowListArrayType, ValueArrayType, ValueType>::addSubscriber( csp::adapters::utils::ValueDispatcher<const DialectGenericType&>::SubscriberType subscriber, std::optional<utils::Symbol> symbol, const DialectGenericListReaderInterface::Ptr &listReader )
+{
+    CSP_TRUE_OR_THROW_RUNTIME( m_listReader == nullptr,
+                               "Trying to subscribe list column in parquet reader more than once, this is not supported" );
+    CSP_TRUE_OR_THROW_RUNTIME( listReader != nullptr,
+                               "Trying to subscribe list column in parquet reader with null listReader" );
+    m_listReader = std::dynamic_pointer_cast<TypedDialectGenericListReaderInterface<ValueType>>( listReader );
+    CSP_TRUE_OR_THROW_RUNTIME( m_listReader != nullptr,
+                               "Subscribed to parquet column " << getColumnName() << " with type "
+                                                               << "NumpyArray[" << listReader -> getValueType() -> type().asString()
+                                                               << "] while "
+                                                               << " column type in file is NumpyArray["
+                                                               << getContainerValueType() -> type().asString() << "]"
+                                                               << " in file " << m_parquetReader.getCurFileOrTableName() );
+    m_dispatcher.addSubscriber( subscriber, symbol );
+}
+
 template< typename ArrowListArrayType, typename ValueArrayType, typename ValueType >
 void NativeListColumnAdapter<ArrowListArrayType, ValueArrayType, ValueType>::readCurValue()
 {

cpp/csp/adapters/parquet/ParquetReaderColumnAdapter.h

@@ -48,6 +48,8 @@ class ParquetColumnAdapter
     virtual void addSubscriber( ManagedSimInputAdapter *inputAdapter, std::optional<utils::Symbol> symbol = {} ) = 0;
     // NOTE: This API is only defined for ListType Column Adapters
     virtual void addSubscriber( ManagedSimInputAdapter *inputAdapter, std::optional<utils::Symbol> symbol, const DialectGenericListReaderInterface::Ptr &listReader ) = 0;
+    // NOTE: This API is only used to add subscriber for ListType column adapters in cases where there is no ManagedSimInputAdapter
+    virtual void addSubscriber( csp::adapters::utils::ValueDispatcher<const DialectGenericType&>::SubscriberType subscriber, std::optional<utils::Symbol> symbol, const DialectGenericListReaderInterface::Ptr &listReader ) = 0;
 
     virtual void dispatchValue( const utils::Symbol *symbol ) = 0;
 
@@ -124,6 +126,10 @@ class MissingColumnAdapter : public ParquetColumnAdapter
 
     virtual void addSubscriber( ManagedSimInputAdapter *inputAdapter, std::optional<utils::Symbol> symbol = {} ) override {};
     virtual void addSubscriber( ManagedSimInputAdapter *inputAdapter, std::optional<utils::Symbol> symbol, const DialectGenericListReaderInterface::Ptr &listReader ) override {};
+    virtual void addSubscriber( csp::adapters::utils::ValueDispatcher<const DialectGenericType&>::SubscriberType subscriber, std::optional<utils::Symbol> symbol, const DialectGenericListReaderInterface::Ptr &listReader ) override
+    {
+        CSP_THROW(TypeError, "Trying to add DIALECT_GENERIC subscriber on non container type");
+    }
 
     virtual void dispatchValue( const utils::Symbol *symbol ) override {};
 
@@ -172,6 +178,10 @@ class BaseTypedColumnAdapter : public ParquetColumnAdapter
     void addSubscriber( ManagedSimInputAdapter *inputAdapter, std::optional<utils::Symbol> symbol = {} ) override;
     void addSubscriber( ManagedSimInputAdapter *inputAdapter,
                         std::optional<utils::Symbol> symbol, const DialectGenericListReaderInterface::Ptr &listReader ) override;
+    virtual void addSubscriber( csp::adapters::utils::ValueDispatcher<const DialectGenericType&>::SubscriberType subscriber, std::optional<utils::Symbol> symbol, const DialectGenericListReaderInterface::Ptr &listReader ) override
+    {
+        CSP_THROW(TypeError, "Trying to add DIALECT_GENERIC subscriber on non container type");
+    }
     void dispatchValue( const utils::Symbol *symbol ) override;
 
     void ensureType( CspType::Ptr cspType ) override;
@@ -288,9 +298,11 @@ class BaseListColumnAdapter : public BaseTypedColumnAdapter<DialectGenericType,
 public:
     using BaseTypedColumnAdapter<DialectGenericType, ArrowListArrayType>::BaseTypedColumnAdapter;
     using BaseTypedColumnAdapter<DialectGenericType, ArrowListArrayType>::getColumnName;
+    using BaseTypedColumnAdapter<DialectGenericType, ArrowListArrayType>::m_dispatcher;
     void addSubscriber( ManagedSimInputAdapter *inputAdapter, std::optional<utils::Symbol> symbol = {} ) override;
     void addSubscriber( ManagedSimInputAdapter *inputAdapter,
                         std::optional<utils::Symbol> symbol, const DialectGenericListReaderInterface::Ptr &listReader ) override;
+    void addSubscriber( csp::adapters::utils::ValueDispatcher<const DialectGenericType&>::SubscriberType subscriber, std::optional<utils::Symbol> symbol, const DialectGenericListReaderInterface::Ptr &listReader ) override;
     CspTypePtr getNativeCspType() const override {return nullptr;}
     bool isListType() const override{ return true; };
     CspTypePtr getContainerValueType() const override{ return CspType::fromCType<ValueType>::type(); }

cpp/csp/python/CMakeLists.txt

@@ -36,6 +36,7 @@ set(CSPIMPL_PUBLIC_HEADERS
         PyBasketInputProxy.h
         PyBasketOutputProxy.h
         PyCppNode.h
+	 PyDialectGenericListsInterface.h
         PyEngine.h
         PyInputAdapterWrapper.h
         PyInputProxy.h