Implement Dereference pushdown for the Delta Lake connector #17085

krvikash · 2023-04-18T07:06:34Z

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 · 2023-04-18T18:35:57Z

...delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeProjectionPushdownPlans.java

+import static java.lang.String.format;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class TestDeltaLakeProjectionPushdownPlans


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?

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 · 2023-04-18T20:21:10Z

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.

.../trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeColumnProjectionInfo.java

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeConfig.java

...n/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/MetadataEntry.java

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java

krvikash · 2023-04-18T22:14:30Z

Added some more tests.

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

...rino-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeMinioConnectorTest.java

testing/trino-testing/src/main/java/io/trino/testing/TestingConnectorBehavior.java

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeColumnHandle.java

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

raunaqmorarka · 2023-04-19T19:02:11Z

...rino-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeMinioConnectorTest.java

+    @Test
+    public void testHighlyNestedData()
+    {
+        // TODO consider moving this in BaseConnectorTest


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)

I am considering moving tests in follow-up PR.

#17085 (comment)

Great finding Raunaq.

...rino-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeMinioConnectorTest.java

krvikash · 2023-05-05T09:53:21Z

Addressed most of the comments. Some of the comments are yet to be addressed.

krvikash · 2023-05-05T09:57:33Z

rebased with master and resolved conflicts

.../trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeColumnProjectionInfo.java

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeColumnHandle.java

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

krvikash · 2023-05-05T13:34:31Z

Addressed rest of the comments:

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

findinpath · 2023-05-06T04:34:38Z

...ake/src/main/java/io/trino/plugin/deltalake/statistics/FileBasedTableStatisticsProvider.java

@@ -102,7 +103,7 @@ public TableStatistics getTableStatistics(ConnectorSession session, DeltaLakeTab
        }

        Set<String> predicatedColumnNames = tableHandle.getNonPartitionConstraint().getDomains().orElseThrow().keySet().stream()
-                .map(DeltaLakeColumnHandle::getName)
+                .map(DeltaLakeColumnHandle::getBaseColumnName)


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

I see this as part of #17164

findinpath · 2023-05-06T05:11:14Z

