Push down type casts to file readers - #22788

You can verify the PR this way:

~/duckdb duckdb -c "explain select list_sum(GT) from '/home/myrrc/vortex/vortex-bench/data/statpopgen/100000/parquet/gnomad.genomes.v3.1.2.hgdp_tgp.chr21.parquet'"

Before:

┌─────────────────────────────┐
│┌───────────────────────────┐│
││       Physical Plan       ││
│└───────────────────────────┘│
└─────────────────────────────┘
┌───────────────────────────┐
│         PROJECTION        │
│    ────────────────────   │
│    list_aggr(CAST(GT AS   │
│     HUGEINT[]), 'sum')    │
│                           │
│       ~100,000 rows       │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│        PARQUET_SCAN       │
│    ────────────────────   │
│         Function:         │
│        PARQUET_SCAN       │
│                           │
│      Projections: GT      │
│                           │
│       ~100,000 rows       │
└───────────────────────────┘

After:

┌─────────────────────────────┐
│┌───────────────────────────┐│
││       Physical Plan       ││
│└───────────────────────────┘│
└─────────────────────────────┘
┌───────────────────────────┐
│         PROJECTION        │
│    ────────────────────   │
│    list_aggr(#0, 'sum')   │
│                           │
│       ~100,000 rows       │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│        PARQUET_SCAN       │
│    ────────────────────   │
│         Function:         │
│        PARQUET_SCAN       │
│                           │
│      Projections: GT      │
│                           │
│       ~100,000 rows       │
└───────────────────────────┘

Thanks, I'll definitely add more tests today.
As for your specific example, I think it depends whether the file reader supports cast expression pushdowns, i.e. if Parquet did some type casting, this would be fine:

┌───────────────────────────┐
│         PROJECTION        │
│    ────────────────────   │
│   CAST(ts AS TIMESTAMP)   │
│           ~1 row          │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│        PARQUET_SCAN       │
│    ────────────────────   │
│         Function:         │
│        PARQUET_SCAN       │
│                           │
│      Projections: ts      │
│                           │
│          Filters:         │
│ (CAST(ts AS DATE) = '2020 │
│       -01-01'::DATE)      │
│           ~1 row          │
└───────────────────────────┘

In the plan filter and projection are decoupled, so reader function can return casted timestamp.

I think you might be conflating filter pushdown and type pushdown. What type_pushdown specifies is the result type of the given column, i.e. in that example we tell the scan the ts column is of type DATE. This was originally implemented for the CSV reader since types are not known there and we do auto-detection, so this basically assists it by (1) allowing it to skip auto-detection, and (2) prevent it from incorrectly detecting the wrong types. No other readers implement this at the moment.

In the given example if we are telling the scan ts is of type DATE then the query plan is wrong, because later on ts is used as a TIMESTAMP. Imagine if ts was a string inside the source file - if we cast it to DATE in the scan first and emit it as a DATE we lose information (namely the hour/minute/second component of the value) and the query produces wrong results.

In the given example if we are telling the scan ts is of type DATE then the query plan is wrong

No, I'm saying if a file reader supports expression pushdown (cast pushdown specifically), we can request "ts AS TIMESTAMP" as projection, because "ts AS DATE" will be pushed as a filter, and there are no casted or direct usages of ts.

And I also believe the reason Parquet/Arrow readers doesn't support type pushdown is because of UHUGEINT/IHUGEINT. This is the case for Vortex as well. So if you're open to breaking changes for 2.0, I'd suggest type_pushdown should either behave how pushdown_complex_filter does (get a list of output types, push down those supported, let duckdb handle the rest), or how expression pushdown does (separate function can_pushdown_type).
With this API i.e. Vortex would push everything but UHUGEINT/IHUGEINT.

I've pushed an updated version with tests.
This still won't push down all types, i.e. on SELECT list_sum(cast(list as USMALLINT[])) only the inner cast to small int will be propagated, but to fix this you need to make the traversal and replacement run multiple times, and I'm not sure it's the best for performance.

@Mytherin I've pushed an updated version. The question about type casts which are not supported in extension i.e. HUGEINT for Vortex still remains - we won't be able to use pushdown if there's no way to decline the cast.

Maybe this PR should also use expression_pushdown TableFunction callback to check whether this is supported?

pushdown_expression is about table filters so I think that's not the right abstraction here.

We could redesign the type_pushdown to maybe pushing a generic expression instead? e.g.:

bool pushdown_projection(ClientContext &context, const LogicalGet &get, const Expression &expr, idx_t proj_index);

That would be more flexible going forward - then we just wouldn't accept anything other than casts for now.

If we redesign the function we should also move towards using a struct for input parameters (similar to TableFunctionGetPartitionInput and such).

bool pushdown_projection(ClientContext &context, const LogicalGet &get, const Expression &expr, idx_t proj_index);

What if multiple projections for a column were requested e.g. SELECT len(str), substr(str, 2)?

I've thought about this for a while in vortex-data/vortex#8190. If we go this route, Duckdb can request multiple same columns in projections i.e. for the query above
column_ids=[0], projection_ids=[0, 0], and pushdown_projection may be called twice:

pushdown_projection(.., Expression{len}, 0) -- first col in SELECT
pushdown_projection(.., Expression{len}, 1) -- second col in SELECT

However, if we want to further expand this to aggregate functions, we should also be notified of the output types. This shouldn't be hard, though

I would say we can skip projection pushdown in that case similar to how we currently skip cast pushdown if the same column is scanned with different types - but it's an interesting angle to consider perhaps going forward.

Rebased, thanks

Thanks!