feat: Add segment persist of closed buffer segment

[pauldix]

This builds on #24624, which should get merged in first.

Fixes #24573. This adds persist to ClosedBufferSegment, which will persist the Catalog, if it has been modified, persist all tables and partitions in the buffer as parquet files and then write the segment file.

It doesn't have retries if errors are encountered during persistence and it doesn't parallelize them, which we should do later. Logged #24657 and #24658 to track those.

[hiltontj]

Just a couple of comments to add.

[hiltontj]

Along with this one, there are a few unwraps (and a panic) in this function. It looks like the function that calls this is fallible (called on 192):

influxdb/influxdb3_write/src/write_buffer/mod.rs

Lines 174 to 192 in 4d9095e

	fn get_table_chunks(
	&self,
	database_name: &str,
	table_name: &str,
	_filters: &[Expr],
	_projection: Option<&Vec<usize>>,
	_ctx: &SessionState,
	) -> Result<Vec<Arc<dyn QueryChunk>>, DataFusionError> {
	let db_schema = self.catalog.db_schema(database_name).unwrap();
	let table = db_schema.tables.get(table_name).unwrap();
	let schema = table.schema.as_ref().cloned().unwrap();
	let table_buffer = self.clone_table_buffer(database_name, table_name).unwrap();
	let mut chunks = Vec::with_capacity(table_buffer.partition_buffers.len());
	for (partition_key, partition_buffer) in table_buffer.partition_buffers {
	let partition_key: PartitionKey = partition_key.into();
	let batch = partition_buffer.rows_to_record_batch(&schema, table.columns());

So, rows_to_record_batch could also be fallible. Any opposition to using anyhow to simplify errors for functions like this?

[pauldix]

Yeah, that function was just a temporary placeholder to wire up the happy path. The underlying buffer implementation (and thus this function) are going to get rewritten after the basic end-to-end persistence is all wired up.

No objections to using anyhow from me. Want me to pick that up in this PR or in a follow on?

[hiltontj]

Might be better for a follow on, I think it is more or less switching from

rows_to_record_batch(/* ... */) -> RecordBatch {
    /* ... */
}

to

rows_to_record_batch(/* ... */) -> Result<RecordBatch, anyhow::Error> {
    /* ... */
}

and using anyhow::Context in place of unwrap/panic/expect here. However, this "poisons" the calling function with anyhow, which then needs to handle it explicitly or propagate it up, etc.

Once you've wired it up, the error handling could be addressed more holistically in a follow on.

[pauldix]

I fixed up the get_table_chunks to return a DataFusionError rather than unwrap. I'll address rows_to_record_batch later when I rework the buffer

[hiltontj]

LGTM - but noted a couple of minor things.