feat: Support serde for JsonSource PhysicalPlan

[westhide]

Which issue does this PR close?

• Closes Unsupported NdJsonExec plan and extension codec

Rationale for this change

Fix datafusion-ballista: Unsupported NdJsonExec plan and extension codec Exception

What changes are included in this PR?

Support serde for JsonSource PhysicalPlan

Are these changes tested?

Unit Test pass✅.

Ballista integration Test✅

   Compiling ballista v45.0.0 (/home/westhide/Code/apache/datafusion-ballista/ballista/client)
warning: `ballista-scheduler` (lib) generated 12 warnings
   Compiling ballista-examples v45.0.0 (/home/westhide/Code/apache/datafusion-ballista/examples)
    Finished `release` profile [optimized] target(s) in 26.98s
     Running `/home/westhide/Code/apache/datafusion-ballista/target/release/examples/remote-sql`
+-----+------------------+---------------+------+
| a   | b                | c             | d    |
+-----+------------------+---------------+------+
| 1   | [2.0, 1.3, -6.1] | [false, true] | 4    |
| -10 | [2.0, 1.3, -6.1] | [true, true]  | 4    |
| 2   | [2.0, , -6.1]    | [false, ]     | text |
|     |                  |               |      |
+-----+------------------+---------------+------+

Are there any user-facing changes?

No

[westhide]

take

[alamb]

Thank you @westhide

Once CI passes I think this PR looks good to me, but I do think we should consider serializing the other field too

[alamb]

It seems like there is one more relevant field for JsonSource, batch_size

datafusion/datafusion/datasource-json/src/source.rs

Lines 251 to 257 in 6d5e00a

	/// JsonSource holds the extra configuration that is necessary for [`JsonOpener`]
	#[derive(Clone, Default)]
	pub struct JsonSource {
	batch_size: Option<usize>,
	metrics: ExecutionPlanMetricsSet,
	projected_statistics: Option<Statistics>,
	}

Perhaps we can add that field to the serialization as well (or file a ticket to add it?)

[westhide]

Currently batch_size will set by context.session_config().batch_size() when DataSource call open(..), serialization batch_size will let the JsonSource use client config instead of the Executor session_config in Ballista. Should we change this behavior for all file format(csv,parquert,avro...)?

datafusion/datafusion/datasource/src/file_scan_config.rs

Line 182 in 19dd46d

.with_batch_size(context.session_config().batch_size())

[milenkovicm]

Client config should be propagated from to executor (task context) so it should be the same.

[westhide]

Got it, I read the code about TaskDefinition in ballista. Thx~

[alamb]

👍

[milenkovicm]

Thanks @westhide, PR makes sense to me.

I left one comment in the original ballista issue, apache/datafusion-ballista#1209 (comment)

If you have time please add test in ballista unsupported features will help us to track this issue until we get to datafusion 47, you can the same one you used in the issue. Otherwise we can add it as a follow up

[milenkovicm]

I agree with @alamb, adding batch_size should be straight forward

[alamb]

Thanks @westhide

For anyone following along:

• feat: Support serde for FileScanConfig batch_size #15335