Parquet output: minor cleanups (scratch reset, batch guard, docs)

Encapsulate the scratch isSet reset behind StructField::clearIsSet, add a
debug-only length assert in buildRecordBatch, and document that file_visitor
runs synchronously on the engine thread.

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

Author: arhamchopra

cpp/csp/adapters/parquet/ArrowBackedArrayBuilder.cpp

@@ -114,8 +114,7 @@ void ArrowBackedArrayBuilder::handleRowFinished()
         m_writer -> writeNext( m_scratch.get() );
 
         // Clear the isSet bit so next cycle starts with null
-        uint8_t * mask = reinterpret_cast<uint8_t *>( m_scratch.get() ) + m_field -> maskOffset();
-        *mask &= ~m_field -> maskBitMask();
+        m_field -> clearIsSet( m_scratch.get() );
     }
     else
     {

cpp/csp/adapters/parquet/ParquetWriter.cpp

@@ -238,7 +238,13 @@ std::shared_ptr<::arrow::RecordBatch> ParquetWriter::buildRecordBatch()
     std::vector<std::shared_ptr<::arrow::Array>> columns;
     columns.reserve( m_columnBuilders.size() );
     for( auto && builder : m_columnBuilders )
-        columns.push_back( builder -> buildArray() );
+    {
+        auto column = builder -> buildArray();
+        // Every column must contain exactly the rows accumulated this chunk; a mismatch would build a
+        // corrupt RecordBatch (debug-only guard, compiled out in release).
+        CSP_ASSERT( column -> length() == static_cast<std::int64_t>( m_curChunkSize ) );
+        columns.push_back( std::move( column ) );
+    }
     return ::arrow::RecordBatch::Make( m_schema, m_curChunkSize, std::move( columns ) );
 }
 

cpp/csp/engine/Struct.h

@@ -77,6 +77,16 @@ class StructField
         (*m) |= m_noneMaskBitMask;
     }
 
+    //Marks this field "not set" without releasing/cleaning its current value.  The same capability is
+    //already reachable through the public maskOffset()/maskBitMask() accessors; this just encapsulates the
+    //bit poke.  Used by callers that reuse a struct as a per-row scratch buffer (e.g. the parquet
+    //ArrowBackedArrayBuilder appends the scratch value, then clears the bit so the next row defaults to null).
+    void clearIsSet( Struct * s ) const
+    {
+        uint8_t * m = reinterpret_cast<uint8_t *>( s ) + m_maskOffset;
+        (*m) &= ~m_maskBitMask;
+    }
+
     //copy methods need not deal with mask set/unset, only copy values
     virtual void copyFrom( const Struct * src, Struct * dest ) const = 0;
 
@@ -121,12 +131,6 @@ class StructField
         return reinterpret_cast<uint8_t *>( s ) + m_offset;
     }
 
-    void clearIsSet( Struct * s ) const
-    {
-        uint8_t * m = reinterpret_cast<uint8_t *>( s ) + m_maskOffset;
-        (*m) &= ~m_maskBitMask;
-    }
-
     void clearIsNone( Struct * s ) const
     {
         uint8_t * m = reinterpret_cast<uint8_t *>( s ) + m_maskOffset;

cpp/csp/python/adapters/parquetadapterimpl.cpp

@@ -686,8 +686,9 @@ csp::AdapterManager *create_parquet_output_adapter_manager( PyEngine *engine, co
     bool        allowOverwrite   = properties.get<bool>( "allow_overwrite" );
     std::string compression      = properties.get<std::string>( "compression" );
 
-    // Optional Python file_visitor: wrapped as a plain C++ callback, invoked (with the GIL
-    // held during the run) after each file is closed.  Only the main sink visits files.
+    // Optional Python file_visitor: wrapped as a plain C++ callback, invoked synchronously (with the
+    // GIL held during the run) after each file is closed, so a slow visitor blocks the engine.  Only
+    // the main sink visits files.
     std::function<void( const std::string & )> fileVisitor;
     DialectGenericType visitorDG;
     if( properties.tryGet( "file_visitor", visitorDG ) )

csp/adapters/output_adapters/parquet.py

@@ -73,7 +73,10 @@ def __init__(
         file_name must be a folder into which the data will be written.
         :param file_metadata: optional str:str dict that will get written as file-level metadata
         :param column_metadata: optional dict of column : { str:str} that will get written as column-level metadata
-        :param file_visitor: optional callable that will be called, after a file is written, with the file name.
+        :param file_visitor: optional callable invoked with the file path after each file is closed (on rotation
+        and at shutdown). It runs synchronously on the engine thread with the GIL held, so a slow visitor (e.g.
+        uploading to remote storage) blocks the engine for its duration -- offload heavy work to a background
+        queue/thread if that matters. An exception raised by the visitor propagates out of csp.run.
         """
         super().__init__()
         config = ParquetOutputConfig() if config is None else config.copy()