feat: metadata cache core impl

[hiltontj]

Part of #25543

Implement the base MetaCache type that holds the hierarchical structure of the metadata cache providing methods to:

• create the cache (here),
• push rows from the WAL into the cache (here),
• get record batches out of the cache with a given set of predicates (here), and
• prune the cache based on a max_age and a max_cardinality (here)

Tests

• Verify predicate evaluation when fetching record batches from the cache (here)
• Verify pruning of the cache (here)

Notes

• This PR adds a influxdb3_cache crate, which just holds the metadata cache for now, but may well also hold the last-n-value cache and parquet cache to get those out of the influxdb3_write crate
• Some accessory changes were made, namely, to expose the WriteValidator from influxdb3_write so it can be used as a more light-weight way of parsing line protocol and producing Rows for tests
• To close the related issue, I still need to add a MetaCacheProvider and other pieces necessary to tie this into the system.

[hiltontj]

@alamb I tagged you on this as it is using StringView and I thought that may interest you. Starting here:

influxdb/influxdb3_cache/src/meta_cache/mod.rs

Lines 183 to 220 in dff0b2e

	/// Gather a record batch from a cache given the set of predicates
	///
	/// This assumes the predicates are well behaved, and validated before being passed in. For example,
	/// there cannot be multiple predicates on a single column; the caller needs to validate/transform
	/// the incoming predicates from Datafusion before calling.
	///
	/// Entries in the cache that have not been seen since before the `max_age` of the cache will be
	/// filtered out of the result.
	pub fn to_record_batch(&self, predicates: &[Predicate]) -> Result<RecordBatch, ArrowError> {
	// predicates may not be provided for all columns in the cache, or not be provided in the
	// order of columns in the cache. This re-orders them to the correct order, and fills in any
	// gaps with None.
	let predicates: Vec<Option<&Predicate>> = self
	.column_ids
	.iter()
	.map(|id| predicates.iter().find(|p| &p.column_id == id))
	.collect();
	// Uses a [`StringViewBuilder`] to compose the set of [`RecordBatch`]es. This is convenient for
	// the sake of nested caches, where a predicate on a higher branch in the cache will need to have
	// its value in the outputted batches duplicated.
	let mut builders: Vec<StringViewBuilder> = (0..self.column_ids.len())
	.map(|_| StringViewBuilder::new())
	.collect();
	let expired_time_ns = self.expired_time_ns();
	let _ = self
	.data
	.evaluate_predicates(expired_time_ns, &predicates, &mut builders);
	RecordBatch::try_new(
	Arc::clone(&self.schema),
	builders
	.into_iter()
	.map(|mut builder| Arc::new(builder.finish()) as ArrayRef)
	.collect(),
	)
	}

[mgattozzi]

Looks good to me if this has follow up PRs (which I'm assuming since this doesn't seem to be hooked in)

[mgattozzi]

This is for a follow up PR when the metadata cache actually gets hooked up right?

[hiltontj]

Yeah, the type will get used when I implement the DataFusion trait to hook this all up to the query path in #25544