...in/java/io/trino/plugin/deltalake/transactionlog/statistics/DeltaLakeJsonFileStatistics.java

        return value.flatMap(o -> deserializeStatisticsValue(columnHandle, String.valueOf(o)));
    }

    private Optional<Object> deserializeStatisticsValue(DeltaLakeColumnHandle columnHandle, String statValue)
    {
+        if (!columnHandle.isBaseColumn()) {


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

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

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

findinpath · 2023-05-06T05:27:44Z

...uct-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeColumnMappingMode.java

@@ -55,20 +55,20 @@ public void testColumnMappingModeNone()

        onDelta().executeQuery("" +
                "CREATE TABLE default." + tableName +
-                " (a_number INT)" +
+                " (a_number INT, nested STRUCT<field1: STRING, field2: STRING>)" +


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 · 2023-05-06T05:39:24Z

@findepi can you pls run this PR with secrets?

findepi · 2023-05-08T07:12:24Z

/test-with-secrets sha=b6be925a0b09044923a17b37133fd58809e00540

findepi

please squash (after the test with secrets run)

findepi · 2023-05-08T08:07:08Z

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

-        return getColumns(table.getMetadataEntry()).stream()
+        return table.getProjectedColumns()
+                .map(projectedColumns -> (List<DeltaLakeColumnHandle>) projectedColumns.stream()
+                        .map(DeltaLakeColumnHandle.class::cast) // TODO DeltaLakeTableHandle.projectedColumns should be a collection of DeltaLakeColumnHandle


I will try to remember to solve this in #17365

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeSessionProperties.java

...uct-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeColumnMappingMode.java

krvikash · 2023-05-08T12:06:58Z

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

krvikash · 2023-05-08T12:13:48Z

Addressed comments

findepi · 2023-05-08T12:46:57Z

@krvikash thank you so much!

krvikash · 2023-05-08T12:48:51Z

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

cla-bot bot added the cla-signed label Apr 18, 2023

krvikash requested review from ebyhr, alexjo2144 and findinpath April 18, 2023 07:16

krvikash self-assigned this Apr 18, 2023

github-actions bot added delta-lake Delta Lake connector hive Hive connector tests:hive labels Apr 18, 2023

krvikash force-pushed the delta-dereference-pushdown branch 2 times, most recently from ef81ebc to b0b04f0 Compare April 18, 2023 18:32

krvikash commented Apr 18, 2023

View reviewed changes

github-actions bot added the iceberg Iceberg connector label Apr 18, 2023

findinpath reviewed Apr 18, 2023

View reviewed changes

.../trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeColumnProjectionInfo.java Show resolved Hide resolved

findinpath reviewed Apr 18, 2023

View reviewed changes

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeConfig.java Show resolved Hide resolved

findinpath reviewed Apr 18, 2023

View reviewed changes

...n/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/MetadataEntry.java Outdated Show resolved Hide resolved

findinpath reviewed Apr 18, 2023

View reviewed changes

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java Outdated Show resolved Hide resolved

krvikash force-pushed the delta-dereference-pushdown branch from b0b04f0 to 9fe1849 Compare April 18, 2023 22:13

ebyhr reviewed Apr 19, 2023

View reviewed changes

krvikash force-pushed the delta-dereference-pushdown branch 4 times, most recently from b1c173b to 3e3167b Compare April 19, 2023 15:35

github-actions bot added the docs label Apr 19, 2023

krvikash force-pushed the delta-dereference-pushdown branch from 3e3167b to 46dc82e Compare April 19, 2023 16:11

krvikash removed the iceberg Iceberg connector label Apr 19, 2023

raunaqmorarka reviewed Apr 19, 2023

View reviewed changes

krvikash force-pushed the delta-dereference-pushdown branch 2 times, most recently from 41659e2 to 7e7acaa Compare April 19, 2023 21:18

krvikash added 3 commits May 5, 2023 15:26

Add getMergeRowIdColumnHandle method in DeltaLakeColumnHandle

65bfc39

Include fieldId and physicalName to calculate retained size in bytes

e611bdc

Remove unnecessary metadata call in DeltaLakeMetadata#beginMerge

7402b71

krvikash force-pushed the delta-dereference-pushdown branch from 4eee057 to dd60e5c Compare May 5, 2023 09:57

krvikash commented May 5, 2023

View reviewed changes

krvikash force-pushed the delta-dereference-pushdown branch from dd60e5c to b6be925 Compare May 5, 2023 12:28

findinpath reviewed May 6, 2023

View reviewed changes

findinpath approved these changes May 6, 2023

View reviewed changes

findepi approved these changes May 8, 2023

View reviewed changes

Implement Dereference pushdown for the Delta Lake connector

1b9b988

krvikash force-pushed the delta-dereference-pushdown branch from b6be925 to 1b9b988 Compare May 8, 2023 12:11

findepi merged commit 43e90f4 into trinodb:master May 8, 2023
7 of 14 checks passed

findepi mentioned this pull request May 8, 2023

Release notes for 417 #17339

Closed

alexjo2144 mentioned this pull request May 8, 2023

Fix reference to createTestingFileHiveMetastore in test #17388

Merged

github-actions bot added this to the 417 milestone May 8, 2023

krvikash deleted the delta-dereference-pushdown branch May 8, 2023 16:59

colebow mentioned this pull request May 10, 2023

Add Trino 417 release notes #17447

Merged

This was referenced May 10, 2023

Ensure statistics predicate only for base column in delta connector #17449

Merged

Add projection pushdown tests in BaseConnectorTest #17414

Merged

krvikash mentioned this pull request May 31, 2023

Implement dereference pushdown for MongoDB connector #17710

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implement Dereference pushdown for the Delta Lake connector #17085

Implement Dereference pushdown for the Delta Lake connector #17085

krvikash commented Apr 18, 2023 •

edited

krvikash Apr 18, 2023

krvikash Apr 18, 2023

findinpath commented Apr 18, 2023

krvikash commented Apr 18, 2023

raunaqmorarka Apr 19, 2023

krvikash Apr 19, 2023

findinpath Apr 21, 2023

krvikash May 10, 2023

krvikash commented May 5, 2023

krvikash commented May 5, 2023

krvikash commented May 5, 2023

findinpath May 6, 2023

findinpath May 6, 2023

krvikash May 8, 2023

findinpath May 6, 2023

findinpath commented May 6, 2023

findepi commented May 8, 2023

findepi left a comment

findepi May 8, 2023

krvikash commented May 8, 2023 •

edited

krvikash commented May 8, 2023

findepi commented May 8, 2023

krvikash commented May 8, 2023

Implement Dereference pushdown for the Delta Lake connector #17085

Implement Dereference pushdown for the Delta Lake connector #17085

Conversation

krvikash commented Apr 18, 2023 • edited

Description

Additional context and related issues

Release notes

Choose a reason for hiding this comment

Choose a reason for hiding this comment

findinpath commented Apr 18, 2023

krvikash commented Apr 18, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

krvikash commented May 5, 2023

krvikash commented May 5, 2023

krvikash commented May 5, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

findinpath commented May 6, 2023

findepi commented May 8, 2023

findepi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

krvikash commented May 8, 2023 • edited

krvikash commented May 8, 2023

findepi commented May 8, 2023

krvikash commented May 8, 2023

krvikash commented Apr 18, 2023 •

edited

krvikash commented May 8, 2023 •

edited