Add tracing spans to last value cache

[hiltontj]

Problem statement

There has been variable latency observed when using the last cache under higher concurrent query loads. There are no tracing spans to indicate where in the scan time is being taken, and contributing to that latency, in the last cache.

Proposed solution

Extract the span context from the DataFusion Session that is provided to the TableProvider::scan implementation for the last_cache:

influxdb/influxdb3_cache/src/last_cache/table_function.rs

Lines 71 to 77 in d8efdb4

	async fn scan(
	&self,
	ctx: &dyn Session,
	projection: Option<&Vec<usize>>,
	filters: &[Expr],
	_limit: Option<usize>,
	) -> Result<Arc<dyn ExecutionPlan>, DataFusionError> {

It would be worth wrapping a span around the following to generate traces:

• Filter predicate conversion:

  influxdb/influxdb3_cache/src/last_cache/table_function.rs

  Lines 84 to 89 in d8efdb4

  	let predicates = convert_filter_exprs(
  	self.table_def.as_ref(),
  	cache.key_column_ids.as_ref(),
  	Arc::clone(&self.schema),
  	filters,
  	)?;

• Record batch conversion:

  influxdb/influxdb3_cache/src/last_cache/table_function.rs

  Line 90 in d8efdb4

  let batches = cache.to_record_batches(Arc::clone(&self.table_def), &predicates)?;

Within the latter, we could break down into more spans:

• Predicate evaluation of the cache key hierarchy:

  influxdb/influxdb3_cache/src/last_cache/cache.rs

  Lines 427 to 458 in d8efdb4

  	for predicate in predicates {
  	if caches.is_empty() {
  	return Ok(vec![]);
  	}
  	let mut new_caches = vec![];
  	for c in caches {
  	let Some(cache_key) = c.state.as_key() else {
  	continue;
  	};
  	if let Some(pred) = predicate {
  	let next_states = cache_key.evaluate_predicate(pred);
  	new_caches.extend(next_states.into_iter().map(|(state, value)| {
  	let mut additional_columns = c.key_column_values.clone();
  	additional_columns.push(value);
  	ExtendedLastCacheState {
  	state,
  	key_column_values: additional_columns,
  	}
  	}));
  	} else {
  	new_caches.extend(cache_key.value_map.iter().map(|(v, state)| {
  	let mut additional_columns = c.key_column_values.clone();
  	additional_columns.push(v);
  	ExtendedLastCacheState {
  	state,
  	key_column_values: additional_columns,
  	}
  	}));
  	}
  	}
  	caches = new_caches;
  	}

• Scanning the filtered leaf nodes:

  influxdb/influxdb3_cache/src/last_cache/cache.rs

  Lines 460 to 463 in d8efdb4

  	caches
  	.into_iter()
  	.map(|c| c.to_record_batch(Arc::clone(&table_def), Arc::clone(&self.schema)))
  	.collect()

Additional context

This is related to #25562