/// Write the given DataFrame in the the writer `W`.
Write the given DataFrame in the the writer `W`.
Write the given DataFrame in the the writer ` W ` .
pub fn finish(mut self, df: &DataFrame) -> Result<()> {
        let df = to_arrow_compatible_df(df);
        let mut fields = df.schema().to_arrow().fields().clone();
        // date64 is not supported by parquet and will be be truncated to date32
        // We coerce these to timestamp(ms)
        let date64_columns = df
            .get_columns()
            .iter()
            .enumerate()
            .filter_map(|(i, s)| match s.dtype() {
                DataType::Date64 => {
                    fields[i] = ArrowField::new(
                        s.name(),
                        ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
                        s.null_count() > 0,
                    );
                    Some(i)
                }
                _ => None,
            })
            .collect::<Vec<_>>();
        let column_names = df
            .get_columns()
            .iter()
            .map(|s| s.name().to_string())
            .collect::<Vec<_>>();
        let iter = df.iter_record_batches().map(|rb| {
            if !date64_columns.is_empty() {
                let mut columns = rb.columns().to_vec();
                for i in &date64_columns {
                    let array = cast::cast(columns[*i].as_ref(), &ArrowDataType::Int64).unwrap();
                    let array = cast::cast(
                        array.as_ref(),
                        &ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
                    )
                    .unwrap()
                    .into();
                    columns[*i] = array;
                }
                RecordBatch::try_from_iter(column_names.iter().zip(columns)).unwrap()
            } else {
                rb
            }
        });
        let options = write::WriteOptions {
            write_statistics: false,
            compression: self.compression,
            version: write::Version::V2,
        };
        let schema = ArrowSchema::new(fields);
        let parquet_schema = write::to_parquet_schema(&schema)?;
        let parquet_schema_iter = parquet_schema.clone();
        let iter = iter.map(|batch| {
            let columns = batch.columns().to_vec();
            Ok(write::DynIter::new(
                columns
                    .into_iter()
                    .zip(parquet_schema_iter.columns().to_vec().into_iter())
                    .map(|(array, type_)| {
                        // one parquet page per array.
                        // we could use `array.slice()` to split it based on some number of rows.
                        Ok(write::DynIter::new(std::iter::once(write::array_to_page(
                            array.as_ref(),
                            type_,
                            options,
                            Encoding::Plain,
                        ))))
                    }),
            ))
        });
        write::write_file(
            &mut self.writer,
            iter,
            &schema,
            parquet_schema,
            options,
            None,
        )?;
        Ok(())
    }
pub fn finish ( mut self , df : & DataFrame ) -> Result < ( ) > { let df = to_arrow_compatible_df ( df ) ; let mut fields = df . schema ( ) . to_arrow ( ) . fields ( ) . clone ( ) ; // date64 is not supported by parquet and will be be truncated to date32 // We coerce these to timestamp(ms) let date64_columns = df . get_columns ( ) . iter ( ) . enumerate ( ) . filter_map ( | ( i , s ) | match s . dtype ( ) { DataType :: Date64 => { fields [ i ] = ArrowField :: new ( s . name ( ) , ArrowDataType :: Timestamp ( TimeUnit :: Millisecond , None ) , s . null_count ( ) > 0 , ) ; Some ( i ) } _ => None , } ) . collect :: < Vec < _ > > ( ) ; let column_names = df . get_columns ( ) . iter ( ) . map ( | s | s . name ( ) . to_string ( ) ) . collect :: < Vec < _ > > ( ) ; let iter = df . iter_record_batches ( ) . map ( | rb | { if ! date64_columns . is_empty ( ) { let mut columns = rb . columns ( ) . to_vec ( ) ; for i in & date64_columns { let array = cast :: cast ( columns [ * i ] . as_ref ( ) , & ArrowDataType :: Int64 ) . unwrap ( ) ; let array = cast :: cast ( array . as_ref ( ) , & ArrowDataType :: Timestamp ( TimeUnit :: Millisecond , None ) , ) . unwrap ( ) . into ( ) ; columns [ * i ] = array ; } RecordBatch :: try_from_iter ( column_names . iter ( ) . zip ( columns ) ) . unwrap ( ) } else { rb } } ) ; let options = write :: WriteOptions { write_statistics : false , compression : self . compression , version : write :: Version :: V2 , } ; let schema = ArrowSchema :: new ( fields ) ; let parquet_schema = write :: to_parquet_schema ( & schema ) ? ; let parquet_schema_iter = parquet_schema . clone ( ) ; let iter = iter . map ( | batch | { let columns = batch . columns ( ) . to_vec ( ) ; Ok ( write :: DynIter :: new ( columns . into_iter ( ) . zip ( parquet_schema_iter . columns ( ) . to_vec ( ) . into_iter ( ) ) . map ( | ( array , type_ ) | { // one parquet page per array. // we could use `array.slice()` to split it based on some number of rows. Ok ( write :: DynIter :: new ( std :: iter :: once ( write :: array_to_page ( array . as_ref ( ) , type_ , options , Encoding :: Plain , ) ) ) ) } ) , ) ) } ) ; write :: write_file ( & mut self . writer , iter , & schema , parquet_schema , options , None , ) ? ; Ok ( ( ) ) }
Write the given DataFrame in the the writer `W`.
Write the given DataFrame in the the writer ` W ` .
mut self
df
&DataFrame