Repository: influxdata/influxdb
File: influxdb3_cache/src/distinct_cache/table_function.rs (Rust)

Code before:
        let (batches, predicates) = if let Some(cache) = read
            .get(&self.db_id)
            .and_then(|db| db.get(&self.table_def.table_id))
            .and_then(|tbl| tbl.get(&self.cache_name))
        {
            let predicates = convert_filter_exprs(&self.table_def, self.schema(), filters)?;
            (
                cache
                    .to_record_batch(
                        Arc::clone(&schema),
                        &predicates,
                        projection.map(|p| p.as_slice()),
                        limit,
                    )
                    .map(|batch| vec![batch])?,
                (!predicates.is_empty()).then_some(predicates),
            )
        } else {
            (vec![], None)
        };

        let mut distinct_exec = DistinctCacheExec::try_new(
            predicates,
            Arc::clone(&self.table_def),
            &[batches],
            self.schema(),
            projection,
            limit,
        )?;

        let show_sizes = ctx.config_options().explain.show_sizes;
        distinct_exec = distinct_exec.with_show_sizes(show_sizes);

        Ok(Arc::new(distinct_exec))
    }
}

/// Convert the given list of filter expressions to a map of [`ColumnId`] to [`Predicate`]
///
/// The resulting map uses [`IndexMap`] to ensure consistent ordering of the map. This makes testing
/// the filter conversion significantly easier using EXPLAIN queries.
fn convert_filter_exprs(
    table_def: &TableDefinition,
    cache_schema: SchemaRef,
    filters: &[Expr],
) -> Result<IndexMap<ColumnId, Predicate>> {
    let mut predicate_map: IndexMap<ColumnId, Option<Predicate>> = IndexMap::new();

    // for create_physical_expr:
    let schema: DFSchema = cache_schema.try_into()?;
    let props = ExecutionProps::new();

Review comment (performance):
@praveen-influx this was the change that fixed the problem. `self.schema()` is the unprojected schema of the cache. `schema` is the projection of the same. Because the unprojected schema was getting passed in here, the internals of `arrow-rs` were using indexes from the unprojected schema to access the projected one leading to the out of bounds error and panic - this is my understanding.

The solution was to pass the projected schema into the execution plan which is this line.

Downstream of that, this broke `DisplayAs` logic that was summarizing projection info, so I simplified that to omit the column indexes and just display the names of columns projected.

Code after:
        let (batches, predicates) = if let Some(cache) = read
            .get(&self.db_id)
            .and_then(|db| db.get(&self.table_def.table_id))
            .and_then(|tbl| tbl.get(&self.cache_name))
        {
            let predicates = convert_filter_exprs(&self.table_def, self.schema(), filters)?;
            (
                cache
                    .to_record_batch(
                        Arc::clone(&schema),
                        &predicates,
                        projection.map(|p| p.as_slice()),
                        limit,
                    )
                    .map(|batch| vec![batch])?,
                (!predicates.is_empty()).then_some(predicates),
            )
        } else {
            (vec![], None)
        };

        let mut distinct_exec = DistinctCacheExec::try_new(
            predicates,
            Arc::clone(&self.table_def),
            &[batches],
            schema,
            projection.is_some(),
            limit,
        )?;

        let show_sizes = ctx.config_options().explain.show_sizes;
        distinct_exec = distinct_exec.with_show_sizes(show_sizes);

        Ok(Arc::new(distinct_exec))
    }
}

/// Convert the given list of filter expressions to a map of [`ColumnId`] to [`Predicate`]
///
/// The resulting map uses [`IndexMap`] to ensure consistent ordering of the map. This makes testing
/// the filter conversion significantly easier using EXPLAIN queries.
fn convert_filter_exprs(
    table_def: &TableDefinition,
    cache_schema: SchemaRef,
    filters: &[Expr],
) -> Result<IndexMap<ColumnId, Predicate>> {
    let mut predicate_map: IndexMap<ColumnId, Option<Predicate>> = IndexMap::new();

    // for create_physical_expr:
    let schema: DFSchema = cache_schema.try_into()?;
    let props = ExecutionProps::new();