Do not closing and reopening parquet files when writing. Correct issue where writer was never finishing. Set default stream rows to 10,000.

Author: curtisalexander

readstat/Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "readstat"
-version = "0.9.1"
+version = "0.9.2"
 authors = ["Curtis Alexander <calex@calex.org>"]
 edition = "2021"
 description = "Rust wrapper of the ReadStat C library"

readstat/Cargo.toml

@@ -1,8 +1,7 @@
 #![allow(non_camel_case_types)]
 
 use colored::Colorize;
-use crossbeam::channel::bounded;
-// use crossbeam::channel::unbounded;
+use crossbeam::channel::unbounded;
 use log::debug;
 use path_abs::{PathAbs, PathInfo};
 use rayon::prelude::*;
@@ -33,7 +32,7 @@ pub use rs_path::ReadStatPath;
 pub use rs_write::ReadStatWriter;
 
 // Default rows to stream
-const STREAM_ROWS: u32 = 50000;
+const STREAM_ROWS: u32 = 10000;
 
 // StructOpt
 #[derive(StructOpt, Debug)]
@@ -339,8 +338,7 @@ pub fn run(rs: ReadStat) -> Result<(), Box<dyn Error + Send + Sync>> {
                     let offsets = build_offsets(total_rows_to_process, total_rows_to_stream)?;
 
                     // Create channels
-                    let (s, r) = bounded(rayon::current_num_threads());
-                    //let (s, r) = unbounded();
+                    let (s, r) = unbounded();
 
                     // Initialize writing
                     let mut wtr = ReadStatWriter::new();
@@ -416,7 +414,7 @@ pub fn run(rs: ReadStat) -> Result<(), Box<dyn Error + Send + Sync>> {
                     for (i, (d, rsp, pairs_cnt)) in r.iter().enumerate() {
                         wtr.write(&d, &rsp)?;
 
-                        if i == pairs_cnt {
+                        if i == (pairs_cnt - 1) {
                             wtr.finish(&d, &rsp)?;
                         }
 

readstat/src/lib.rs

@@ -160,16 +160,12 @@ impl ReadStatData {
 
         // setup parser
         // once call parse_sas7bdat, iteration begins
-        let mut parser = ReadStatParser::new()
+        let error = ReadStatParser::new()
             // do not set metadata handler nor variable handler as already processed
             .set_value_handler(Some(cb::handle_value))?
             .set_row_limit(Some(self.batch_rows_to_process.try_into().unwrap()))?
-            .set_row_offset(Some(self.batch_row_start.try_into().unwrap()))?;
-
-        let error = parser.parse_sas7bdat(ppath, ctx);
-
-        // drop parser after finished
-        // drop(parser);
+            .set_row_offset(Some(self.batch_row_start.try_into().unwrap()))?
+            .parse_sas7bdat(ppath, ctx);
 
         match FromPrimitive::from_i32(error as i32) {
             Some(ReadStatError::READSTAT_OK) => Ok(()),

readstat/src/rs_data.rs

@@ -401,17 +401,15 @@ impl ReadStatWriter {
             //if !self.wrote_start {
                 self.wtr = Some(ReadStatWriterFormat::Parquet(ArrowWriter::try_new(
                     f,
-                    //Arc::new(d.schema.clone()),
                     d.batch.schema(),
-                    // Arc::new(d.schema.clone()),
                     Some(WriterProperties::builder().set_write_batch_size(d.batch_rows_to_process).set_max_row_group_size(d.batch_rows_to_process).build()),
                 )?));
 //            };
 
             // write
             if let Some(rswf) = &mut self.wtr {
                 match rswf {
-                    ReadStatWriterFormat::Parquet(wtr) => { wtr.write(&d.batch)?; wtr.close()?; },
+                    ReadStatWriterFormat::Parquet(wtr) => { wtr.write(&d.batch)? },
                     _ => unreachable!()
                 }
             };