Proto: migrate file sink serialization

[Phoenix500526]

This is a stacked PR based on #23752. Only the commits on top of #23752 are part of this review.

Which issue does this PR close?

• Closes Proto: migrate the file sinks (CsvSink, JsonSink, ParquetSink) #23519.

Rationale for this change

#23752 adds the DataSink::try_to_proto hook and moves the shared
FileSinkConfig protobuf conversion into datafusion-datasource.

This PR uses that foundation to move CSV, JSON, and Parquet sink
serialization out of the central datafusion-proto downcast chain.
Each concrete sink now owns its format-specific protobuf encoding and
decoding logic.

What changes are included in this PR?

• Implement DataSink::try_to_proto for CsvSink, JsonSink, and
  ParquetSink.
• Add inherent try_from_proto methods to reconstruct each sink's
  DataSinkExec.
• Repoint the physical-plan decode arms to the sink-owned decoders.
• Add feature-gated protobuf dependencies to the three format crates.
• Remove the active central DataSinkExec serialization dispatch after
  migrating its final built-in sink.
• Retain the old serialization helpers as deprecated compatibility
  delegates.
• Let sinks without a built-in protobuf representation fall through to
  the physical extension codec.
• Preserve the existing protobuf wire representation.

The migrations are split into one commit per sink.

Are these changes tested?

Yes. Existing sink round-trip tests cover the protobuf representation.

The following checks passed:

• Focused CSV, JSON, and Parquet sink round-trip tests.
• All datafusion-proto integration tests.
• cargo check -p datafusion-proto --no-default-features.
• cargo fmt --all.
• cargo clippy --all-targets --all-features -- -D warnings.
• The required extended workspace test suite, including all 495 SQL
  logic test files.

Are there any user-facing changes?

No functional or wire-format changes are intended.

The old compatibility helpers remain available but are deprecated.

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion-datasource v54.1.0 (current)
       Built [  49.375s] (current)
     Parsing datafusion-datasource v54.1.0 (current)
      Parsed [   0.035s] (current)
    Building datafusion-datasource v54.1.0 (baseline)
       Built [  42.667s] (baseline)
     Parsing datafusion-datasource v54.1.0 (baseline)
      Parsed [   0.034s] (baseline)
    Checking datafusion-datasource v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.354s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [  93.861s] datafusion-datasource
    Building datafusion-datasource-csv v54.1.0 (current)
       Built [  42.802s] (current)
     Parsing datafusion-datasource-csv v54.1.0 (current)
      Parsed [   0.013s] (current)
    Building datafusion-datasource-csv v54.1.0 (baseline)
       Built [  39.181s] (baseline)
     Parsing datafusion-datasource-csv v54.1.0 (baseline)
      Parsed [   0.013s] (baseline)
    Checking datafusion-datasource-csv v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.134s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [  83.113s] datafusion-datasource-csv
    Building datafusion-datasource-json v54.1.0 (current)
       Built [  42.609s] (current)
     Parsing datafusion-datasource-json v54.1.0 (current)
      Parsed [   0.014s] (current)
    Building datafusion-datasource-json v54.1.0 (baseline)
       Built [  39.795s] (baseline)
     Parsing datafusion-datasource-json v54.1.0 (baseline)
      Parsed [   0.014s] (baseline)
    Checking datafusion-datasource-json v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.120s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [  83.574s] datafusion-datasource-json
    Building datafusion-datasource-parquet v54.1.0 (current)
       Built [  49.483s] (current)
     Parsing datafusion-datasource-parquet v54.1.0 (current)
      Parsed [   0.034s] (current)
    Building datafusion-datasource-parquet v54.1.0 (baseline)
       Built [  44.679s] (baseline)
     Parsing datafusion-datasource-parquet v54.1.0 (baseline)
      Parsed [   0.034s] (baseline)
    Checking datafusion-datasource-parquet v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.221s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [  95.603s] datafusion-datasource-parquet
    Building datafusion-proto v54.1.0 (current)
       Built [  59.666s] (current)
     Parsing datafusion-proto v54.1.0 (current)
      Parsed [   0.020s] (current)
    Building datafusion-proto v54.1.0 (baseline)
       Built [  59.965s] (baseline)
     Parsing datafusion-proto v54.1.0 (baseline)
      Parsed [   0.021s] (baseline)
    Checking datafusion-proto v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.370s] 223 checks: 222 pass, 1 fail, 0 warn, 30 skip

