feat: add metrics tracking for file sinks and update orchestration fo…

[CadyLinn]

Which issue does this PR close?

• Closes Implement sink metrics but CSV/JSON #20644.

Rationale for this change

CSV and JSON sinks do not currently expose write metrics through DataSinkExec, while the Parquet sink exposes rows_written, bytes_written, and elapsed_compute.

This change provides consistent metrics for CSV and JSON sinks and centralizes the common file sink metrics wiring to reduce duplication across file formats.

What changes are included in this PR?

• Add a shared FileSinkMetrics implementation for rows_written, bytes_written, and elapsed_compute.
• Use the shared metrics implementation for Parquet, CSV, and JSON sinks.
• Track rows and serialized bytes in the common stateless write orchestration.
• Preserve the existing spawn_writer_tasks_and_join API and add an opt-in metrics variant.
• Add SQL logic test coverage for EXPLAIN ANALYZE COPY with CSV and JSON output.

Are these changes tested?

Yes. The affected datasource unit tests and the copy.slt SQL logic test pass.

cargo test -p datafusion-datasource \
  -p datafusion-datasource-csv \
  -p datafusion-datasource-json \
  -p datafusion-datasource-parquet --lib

cargo test -p datafusion-sqllogictest \
  --test sqllogictests -- copy.slt --complete

The affected crates also pass targeted Clippy checks with warnings denied.

Are there any user-facing changes?

Yes. EXPLAIN ANALYZE COPY now reports rows_written, bytes_written, and elapsed_compute for CSV and JSON sinks, consistent with the existing Parquet sink metrics.

There are no breaking public API changes.

[mkleen]

This is a little bit confusing. This reports the uncompressed size, but in Parquet it's the compressed row-group size. A gzipped CSV will report the uncompressed size. Do you think we could have the actual size written ?

https://github.com/apache/datafusion/blob/main/datafusion/datasource-parquet/src/sink.rs#L268

[mkleen]

This timer measures the entire spawn_writer_tasks_and_join_with_metrics function including object store io, channel waits, etc. The Parquet version counts only poll time. This will overstate the metric for slow object stores.

Can we maybe only measure the serialization work inside serialize_rb_stream_to_object_store e.g.:

let elapsed_compute = options.metrics.map(|m| m.elapsed_compute().clone());

pass that on over spawn_writer_tasks_and_join_with_metrics to serialize_rb_stream_to_object_store and then we could do something like:

 while let Some(batch) = data_rx.recv().await {
            let serializer_clone = Arc::clone(&serializer);
            let task = SpawnedTask::spawn(async move {
                let num_rows = batch.num_rows();
                let timer = elapsed_compute.as_ref().map(|t| t.timer()); // <- measure only the poll time 
                let bytes = serializer_clone.serialize(batch, initial)?;
                // stop measuring
                Ok((num_rows, bytes))
            });

then we would measure only poll time.

[mkleen]

Thanks for working on this. I left a few comments.