WIP: Address upgrade DF comments, use fewer clones

Author: srh

datafusion/common/src/file_options/parquet_writer.rs

@@ -70,9 +70,6 @@ impl WriterPropertiesConfig {
     pub fn noop() -> Arc<dyn WriterPropertiesCustomizer> {
         Arc::new(NoopWriterPropertiesCustomizer{})
     }
-    pub fn todo() -> Arc<dyn WriterPropertiesCustomizer> {  // TODO upgrade DF: no
-        Self::noop()
-    }
 }
 
 pub trait WriterPropertiesCustomizer: Sync + Send + std::fmt::Debug {

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

@@ -434,7 +434,7 @@ impl FileFormat for ParquetFormat {
         }
 
         let sink_schema = conf.output_schema().clone();
-        let sink = Arc::new(ParquetSink::new(conf, self.options.clone(), &self.customizer));
+        let sink = Arc::new(ParquetSink::new(conf, self.options.clone(), self.customizer.clone()));
 
         Ok(Arc::new(DataSinkExec::new(
             input,
@@ -703,8 +703,7 @@ impl DisplayAs for ParquetSink {
 
 impl ParquetSink {
     /// Create from config.
-    pub fn new(config: FileSinkConfig, parquet_options: TableParquetOptions, customizer: &Arc<dyn WriterPropertiesCustomizer>) -> Self {
-        let customizer = customizer.clone();
+    pub fn new(config: FileSinkConfig, parquet_options: TableParquetOptions, customizer: Arc<dyn WriterPropertiesCustomizer>) -> Self {
         Self {
             config,
             parquet_options,
@@ -2300,7 +2299,7 @@ mod tests {
                 ]),
                 ..Default::default()
             },
-            &customizer,
+            customizer,
         ));
 
         // create data
@@ -2391,7 +2390,7 @@ mod tests {
         let parquet_sink = Arc::new(ParquetSink::new(
             file_sink_config,
             TableParquetOptions::default(),
-            &customizer,
+            customizer,
         ));
 
         // create data with 2 partitions
@@ -2483,7 +2482,7 @@ mod tests {
                     global,
                     ..Default::default()
                 },
-                &customizer,
+                customizer,
             ));
 
             // create data

datafusion/proto/src/physical_plan/from_proto.rs

@@ -596,18 +596,18 @@ impl TryFrom<&protobuf::JsonSink> for JsonSink {
     }
 }
 
-// TODO upgrade DF: Who uses this?
 #[cfg(feature = "parquet")]
 impl TryFrom<&protobuf::ParquetSink> for ParquetSink {
     type Error = DataFusionError;
 
     fn try_from(value: &protobuf::ParquetSink) -> Result<Self, Self::Error> {
-        let customizer: Arc<dyn WriterPropertiesCustomizer> = WriterPropertiesConfig::todo();
+        // Cube: This function isn't reached, so, we can have a noop hard-coded here.
+        let customizer: Arc<dyn WriterPropertiesCustomizer> = WriterPropertiesConfig::noop();
 
         Ok(Self::new(
             convert_required!(value.config)?,
             convert_required!(value.parquet_options)?,
-            &customizer,
+            customizer,
         ))
     }
 }

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -1244,7 +1244,7 @@ fn roundtrip_parquet_sink() -> Result<()> {
     let data_sink = Arc::new(ParquetSink::new(
         file_sink_config,
         TableParquetOptions::default(),
-        &customizer,
+        customizer,
     ));
     let sort_order = vec![PhysicalSortRequirement::new(
         Arc::new(Column::new("plan_type", 0)),