Function execute_with_predicate

datafusion/core/tests/fuzz_cases/pruning.rs:273–315

(
    files: &[TestFile],
    predicate: Arc<dyn PhysicalExpr>,
    prune_stats: bool,
    schema: Arc<Schema>,
    ctx: &SessionContext,
)

}

async fn execute_with_predicate(
    files: &[TestFile],
    predicate: Arc<dyn PhysicalExpr>,
    prune_stats: bool,
    schema: Arc<Schema>,
    ctx: &SessionContext,
) -> Vec<String> {
    let parquet_source = if prune_stats {
        ParquetSource::new(schema.clone()).with_predicate(predicate.clone())
    } else {
        ParquetSource::new(schema.clone())
    };
    let config = FileScanConfigBuilder::new(
        ObjectStoreUrl::parse("memory://").unwrap(),
        Arc::new(parquet_source),
    )
    .with_file_group(
        files
            .iter()
            .map(|test_file| {
                PartitionedFile::new(test_file.path.clone(), test_file.size as u64)
            })
            .collect(),
    )
    .build();
    let exec = DataSourceExec::from_data_source(config);
    let exec =
        Arc::new(FilterExec::try_new(predicate, exec).unwrap()) as Arc<dyn ExecutionPlan>;

    let batches = collect(exec, ctx.task_ctx()).await.unwrap();
    let mut values = vec![];
    for batch in batches {
        let column = batch
            .column(0)
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        for i in 0..column.len() {
            values.push(column.value(i).to_string());
        }
    }
    values
}

async fn write_parquet_file(
    truncation_length: Option<usize>,