WIP: Add WriterPropertiesCustomizer::allow_single_file_parallelism

Author: srh

datafusion/common/src/file_options/parquet_writer.rs

@@ -77,6 +77,9 @@ impl WriterPropertiesConfig {
 
 pub trait WriterPropertiesCustomizer: Sync + Send + std::fmt::Debug {
     fn adjust_write_properties(&self, builder: WriterPropertiesBuilder) -> WriterPropertiesBuilder;
+    fn allow_single_file_parallelism(&self) -> bool {
+        true
+    }
 }
 
 #[derive(Debug)]

datafusion/core/src/datasource/file_format/parquet.rs

@@ -477,7 +477,6 @@ pub async fn fetch_parquet_metadata(
     let mut footer = [0; 8];
     footer.copy_from_slice(&suffix[suffix_len - 8..suffix_len]);
 
-    // TODO upgrade DF: Can't use decode_footer directly.
     #[allow(deprecated)]
     let length = decode_footer(&footer)?;
 
@@ -798,7 +797,7 @@ impl DataSink for ParquetSink {
 
         let parquet_opts = &self.parquet_options;
         let allow_single_file_parallelism =
-            parquet_opts.global.allow_single_file_parallelism;
+            parquet_opts.global.allow_single_file_parallelism && self.customizer.allow_single_file_parallelism();
 
         let part_col = if !self.config.table_partition_cols.is_empty() {
             Some(self.config.table_partition_cols.clone())
@@ -937,8 +936,8 @@ fn spawn_column_parallel_row_group_writer(
     pool: &Arc<dyn MemoryPool>,
 ) -> Result<(Vec<ColumnWriterTask>, Vec<ColSender>)> {
     let schema_desc = arrow_to_parquet_schema(&schema)?;
-    let row_group_ordinal = 0;  // TODO upgrade DF: encrypt
-    let col_writers = get_column_writers(&schema_desc, &parquet_props, &schema, row_group_ordinal)?;
+    let row_group_ordinal_unused = 0;
+    let col_writers = get_column_writers(&schema_desc, &parquet_props, &schema, row_group_ordinal_unused)?;
     let num_columns = col_writers.len();
 
     let mut col_writer_tasks = Vec::with_capacity(num_columns);