Implement Dereference pushdown for the Delta Lake connector

[krvikash]

Description

This PR implements dereference pushdown for Delta Lake connector(similar to #8129).

This adds significant performance improvements for queries accessing nested fields inside struct/row columns. They have been optimized through the pushdown of dereference expressions. With this feature, the query execution prunes structural data eagerly, extracting the necessary fields.

For Example:

I have a table having a nested field col. When perform selecting col.a, we can see the difference in Input and Physical Input values in the query plan when running with and without dereference pushdown.

Table Schema as below:

trino:dereference_pushdown_test> SHOW COLUMNS FROM ntest;
  Column  |          Type           | Extra | Comment
----------+-------------------------+-------+---------
 orderkey | bigint                  |       |
 col      | row(a bigint, b bigint) |       |
(2 rows)

Query Plan without Dereference pushdown:

trino:dereference_pushdown_test> SET SESSION delta_lake.projection_pushdown_enabled=false;
SET SESSION
trino:dereference_pushdown_test> EXPLAIN ANALYZE SELECT col.a FROM ntest;
                                                                                           Query Plan
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Trino version: testversion
 Queued: 453.79us, Analysis: 306.94ms, Planning: 91.70ms, Execution: 490.10ms
 Fragment 1 [SOURCE]
     CPU: 60.38ms, Scheduled: 216.02ms, Blocked 0.00ns (Input: 0.00ns, Output: 0.00ns), Input: 1500000 rows (32.90MB); per task: avg.: 1500000.00 std.dev.: 0.00, Output: 1500000 rows (12.87MB)
     Output layout: [expr]
     Output partitioning: SINGLE []
     ScanProject[table = delta_lake:dereference_pushdown_test.ntest]
         Layout: [expr:bigint]
         Estimates: {rows: 1500000 (12.87MB), cpu: 78.68M, memory: 0B, network: 0B}/{rows: 1500000 (12.87MB), cpu: 12.87M, memory: 0B, network: 0B}
         CPU: 60.00ms (100.00%), Scheduled: 216.00ms (100.00%), Blocked: 0.00ns (?%), Output: 1500000 rows (12.87MB)
         Input avg.: 1500000.00 rows, Input std.dev.: 0.00%
         expr := "col"[1]
         col := col:row(a bigint, b bigint):REGULAR
         Input: 1500000 rows (32.90MB), Filtered: 0.00%, Physical input: 425.08kB, Physical input time: 167700000.00ns

Query Plan with Dereference pushdown:

trino:dereference_pushdown_test> SET SESSION delta_lake.projection_pushdown_enabled=true;
SET SESSION
trino:dereference_pushdown_test> EXPLAIN ANALYZE SELECT col.a FROM ntest;
                                                                                           Query Plan
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Trino version: testversion
 Queued: 431.92us, Analysis: 365.17ms, Planning: 105.14ms, Execution: 501.89ms
 Fragment 1 [SOURCE]
     CPU: 47.47ms, Scheduled: 208.86ms, Blocked 0.00ns (Input: 0.00ns, Output: 0.00ns), Input: 1500000 rows (12.87MB); per task: avg.: 1500000.00 std.dev.: 0.00, Output: 1500000 rows (12.87MB)
     Output layout: [col.a]
     Output partitioning: SINGLE []
     TableScan[table = delta_lake:dereference_pushdown_test.ntest]
         Layout: [col.a:bigint]
         Estimates: {rows: 1500000 (12.87MB), cpu: 12.87M, memory: 0B, network: 0B}
         CPU: 48.00ms (100.00%), Scheduled: 208.00ms (100.00%), Blocked: 0.00ns (?%), Output: 1500000 rows (12.87MB)
         Input avg.: 1500000.00 rows, Input std.dev.: 0.00%
         col.a := col.a:bigint:REGULAR
         Input: 1500000 rows (12.87MB), Physical input: 236.54kB, Physical input time: 175040000.00ns

More Details about dereference pushdown: https://trino.io/blog/2020/08/14/dereference-pushdown.html

Additional context and related issues

The feature is enabled by default.

The feature can be disabled by setting delta.projection-pushdown-enabled configuration property or delta.projection_pushdown_enabled session property to false.

Release notes

(X) Release notes are required, with the following suggested text:

# DeltaLake
* Add Dereference pushdown for the Delta Lake connector. ({issue}`17085`)

[krvikash]

Mostly Copy of https://github.com/trinodb/trino/blob/master/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergProjectionPushdownPlans.java

Note to self: Possible to make it common?

[krvikash]

Duplicate code https://github.com/trinodb/trino/blob/master/plugin/trino-hive/src/test/java/io/trino/plugin/hive/optimizer/TestHiveProjectionPushdownIntoTableScan.java here as well.

[findinpath]

Please add a product test reading from Delta Lake tables (created either by Databricks/Delta OSS) with column mapping mode name & id to make sure that the dereferencing actually works for such tables as well.

[krvikash]

Added some more tests.

[raunaqmorarka]

You could put these tests in a base class in io.trino.testing and then run it for the relevant connectors and file formats by adding derived classes (e.g. BaseOrcWithBloomFiltersTest)

[krvikash]

I am considering moving tests in follow-up PR.

#17085 (comment)

[findinpath]

Great finding Raunaq.

[krvikash]

#17414

[krvikash]

rebased with master and resolved conflicts

[krvikash]

Addressed rest of the comments:

will raise follow-up PR for below comments.
#17085 (comment)
#17085 (comment)

[findinpath]

should we use projected column name here if it dereference field?

I see this as part of #17164

[findinpath]

nit: FYI This seems to be done as well in deserializeColumnValue one line below.
No change needed.

[krvikash]

deserializeColumnValue throw a verification exception to verify the column is base column.

verify(column.isBaseColumn(), "Unexpected dereference: %s", column);

[findinpath]

We should have tests for tables that underwent insert/update/delete/merge with onDelta + dereference pushdown (for all column mapping modes)

@findepi what is the rationale for this request in the context of this read oriented feature ? Is this more about being future-proof?

[findinpath]

@findepi can you pls run this PR with secrets?

[findepi]

/test-with-secrets sha=b6be925a0b09044923a17b37133fd58809e00540

[findepi]

please squash (after the test with secrets run)

[findepi]

I will try to remember to solve this in #17365

[krvikash]

test-with-secrets run is 🟢 https://github.com/trinodb/trino/actions/runs/4912972173

[krvikash]

Addressed comments

[findepi]

@krvikash thank you so much!

[krvikash]

Thank you so much, @findepi | @findinpath | @raunaqmorarka | @alexjo2144 for the reviews and constant support.