--- failure trait_method_marked_deprecated: trait method #[deprecated] added ---

Description:
A trait method is now #[deprecated]. Downstream crates will get a compiler warning when using this method.
        ref: https://doc.rust-lang.org/reference/attributes/diagnostics.html#the-deprecated-attribute
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.49.0/src/lints/trait_method_marked_deprecated.ron

Failed in:
  method try_into_json_sink_physical_plan in trait datafusion_proto::physical_plan::PhysicalPlanNodeExt in /home/runner/work/datafusion/datafusion/datafusion/proto/src/physical_plan/mod.rs:662
  method try_into_csv_sink_physical_plan in trait datafusion_proto::physical_plan::PhysicalPlanNodeExt in /home/runner/work/datafusion/datafusion/datafusion/proto/src/physical_plan/mod.rs:662
  method try_into_parquet_sink_physical_plan in trait datafusion_proto::physical_plan::PhysicalPlanNodeExt in /home/runner/work/datafusion/datafusion/datafusion/proto/src/physical_plan/mod.rs:662
  method try_from_data_sink_exec in trait datafusion_proto::physical_plan::PhysicalPlanNodeExt in /home/runner/work/datafusion/datafusion/datafusion/proto/src/physical_plan/mod.rs:662

     Summary semver requires new minor version: 0 major and 1 minor checks failed
    Finished [ 121.166s] datafusion-proto

[codecov-commenter]

Codecov Report

❌ Patch coverage is 59.42623% with 99 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.02%. Comparing base (24483db) to head (d2f7924).

Files with missing lines	Patch %	Lines
datafusion/proto/src/physical_plan/mod.rs	9.09%	30 Missing ⚠️
datafusion/datasource-csv/src/file_format.rs	68.33%	8 Missing and 11 partials ⚠️
datafusion/datasource-json/src/file_format.rs	68.33%	8 Missing and 11 partials ⚠️
datafusion/datasource-parquet/src/sink.rs	68.33%	8 Missing and 11 partials ⚠️
datafusion/datasource/src/sink.rs	76.00%	3 Missing and 3 partials ⚠️
datafusion/proto/src/physical_plan/from_proto.rs	0.00%	3 Missing ⚠️
datafusion/proto/src/physical_plan/to_proto.rs	0.00%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #23781      +/-   ##
==========================================
- Coverage   81.05%   81.02%   -0.04%     
==========================================
  Files        1106     1106              
  Lines      380556   380652      +96     
  Branches   380556   380652      +96     
==========================================
- Hits       308477   308425      -52     
- Misses      53861    54002     +141     
- Partials    18218    18225       +7     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[adriangb]

Thanks @Phoenix500526 — reviewing the commits on top of #23752 as you asked. The migration itself is faithful: each sink's encode matches the arm it replaces field-for-field (including sink_schema and decoding sort_order against the input schema), the central DataSinkExec downcast block is removed, the decode arms are repointed, and the old helpers are kept as deprecated delegates. That is the shape #23494 asks for.

Four things before this comes out of draft.

Rebase. Main has moved: #24006 merged and this now conflicts in datafusion/datasource/Cargo.toml, datafusion/proto/src/physical_plan/mod.rs, from_proto.rs and roundtrip_physical_plan.rs. Most of the resolution is deletion — datafusion-datasource already carries a proto feature, and PartitionedFile / FileGroup <-> proto already live in datafusion_datasource::proto.

Use expect_plan_variant! instead of hand-rolling the variant match. Each try_from_proto currently opens with

