Review fixes, and cleanup

arhamchopra · arhamchopra · commit 9bf26033752a · 2026-06-17T10:48:50.000-04:00
Signed-off-by: Arham Chopra &lt;arham.chopra@cubistsystematic.com&gt;
diff --git a/cpp/csp/adapters/arrow/ArrowFieldWriter.cpp b/cpp/csp/adapters/arrow/ArrowFieldWriter.cpp
@@ -345,7 +345,11 @@ CreatedFieldWriter createFieldWriter(
             std::vector<std::shared_ptr<::arrow::ArrayBuilder>> childBuilders;
             std::vector<std::unique_ptr<FieldWriter>> childWriters;
 
-            // Use fieldNames() for stable insertion order (fields() is sorted for memory layout)
+            // Ordering convention: csp writes struct child columns in declaration order, using
+            // fieldNames() rather than fields() (which is sorted by struct memory layout). This keeps
+            // nested-struct child order consistent with top-level column order and stable against
+            // internal field packing. The reader matches children by name, so either order round-trips,
+            // but declaration order is the on-disk contract.
             for( auto & subFieldName : nestedMeta -> fieldNames() )
             {
                 auto subField = nestedMeta -> field( subFieldName );
diff --git a/cpp/csp/adapters/parquet/ArrowBackedArrayBuilder.cpp b/cpp/csp/adapters/parquet/ArrowBackedArrayBuilder.cpp
@@ -88,6 +88,13 @@ void ArrowBackedArrayBuilder::init( const std::string & columnName, const Struct
 {
     auto created = csp::adapters::arrow::createFieldWriter( columnName, field );
     m_writer = std::move( created.writer );
+    // Reserve the first chunk up front so the per-row append path doesn't reallocate as rows accumulate.
+    m_writer -> reserve( getChunkSize() );
+}
+
+void ArrowBackedArrayBuilder::reserve( int64_t numRows )
+{
+    m_writer -> reserve( numRows );
 }
 
 std::shared_ptr<::arrow::DataType> ArrowBackedArrayBuilder::getDataType()
@@ -132,6 +139,9 @@ std::shared_ptr<::arrow::Array> ArrowBackedArrayBuilder::buildArray()
     auto arrays = m_writer -> finish();
     CSP_TRUE_OR_THROW_RUNTIME( arrays.size() == 1,
         "ArrowBackedArrayBuilder expected 1 array from FieldWriter, got " << arrays.size() );
+    // finish() resets the underlying builder's capacity; re-reserve for the next chunk so the per-row
+    // append path stays allocation-free (and keeps UnsafeAppend within reserved capacity).
+    m_writer -> reserve( getChunkSize() );
     return arrays[0];
 }
 
diff --git a/cpp/csp/adapters/parquet/ArrowBackedArrayBuilder.h b/cpp/csp/adapters/parquet/ArrowBackedArrayBuilder.h
@@ -42,6 +42,7 @@ class ArrowBackedArrayBuilder : public ArrowSingleColumnArrayBuilder
     int64_t length() const override;
     void handleRowFinished() override;
     std::shared_ptr<::arrow::Array> buildArray() override;
+    void reserve( int64_t numRows ) override;
 
     // --- Scratch mode API ---
     // Get the scratch struct for value setting
diff --git a/cpp/csp/adapters/parquet/ArrowSingleColumnArrayBuilder.h b/cpp/csp/adapters/parquet/ArrowSingleColumnArrayBuilder.h
@@ -37,6 +37,10 @@ class ArrowSingleColumnArrayBuilder
     // Release the array from the currently built values
     virtual std::shared_ptr<::arrow::Array> buildArray() = 0;
 
+    // Pre-reserve builder capacity for numRows rows. Default no-op; fixed-width builders override to
+    // reserve their Arrow builder so the per-row append path avoids repeated reallocation.
+    virtual void reserve( int64_t /*numRows*/ ) {}
+
 private:
     const std::string   m_columnName;
     const std::uint32_t m_chunkSize;
diff --git a/cpp/csp/adapters/parquet/ParquetDictBasketOutputWriter.cpp b/cpp/csp/adapters/parquet/ParquetDictBasketOutputWriter.cpp
@@ -151,12 +151,15 @@ ParquetStructDictBasketOutputWriter::ParquetStructDictBasketOutputWriter( Parque
         : ParquetDictBasketOutputWriter( outputAdapterManager, columnName )
 {
 
-    // We don't support fieldMap for now, only default field map
+    // We don't support fieldMap for now, only default field map.
+    // Iterate fieldNames() (declaration order), NOT fields() (sorted by memory layout): csp writes
+    // struct columns -- both top-level and nested -- in declaration order, so the on-disk column order
+    // matches the struct definition and stays stable regardless of internal field packing.
     auto structMetaPtr = std::static_pointer_cast<const CspStructType>( cspTypePtr ) -> meta().get();
     DictionaryPtr dict = std::make_shared<Dictionary>();
-    for(auto&& field:structMetaPtr->fields())
+    for( auto && fieldName : structMetaPtr -> fieldNames() )
     {
-        dict->insert(field->fieldname(), columnName + "." + field->fieldname());
+        dict->insert( fieldName, columnName + "." + fieldName );
     }
     m_valueOutputAdapter = getStructOutputHandler( cspTypePtr, dict );
 }
diff --git a/cpp/csp/adapters/parquet/ParquetOutputAdapter.cpp b/cpp/csp/adapters/parquet/ParquetOutputAdapter.cpp
@@ -100,7 +100,8 @@ StructParquetOutputHandler::StructParquetOutputHandler( Engine *engine, ParquetW
             CSP_THROW( ValueError, "Struct field '" << structFieldName << "' not found" );
 
         if( structField -> type() -> type() == CspType::Type::DIALECT_GENERIC )
-            continue;
+            CSP_THROW( TypeError, "Writing of column " << structField -> fieldname()
+                << " of type DIALECT_GENERIC is not supported" );
 
         auto builder = createArrowBackedArrayBuilderForField( columnName, getChunkSize(), structField );
         m_arrowBuilders.push_back( builder.get() );
diff --git a/cpp/csp/adapters/parquet/ParquetWriter.cpp b/cpp/csp/adapters/parquet/ParquetWriter.cpp
@@ -215,12 +215,9 @@ void ParquetWriter::adoptMetadataFrom( ParquetWriter & source )
 
 void ParquetWriter::onEndCycle()
 {
-    // NOTE: when no file is open -- e.g. a filename_provider ticked "" to "pause" output -- this body is
-    // skipped, so handleRowFinished() does NOT run and each scratch column's isSet bit is never cleared
-    // for that cycle. If a column ticks *while paused* and then does NOT tick on the cycle the file
-    // reopens, the stale paused-window value is emitted instead of null (the dropped tick "leaks" into
-    // the first row after resume). This matches the previous writer's behaviour. If dropping ticks while
-    // paused should instead null them out, clear each builder's scratch/isSet here in the else path.
+    // When no file is open -- e.g. a filename_provider ticked "" to "pause" output -- this body is
+    // skipped, so handleRowFinished() does not run for that cycle. As a result, a value that ticked
+    // while paused is emitted on the first row after the file reopens, rather than being nulled.
     if( isFileOpen() ) [[likely]]
     {
         DateTime now;
diff --git a/cpp/csp/adapters/parquet/RecordBatchFileSink.cpp b/cpp/csp/adapters/parquet/RecordBatchFileSink.cpp
@@ -20,6 +20,7 @@
 
 namespace csp::adapters::parquet
 {
+
 namespace
 {
 
@@ -70,7 +71,7 @@ std::shared_ptr<::arrow::io::OutputStream> openOutputStream( const std::string &
 }
 
 FileWriter makeParquetFileWriter( const std::string & path, const std::shared_ptr<::arrow::Schema> & schema,
-                                  const std::string & compression, bool allowOverwrite )
+                                  const std::string & compression, bool allowOverwrite, bool useDictionary )
 {
     auto stream = openOutputStream( path, allowOverwrite );
 
@@ -80,6 +81,11 @@ FileWriter makeParquetFileWriter( const std::string & path, const std::shared_pt
     // Arrow 6.0) is always present, so there is no need to gate on the Arrow version or fall back to
     // the deprecated PARQUET_2_0.
     props.version( ::parquet::ParquetVersion::PARQUET_2_6 );
+    // Dictionary encoding is on by default. For high-cardinality numeric columns it is a large write-time
+    // cost (hash table build) and can even grow files; allow it to be turned off (trading file size for
+    // write throughput). Low-cardinality columns still benefit from leaving it on (the default).
+    if( !useDictionary )
+        props.disable_dictionary();
     ::parquet::ArrowWriterProperties::Builder arrowProps;
     arrowProps.store_schema(); // preserve arrow (file/column) schema metadata in the parquet file
 
@@ -204,16 +210,17 @@ FileWriter makeSplitWriter( const std::string & dir, const std::shared_ptr<::arr
 
 RecordBatchSink makeFileSink( bool writeArrowBinary, bool splitColumns,
                               const std::string & compression, bool allowOverwrite,
-                              std::function<void( const std::string & )> fileVisitor )
+                              std::function<void( const std::string & )> fileVisitor,
+                              bool useDictionary )
 {
     const std::string extension = writeArrowBinary ? ".arrow" : ".parquet";
 
     // Per-file factory: format choice is the only difference.
     FileWriterFactory perFile =
-        [writeArrowBinary, compression, allowOverwrite]( const std::string & path, const std::shared_ptr<::arrow::Schema> & schema )
+        [writeArrowBinary, compression, allowOverwrite, useDictionary]( const std::string & path, const std::shared_ptr<::arrow::Schema> & schema )
         {
             return writeArrowBinary ? makeIpcFileWriter( path, schema, compression, allowOverwrite )
-                                    : makeParquetFileWriter( path, schema, compression, allowOverwrite );
+                                    : makeParquetFileWriter( path, schema, compression, allowOverwrite, useDictionary );
         };
 
     // Layout choice: single file vs. one file per column.
diff --git a/cpp/csp/adapters/parquet/RecordBatchFileSink.h b/cpp/csp/adapters/parquet/RecordBatchFileSink.h
@@ -22,7 +22,8 @@ RecordBatchSink makeFileSink( bool writeArrowBinary,
                               bool splitColumns,
                               const std::string & compression,
                               bool allowOverwrite,
-                              std::function<void( const std::string & )> fileVisitor = {} );
+                              std::function<void( const std::string & )> fileVisitor = {},
+                              bool useDictionary = true );
 
 } // namespace csp::adapters::parquet
 
diff --git a/cpp/csp/python/adapters/parquetadapterimpl.cpp b/cpp/csp/python/adapters/parquetadapterimpl.cpp
@@ -685,6 +685,7 @@ csp::AdapterManager *create_parquet_output_adapter_manager( PyEngine *engine, co
     bool        splitColumns     = properties.get<bool>( "split_columns_to_files" );
     bool        allowOverwrite   = properties.get<bool>( "allow_overwrite" );
     std::string compression      = properties.get<std::string>( "compression" );
+    bool        useDictionary    = properties.get( "write_dictionary", true );
 
     // 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
@@ -707,13 +708,13 @@ csp::AdapterManager *create_parquet_output_adapter_manager( PyEngine *engine, co
     }
 
     auto * mgr = engine -> engine() -> createOwnedObject<ParquetOutputAdapterManager>( properties );
-    mgr -> setSink( makeFileSink( writeArrowBinary, splitColumns, compression, allowOverwrite, fileVisitor ) );
+    mgr -> setSink( makeFileSink( writeArrowBinary, splitColumns, compression, allowOverwrite, fileVisitor, useDictionary ) );
 
     // Dict-basket writers always write split-column files (one per value/symbol/index column),
     // without a file visitor.
-    mgr -> setSinkFactory( [writeArrowBinary, compression, allowOverwrite]( const std::string & ) -> RecordBatchSink
+    mgr -> setSinkFactory( [writeArrowBinary, compression, allowOverwrite, useDictionary]( const std::string & ) -> RecordBatchSink
     {
-        return makeFileSink( writeArrowBinary, /*splitColumns=*/true, compression, allowOverwrite, {} );
+        return makeFileSink( writeArrowBinary, /*splitColumns=*/true, compression, allowOverwrite, {}, useDictionary );
     } );
 
     return mgr;
diff --git a/csp/adapters/output_adapters/parquet.py b/csp/adapters/output_adapters/parquet.py
@@ -30,6 +30,8 @@ class ParquetOutputConfig(Struct):
     batch_size: int = 2**15
     compression: str
     write_arrow_binary: bool = False  # If true will write output as binary arrow data rather than parquet
+    write_dictionary: bool = True  # If false, disables parquet dictionary encoding (faster writes for
+    # high-cardinality numeric columns, at the cost of larger files; low-cardinality data benefits from True)
 
     def resolve_compression(self):
         if not hasattr(self, "compression"):
@@ -107,6 +109,7 @@ def __init__(
             "batch_size": config.batch_size,
             "compression": config.compression,
             "write_arrow_binary": config.write_arrow_binary,
+            "write_dictionary": config.write_dictionary,
             "split_columns_to_files": split_columns_to_files,
             "file_metadata": file_metadata,
             "column_metadata": column_metadata,
diff --git a/csp/tests/adapters/test_parquet_output.py b/csp/tests/adapters/test_parquet_output.py
@@ -281,6 +281,73 @@ def g():
             self.assertEqual(df["f7"].iloc[0], "eight")
             self.assertEqual(df["f8"].iloc[0], True)
 
+    def test_struct_with_generic_field_raises(self):
+        """Publishing a struct with a DIALECT_GENERIC (object) field raises rather than silently
+        dropping the column.
+        """
+
+        class WithObj(csp.Struct):
+            x: int
+            obj: object
+
+        with tempfile.TemporaryDirectory() as d:
+            fname = os.path.join(d, "out.parquet")
+
+            @csp.graph
+            def g():
+                writer = ParquetWriter(fname, "csp_timestamp", config=ParquetOutputConfig(allow_overwrite=True))
+                writer.publish_struct(csp.const(WithObj(x=1, obj={"a": 1})))
+
+            with self.assertRaisesRegex(Exception, "DIALECT_GENERIC"):
+                csp.run(g, starttime=START, endtime=timedelta(seconds=2))
+
+    def test_nested_struct_child_field_order_is_declaration_order(self):
+        """Nested-struct child columns are written in declaration order (not struct memory-layout
+        order), consistent with top-level column order. Pins the on-disk schema shape: a_flag (a 1-byte
+        bool declared first) must come first, which memory-layout packing -- which front-loads the
+        8-byte fields -- would never produce.
+        """
+
+        class Inner(csp.Struct):
+            a_flag: bool
+            z_price: float
+            m_count: int
+
+        class Outer(csp.Struct):
+            o_flag: bool
+            o_price: float
+            inner: Inner
+
+        with tempfile.TemporaryDirectory() as d:
+            fname = os.path.join(d, "out.parquet")
+            v = Outer(o_flag=True, o_price=1.5, inner=Inner(a_flag=False, z_price=2.5, m_count=3))
+
+            @csp.graph
+            def g_write():
+                writer = ParquetWriter(fname, "csp_timestamp", config=ParquetOutputConfig(allow_overwrite=True))
+                writer.publish_struct(csp.const(v))
+
+            csp.run(g_write, starttime=START, endtime=timedelta(seconds=2))
+
+            schema = pyarrow.parquet.read_schema(fname)
+            # Top-level columns in declaration order (after the timestamp column).
+            self.assertEqual(schema.names, ["csp_timestamp", "o_flag", "o_price", "inner"])
+            # Nested struct child fields in declaration order.
+            inner_type = schema.field("inner").type
+            self.assertEqual(
+                [inner_type.field(i).name for i in range(inner_type.num_fields)],
+                ["a_flag", "z_price", "m_count"],
+            )
+
+            # And it round-trips back to the original struct.
+            @csp.graph
+            def g_read():
+                reader = ParquetReader(fname, time_column="csp_timestamp")
+                csp.add_graph_output("data", reader.subscribe_all(Outer))
+
+            res = csp.run(g_read, starttime=START, endtime=START + timedelta(seconds=2))
+            self.assertEqual(res["data"][0][1], v)
+
 
 class TestOutputFileRotation(unittest.TestCase):
     """Test file rotation via filename_provider."""
@@ -1067,6 +1134,45 @@ def g():
             with self.assertRaisesRegex(Exception, "snappy"):
                 csp.run(g, starttime=START, endtime=timedelta(seconds=3))
 
+    def test_write_dictionary_disabled(self):
+        """write_dictionary=False disables parquet dictionary encoding; data still round-trips and the
+        column uses a non-dictionary encoding (default write_dictionary=True keeps dictionary on).
+        """
+        values = [float(i) for i in range(2000)]
+        with tempfile.TemporaryDirectory() as d:
+            on = os.path.join(d, "dict_on.parquet")
+            off = os.path.join(d, "dict_off.parquet")
+
+            def write(fname, use_dict):
+                @csp.graph
+                def g():
+                    config = ParquetOutputConfig(allow_overwrite=True, write_dictionary=use_dict)
+                    writer = ParquetWriter(fname, "csp_timestamp", config=config)
+                    writer.publish("x", csp.curve(float, [(timedelta(seconds=i + 1), v) for i, v in enumerate(values)]))
+
+                csp.run(g, starttime=START, endtime=timedelta(seconds=len(values) + 2))
+
+            write(on, True)
+            write(off, False)
+
+            # Both round-trip identically.
+            self.assertEqual(pandas.read_parquet(off)["x"].tolist(), values)
+            self.assertEqual(pandas.read_parquet(on)["x"].tolist(), values)
+
+            # The 'x' column encodings differ: dictionary on (default) uses PLAIN_DICTIONARY/RLE_DICTIONARY;
+            # with write_dictionary=False it must not.
+            def x_encodings(fname):
+                md = pyarrow.parquet.ParquetFile(fname).metadata
+                return set(md.row_group(0).column(1).encodings)
+
+            off_enc = x_encodings(off)
+            self.assertFalse(
+                any("DICTIONARY" in e for e in off_enc), f"dict-off file unexpectedly dictionary-encoded: {off_enc}"
+            )
+            self.assertTrue(
+                any("DICTIONARY" in e for e in x_encodings(on)), "dict-on (default) should use dictionary encoding"
+            )
+
 
 class TestOutputBatchSize(unittest.TestCase):
     """Test batch_size controls row group flushing."""
@@ -1212,10 +1318,7 @@ def g():
             self.assertEqual(reader.schema.metadata[b"source"], b"ipc_test")
 
     def test_split_and_single_metadata_equivalent(self):
-        """The PR's headline claim: file- and column-level metadata (and field types) are identical
-        between single-file and split-column modes. Writes the same data both ways and compares the
-        per-column schema field plus file metadata directly.
-        """
+        """Metadata and field types are identical between single-file and split-column modes."""
         file_md = {"author": "me", "version": "2.0"}
         col_md = {"x": {"units": "kg", "source": "sensor"}}
         curve = [(timedelta(seconds=1), 1), (timedelta(seconds=2), 2)]
@@ -1304,11 +1407,7 @@ class TestOutputEdgeCases(unittest.TestCase):
     """Edge cases and special scenarios."""
 
     def test_bare_relative_filename_no_directory(self):
-        """Writing to a bare relative filename (no directory component) must work.
-
-        Regression: the C++ sink called mkdir(dirname(path)); dirname("out.parquet")
-        is "" and mkdir("") fails with 'Invalid argument'.
-        """
+        """Writing to a bare relative filename with no directory component succeeds."""
         with tempfile.TemporaryDirectory() as d:
             cwd = os.getcwd()
             os.chdir(d)

Original file line number	Diff line number	Diff line change
`@@ -151,12 +151,15 @@ ParquetStructDictBasketOutputWriter::ParquetStructDictBasketOutputWriter( Parque`
`151`	`151`	`: ParquetDictBasketOutputWriter( outputAdapterManager, columnName )`
`152`	`152`	`{`
`153`	`153`
`154`		`- // We don't support fieldMap for now, only default field map`
	`154`	`+ // We don't support fieldMap for now, only default field map.`
	`155`	`+ // Iterate fieldNames() (declaration order), NOT fields() (sorted by memory layout): csp writes`
	`156`	`+ // struct columns -- both top-level and nested -- in declaration order, so the on-disk column order`
	`157`	`+ // matches the struct definition and stays stable regardless of internal field packing.`
`155`	`158`	`auto structMetaPtr = std::static_pointer_cast<const CspStructType>( cspTypePtr ) -> meta().get();`
`156`	`159`	`DictionaryPtr dict = std::make_shared<Dictionary>();`
`157`		`- for(auto&& field:structMetaPtr->fields())`
	`160`	`+ for( auto && fieldName : structMetaPtr -> fieldNames() )`
`158`	`161`	`{`
`159`		`- dict->insert(field->fieldname(), columnName + "." + field->fieldname());`
	`162`	`+ dict->insert( fieldName, columnName + "." + fieldName );`
`160`	`163`	`}`
`161`	`164`	`m_valueOutputAdapter = getStructOutputHandler( cspTypePtr, dict );`
`162`	`165`	`}`