feat: support system.parquet_files table

[hiltontj]

This PR is intended to follow #25000.

Extend system table support to allow queries against system.parquet_files, a new system table that allows queries for parquet files associated with a given database and table.

Currently, the query will only look back on 1,000 segments to get associated parquet files, we may want to adjust that, or make it more flexible.

Two unit tests were added to query_executor for this, instead of E2E integration tests, as more control was needed over file persistence than what the E2E test harness provides.

Example

The system.parquet_files table can be queried like so:

SELECT * FROM system.parquet_files WHERE table_name = 'cpu'

And responds accordingly:

+------------+-------------------------------------------------+------------+-----------+---------------------+---------------------+
| table_name | path                                            | size_bytes | row_count | min_time            | max_time            |
+------------+-------------------------------------------------+------------+-----------+---------------------+---------------------+
| cpu        | dbs/foo/cpu/2024-05-14T20-53/4294967294.parquet | 1775       | 18        | 1715720025774406000 | 1715720035952490000 |
+------------+-------------------------------------------------+------------+-----------+---------------------+---------------------+

Other Changes

Another change in this PR was to go back to using dynamic dispatch for passing around the Persister. The associated types and generics made using the Persister a pain in the ParquetFilesTable implementation, so I removed those in favour of using Arc<dyn Persister>, as we had been doing previously (see 42d36d5).

Closes #24988

[pauldix]

The SegmentState in the write buffer actually has the information for the last 1,000 persisted segments and their parquet files: https://github.com/influxdata/influxdb/blob/main/influxdb3_write/src/write_buffer/segment_state.rs#L45. This is populated on startup.

It may be a good idea to pull the parquet files information from there, rather than the persister/object store. When the buffer gets the ability to persist some table data ahead of the segment rolling over, that state will be kept there in memory and won't show up in a segment info file until after the segment rolls over.

I think I'll be refactoring the structure of SegmentState when doing that. So maybe wait until after that to change this?

[hiltontj]

It may be a good idea to pull the parquet files information from [the SegmentState]

That makes sense and seems simpler. Giving the system table provider read access to the SegmentState shouldn't be too hard and reading the PersistedSegments off of that should be much cheaper than how I'm doing it here. In fact, with making this API public, I should be able to get them directly:

influxdb/influxdb3_write/src/write_buffer/segment_state.rs

Lines 216 to 232 in 2381cc6

	pub(crate) fn get_parquet_files(
	&self,
	database_name: &str,
	table_name: &str,
	) -> Vec<ParquetFile> {
	let mut parquet_files = vec![];
	for segment in self.persisted_segments.values() {
	segment.databases.get(database_name).map(|db| {
	db.tables.get(table_name).map(|table| {
	parquet_files.extend(table.parquet_files.clone());
	})
	});
	}
	parquet_files
	}

I can at least try that out and flag if there is an issue with doing so.

[hiltontj]

@pauldix - after taking a quick look at using the SegmentState directly, I found it gets a bit messy with the generics on SegmentState. Its (truncated) definition is:

// T -> TimeProvider
// W -> Wal
struct SegmentState<T, W> {
    /* ... */
}

The reason it gets messy is because I need to get access via the Bufferer trait to whatever I need from the write buffer. So returning a thing with generics from the Bufferer trait becomes a nuisance. Getting the persister was easy because I could just put the persister method on Bufferer:

trait Bufferer {
    /* ... */
    fn persister(&self) -> Arc<dyn Persister>;
}

If you are re-thinking the segment state, it may be helpful to have a trait, e.g.,

trait SegmentStateProvider {
    fn get_parquet_files(&self) -> Vec<ParquetFile>;

    /* Other methods you are considering for this API */
}

Then that would be more easily extracted via the Bufferer trait and passed to the SystemTableProvider as Arc<dyn SegmentStateProvider>.

I can leave this for now, since you're planning to refactor, don't want to make doing so more difficult with changes here.

[mgattozzi]

This LGTM code wise. I'll defer to @pauldix though if there's anything else we need to get into this PR, but from a code perspective it looks great.

[hiltontj]

Planning to wait on #25144 and any related refactoring to the influxdb3_write crate before revisiting this.

[hiltontj]

Closing in favour of #25225