let sink_node = match &node.physical_plan_type {
    Some(protobuf::physical_plan_node::PhysicalPlanType::CsvSink(sink)) => sink.as_ref(),
    _ => return datafusion_common::internal_err!("PhysicalPlanNode is not a CsvSink"),
};

which is what the macro exists for. It is #[macro_export]ed from datafusion-physical-plan, so the format crates can use it as datafusion_physical_plan::expect_plan_variant!, and every plan migrated under this epic on main already does. It also produces the same error string, so nothing changes behaviorally.

The TryFromProto impls for the sink messages are now duplicated logic. impl TryFromProto<&CsvSink> for protobuf::CsvSink (and the Json/Parquet siblings, both directions) still carry their original bodies in datafusion-proto, while this PR adds a second copy of the same field mapping inside CsvSink::try_to_proto. That is the drift risk the epic is trying to remove, and it is the one thing here that is not yet single-sourced.

Since these types live in the format crates, the conversions can move next to them as standard TryFrom impls — see #24006 for the pattern and #24019 for why. Concretely, give each sink an inherent conversion for its message:

impl TryFrom<&CsvSink> for protobuf::CsvSink { ... }
impl TryFrom<&protobuf::CsvSink> for CsvSink { ... }

then have CsvSink::try_to_proto / try_from_proto use them for the sink field, and reduce datafusion-proto's TryFromProto impls to delegates. As a bonus this restores the TryFrom<&CsvSink> / TryFrom<&protobuf::CsvSink> impls that shipped in 54.1.0 and were dropped on main (#24019), so it is a regression fix rather than new API.

Consider merging this into #23752. The two are one logical change — #23752 adds a hook with no built-in implementor, and until this PR lands every built-in sink encodes its input subtree twice (the hook encodes it, the sink returns Ok(None), and the central chain encodes it again). Merged together they are also easier to review as a single before/after, and it removes the need for parse_sink_sort_order to be public API in #23752 with no caller. Entirely your call — the stack is reviewable as-is if you would rather keep it split.

One small thing: the non-parquet arm changed its panic message from "Trying to use ParquetSink without parquet feature enabled" to "Unable to process a Parquet PhysicalPlan when parquet feature is not enabled". Harmless, but if it was not deliberate it is worth keeping the original so the string stays greppable.

[Phoenix500526]

Thanks @Phoenix500526 — reviewing the commits on top of #23752 as you asked. The migration itself is faithful: each sink's encode matches the arm it replaces field-for-field (including sink_schema and decoding sort_order against the input schema), the central DataSinkExec downcast block is removed, the decode arms are repointed, and the old helpers are kept as deprecated delegates. That is the shape #23494 asks for.

Four things before this comes out of draft.

Rebase. Main has moved: #24006 merged and this now conflicts in datafusion/datasource/Cargo.toml, datafusion/proto/src/physical_plan/mod.rs, from_proto.rs and roundtrip_physical_plan.rs. Most of the resolution is deletion — datafusion-datasource already carries a proto feature, and PartitionedFile / FileGroup <-> proto already live in datafusion_datasource::proto.

Use expect_plan_variant! instead of hand-rolling the variant match. Each try_from_proto currently opens with

let sink_node = match &node.physical_plan_type {
    Some(protobuf::physical_plan_node::PhysicalPlanType::CsvSink(sink)) => sink.as_ref(),
    _ => return datafusion_common::internal_err!("PhysicalPlanNode is not a CsvSink"),
};

which is what the macro exists for. It is #[macro_export]ed from datafusion-physical-plan, so the format crates can use it as datafusion_physical_plan::expect_plan_variant!, and every plan migrated under this epic on main already does. It also produces the same error string, so nothing changes behaviorally.

The TryFromProto impls for the sink messages are now duplicated logic. impl TryFromProto<&CsvSink> for protobuf::CsvSink (and the Json/Parquet siblings, both directions) still carry their original bodies in datafusion-proto, while this PR adds a second copy of the same field mapping inside CsvSink::try_to_proto. That is the drift risk the epic is trying to remove, and it is the one thing here that is not yet single-sourced.

