Add SessionContext::read_batches

[Lordworms]

Which issue does this PR close?

Closes #9157

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

[Jefffrey]

Is there any way we can avoid an unwrap here?

[alamb]

Perhaps something like

Suggested change

	let schema: SchemaRef = batches.peek().unwrap().schema().clone();
	let schema: SchemaRef = if let Some(schema) = batches.peek() {
	schema.clone()
	} else {
	// use empty schema)
	Arc::new(Schema::new())
	};

[Lordworms]

Got it!

[alamb]

Thank you very much @Lordworms 🙏

As @Jefffrey pointed out, I think the only thing left for this PR is to handle the case with zero batches and add a test for that case too

[alamb]

Perhaps something like

Suggested change

	let schema: SchemaRef = batches.peek().unwrap().schema().clone();
	let schema: SchemaRef = if let Some(schema) = batches.peek() {
	schema.clone()
	} else {
	// use empty schema)
	Arc::new(Schema::new())
	};

[alamb]

Minor: I suggest making the test shorter by feeding in 2 batches rather than 5

[alamb]

Thank you @Lordworms -- this looks great.

cc @carols10cents

[alamb]

I think you could simply delete this statement entirely as the Schema is not needed

[Lordworms]

done

[alamb]

CI failure seems unrelated -- I retriggered it

[Jefffrey]

This looks good, just have a minor question 👍

[Jefffrey]

Hmm, is there any difference to mapping each batch into its own vec (which seemingly represents a different partition?) vs just collecting the batches into a single vec and passing in a single vec?

e.g. something like

Suggested change

	let provider =
	MemTable::try_new(schema, batches.map(|batch| vec![batch]).collect())?;
	let provider =
	MemTable::try_new(schema, vec![batches.collect()])?;

[alamb]

There is a difference (as you say the code in this PR makes its own partition). I think you are right that a single partition might be better (and DataFusion will repartition the plan into multiple partitions) if necessary

Is this something you can do @Lordworms ? Otherwise we can merge this PR as is and make it as a follow on change.

[Lordworms]

sure, so what we need to do is to flatten all the batches into a vec? I wonder in what situation the datafusion would do the repartition? Is there any doc for this?

[alamb]

TLDR is that it is added by https://docs.rs/datafusion/latest/datafusion/physical_optimizer/enforce_distribution/struct.EnforceDistribution.html -- the rules about when it happens are non trivial but basically are "when it helps"

[Lordworms]

Thanks, I would read about it and commit the changes soon.

[Jefffrey]

Thanks for this 👍