Since these types live in the format crates, the conversions can move next to them as standard TryFrom impls — see #24006 for the pattern and #24019 for why. Concretely, give each sink an inherent conversion for its message:

impl TryFrom<&CsvSink> for protobuf::CsvSink { ... }
impl TryFrom<&protobuf::CsvSink> for CsvSink { ... }

then have CsvSink::try_to_proto / try_from_proto use them for the sink field, and reduce datafusion-proto's TryFromProto impls to delegates. As a bonus this restores the TryFrom<&CsvSink> / TryFrom<&protobuf::CsvSink> impls that shipped in 54.1.0 and were dropped on main (#24019), so it is a regression fix rather than new API.

Consider merging this into #23752. The two are one logical change — #23752 adds a hook with no built-in implementor, and until this PR lands every built-in sink encodes its input subtree twice (the hook encodes it, the sink returns Ok(None), and the central chain encodes it again). Merged together they are also easier to review as a single before/after, and it removes the need for parse_sink_sort_order to be public API in #23752 with no caller. Entirely your call — the stack is reviewable as-is if you would rather keep it split.

One small thing: the non-parquet arm changed its panic message from "Trying to use ParquetSink without parquet feature enabled" to "Unable to process a Parquet PhysicalPlan when parquet feature is not enabled". Harmless, but if it was not deliberate it is worth keeping the original so the string stays greppable.

Done

[Copilot]

Pull request overview

This PR continues the proto-migration work from #23752 by moving CSV/JSON/Parquet file sink protobuf serialization/deserialization out of datafusion-proto and into the concrete sink crates via the DataSink::try_to_proto hook, while keeping wire compatibility and leaving deprecated compatibility delegates in place.

Changes:

• Implement DataSink::try_to_proto + sink-owned try_from_proto reconstruction for CsvSink, JsonSink, and ParquetSink.
• Repoint datafusion-proto physical plan decoding to call the sink-owned decoders and remove the central DataSinkExec downcast-based serialization dispatch.
• Add feature-gated proto dependencies/features to the format crates to host their protobuf logic.

Reviewed changes

Copilot reviewed 11 out of 12 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
datafusion/proto/src/physical_plan/to_proto.rs	Switch legacy TryFromProto sink conversions to delegate to sink-owned TryFrom impls.
datafusion/proto/src/physical_plan/mod.rs	Route decode arms to sink-owned try_from_proto; remove central DataSinkExec serialization special-case; keep deprecated compatibility helpers.
datafusion/proto/src/physical_plan/from_proto.rs	Switch legacy TryFromProto sink conversions to delegate to sink-owned TryFrom impls.
datafusion/proto/Cargo.toml	Enable proto feature on csv/json/parquet datasource crates to access sink-owned protobuf implementations.
datafusion/datasource/src/sink.rs	Add DataSinkExec::decode_sort_order helper for sink-owned protobuf decoding.
datafusion/datasource-parquet/src/sink.rs	Add Parquet sink-owned protobuf encode/decode (try_to_proto, TryFrom, try_from_proto).
datafusion/datasource-parquet/Cargo.toml	Add proto feature + optional datafusion-proto-models dependency wiring.
datafusion/datasource-json/src/file_format.rs	Add JSON sink-owned protobuf encode/decode (try_to_proto, TryFrom, try_from_proto).
datafusion/datasource-json/Cargo.toml	Add proto feature + optional datafusion-proto-models dependency wiring.
datafusion/datasource-csv/src/file_format.rs	Add CSV sink-owned protobuf encode/decode (try_to_proto, TryFrom, try_from_proto).
datafusion/datasource-csv/Cargo.toml	Add proto feature + optional datafusion-proto-models dependency wiring.
Cargo.lock	Record new dependency edges from enabling proto-models in the sink crates.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Phoenix500526]

Done

[Phoenix500526]

Done