Trino Merge query scanning complete target table.

[ratuldawar11]

I have a use case of merge query where I use MERGE query to provide upsert type of semantics in my iceberg table.

MERGE query leads to complete scan from the target table which leads to drastic query inefficiencies which increases with scale.

This should have been optimised by dynamic filtering but complete table is still getting scanned

Steps to reproduce it :

Creating source table

CREATE TABLE catalog.schema.multiple_partitions_test (
    id INTEGER,
    name VARCHAR,
    is_active BOOLEAN,
    created_at TIMESTAMP,
    price DOUBLE,
    quantity BIGINT,
    date DATE,
    outlet_id INTEGER
) WITH (
    format = 'PARQUET', 
    partitioning = ARRAY['date', 'outlet_id']
) 

INSERT INTO catalog.schema.multiple_partitions_test (id, name, is_active, created_at, price, quantity, date, outlet_id) VALUES
(1, 'Product A', TRUE, TIMESTAMP '2023-09-01 12:00:00', 19.99, 100, DATE '2023-09-01', 101),
(2, 'Product B', FALSE, TIMESTAMP '2023-08-15 08:30:00', 29.99, 50, DATE '2023-08-15', 102),
(3, 'Product C', TRUE, TIMESTAMP '2023-07-20 09:45:00', 39.99, 75, DATE '2023-07-20', 103),
(4, 'Product D', TRUE, TIMESTAMP '2023-06-10 14:20:00', 49.99, 200, DATE '2023-06-10', 101),
(5, 'Product E', FALSE, TIMESTAMP '2023-05-05 11:00:00', 59.99, 60, DATE '2023-05-05', 102),
(6, 'Product F', TRUE, TIMESTAMP '2023-04-15 16:30:00', 69.99, 90, DATE '2023-04-15', 103),
(7, 'Product G', FALSE, TIMESTAMP '2023-03-25 13:15:00', 79.99, 120, DATE '2023-03-25', 101),
(8, 'Product H', TRUE, TIMESTAMP '2023-02-10 10:00:00', 89.99, 30, DATE '2023-02-10', 102),
(9, 'Product I', TRUE, TIMESTAMP '2023-01-05 09:30:00', 99.99, 85, DATE '2023-01-05', 103),
(10, 'Product J', FALSE, TIMESTAMP '2023-09-02 12:00:00', 19.99, 45, DATE '2023-09-02', 101),
(11, 'Product K', TRUE, TIMESTAMP '2023-08-16 08:30:00', 29.99, 95, DATE '2023-08-16', 102),
(12, 'Product L', FALSE, TIMESTAMP '2023-07-21 09:45:00', 39.99, 65, DATE '2023-07-21', 103),
(13, 'Product M', TRUE, TIMESTAMP '2023-06-11 14:20:00', 49.99, 110, DATE '2023-06-11', 101),
(14, 'Product N', FALSE, TIMESTAMP '2023-05-06 11:00:00', 59.99, 55, DATE '2023-05-06', 102),
(15, 'Product O', TRUE, TIMESTAMP '2023-04-16 16:30:00', 69.99, 80, DATE '2023-04-16', 103),
(16, 'Product P', FALSE, TIMESTAMP '2023-03-26 13:15:00', 79.99, 100, DATE '2023-03-26', 101),
(17, 'Product Q', TRUE, TIMESTAMP '2023-02-11 10:00:00', 89.99, 40, DATE '2023-02-11', 102),
(18, 'Product R', FALSE, TIMESTAMP '2023-01-06 09:30:00', 99.99, 70, DATE '2023-01-06', 103),
(19, 'Product S', TRUE, TIMESTAMP '2023-09-03 12:00:00', 19.99, 50, DATE '2023-09-03', 101),
(20, 'Product T', FALSE, TIMESTAMP '2023-08-17 08:30:00', 29.99, 85, DATE '2023-08-17', 102) 

Creating target table

CREATE TABLE catalog.schema.multiple_partitions_test_target (
    id INTEGER,
    name VARCHAR,
    is_active BOOLEAN,
    created_at TIMESTAMP,
    price DOUBLE,
    quantity BIGINT,
    date DATE,
    outlet_id INTEGER
) WITH (
    format = 'PARQUET', 
    partitioning = ARRAY['date', 'outlet_id']
) 

INSERT INTO catalog.schema.multiple_partitions_test_target (id, name, is_active, created_at, price, quantity, date, outlet_id) VALUES
(1, 'Product A', TRUE, TIMESTAMP '2023-09-01 12:00:00', 19.99, 100, DATE '2023-09-01', 101),
(2, 'Product B', FALSE, TIMESTAMP '2023-08-15 08:30:00', 29.99, 50, DATE '2023-08-15', 102),
(3, 'Product C', TRUE, TIMESTAMP '2023-07-20 09:45:00', 39.99, 75, DATE '2023-07-20', 103),
(4, 'Product D', TRUE, TIMESTAMP '2023-06-10 14:20:00', 49.99, 200, DATE '2023-06-10', 101),
(5, 'Product E', FALSE, TIMESTAMP '2023-05-05 11:00:00', 59.99, 60, DATE '2023-05-05', 102),
(6, 'Product F', TRUE, TIMESTAMP '2023-04-15 16:30:00', 69.99, 90, DATE '2023-04-15', 103),
(7, 'Product G', FALSE, TIMESTAMP '2023-03-25 13:15:00', 79.99, 120, DATE '2023-03-25', 101),
(8, 'Product H', TRUE, TIMESTAMP '2023-02-10 10:00:00', 89.99, 30, DATE '2023-02-10', 102),
(9, 'Product I', TRUE, TIMESTAMP '2023-01-05 09:30:00', 99.99, 85, DATE '2023-01-05', 103),
(10, 'Product J', FALSE, TIMESTAMP '2023-09-02 12:00:00', 19.99, 45, DATE '2023-09-02', 101),
(11, 'Product K', TRUE, TIMESTAMP '2023-08-16 08:30:00', 29.99, 95, DATE '2023-08-16', 102),
(12, 'Product L', FALSE, TIMESTAMP '2023-07-21 09:45:00', 39.99, 65, DATE '2023-07-21', 103),
(13, 'Product M', TRUE, TIMESTAMP '2023-06-11 14:20:00', 49.99, 110, DATE '2023-06-11', 101),
(14, 'Product N', FALSE, TIMESTAMP '2023-05-06 11:00:00', 59.99, 55, DATE '2023-05-06', 102),
(15, 'Product O', TRUE, TIMESTAMP '2023-04-16 16:30:00', 69.99, 80, DATE '2023-04-16', 103),
(16, 'Product P', FALSE, TIMESTAMP '2023-03-26 13:15:00', 79.99, 100, DATE '2023-03-26', 101),
(17, 'Product Q', TRUE, TIMESTAMP '2023-02-11 10:00:00', 89.99, 40, DATE '2023-02-11', 102),
(18, 'Product R', FALSE, TIMESTAMP '2023-01-06 09:30:00', 99.99, 70, DATE '2023-01-06', 103),
(19, 'Product S', TRUE, TIMESTAMP '2023-09-03 12:00:00', 19.99, 50, DATE '2023-09-03', 101),
(20, 'Product T', FALSE, TIMESTAMP '2023-08-17 08:30:00', 29.99, 85, DATE '2023-08-17', 102) 

Merge query with just comparision filter (with target table) and no explicit filter on partition key leads to complete table scan

MERGE INTO "catalog"."schema"."multiple_partitions_test_target" AS t
                USING (SELECT id,name,is_active,created_at,price,quantity,date,outlet_id FROM  "schema"."catalog"."multiple_partitions_test"   where date IS NOT NULL) AS s
                    ON ((s.id = t.id) AND (s.outlet_id = t.outlet_id) AND (s.date = t.date))  AND t.date is not null  
                    WHEN MATCHED
                        THEN UPDATE SET id=s.id,name=s.name,is_active=s.is_active,created_at=s.created_at,price=s.price,quantity=s.quantity,date=s.date,outlet_id=s.outlet_id
                    WHEN NOT MATCHED
                        THEN INSERT (id,name,is_active,created_at,price,quantity,date,outlet_id)
                        VALUES (s.id,s.name,s.is_active,s.created_at,s.price,s.quantity,s.date,s.outlet_id)

Merge query with relevant filters leads to a single row scan

MERGE INTO "catalog"."schema"."multiple_partitions_test_target" AS t
                USING (SELECT id,name,is_active,created_at,price,quantity,date,outlet_id FROM  "catalog"."schema"."multiple_partitions_test"   where date IS NOT NULL) AS s
                    ON ((s.id = t.id) AND (s.outlet_id = t.outlet_id) AND (s.date = t.date))  AND (t.date,t.outlet_id) IN ((CAST('2023-04-15' AS DATE),CAST('103' AS INTEGER)))
                    WHEN MATCHED
                        THEN UPDATE SET id=s.id,name=s.name,is_active=s.is_active,created_at=s.created_at,price=s.price,quantity=s.quantity,date=s.date,outlet_id=s.outlet_id
                    WHEN NOT MATCHED
                        THEN INSERT (id,name,is_active,created_at,price,quantity,date,outlet_id)
                        VALUES (s.id,s.name,s.is_active,s.created_at,s.price,s.quantity,s.date,s.outlet_id)

This leads to increasing inefficiencies when data in target table keeps on increasing. Any help will be appreciated

[findinpath]

cc @raunaqmorarka

[raunaqmorarka]

Dynamic filtering requires a RIGHT JOIN. In this case the CBO is picking LEFT JOIN because the target table is small enough to be broadcast as the build side of a LEFT JOIN.
The CBO will stop doing that when the target table gets bigger.
You can manually force a RIGHT JOIN at smaller sizes using SET SESSION join_distribution_type = 'PARTITIONED' or tuning join_max_broadcast_table_size lower.

[ratuldawar11]

Hey @raunaqmorarka I have tried the config you provided, also tried it with a larger table too. Even with right join the whole target table is getting scanned. Can you please help me. Any help would be appreciated !

[raunaqmorarka]

Can you share output of EXPLAIN ANALYZE with right join ?

[ratuldawar11]

Here are all the details @raunaqmorarka

CREATE TABLE blinkit_iceberg_staging.de_etls.merge_query_source_table (
    id INTEGER,
    name VARCHAR,
    is_active BOOLEAN,
    created_at TIMESTAMP,
    price DOUBLE,
    quantity BIGINT,
    date DATE,
    outlet_id INTEGER
) WITH (
    format = 'PARQUET', 
    partitioning = ARRAY['date', 'outlet_id']
) 

INSERT INTO blinkit_iceberg_staging.de_etls.merge_query_source_table (id, name, is_active, created_at, price, quantity, date, outlet_id) VALUES
(1, 'Product A', TRUE, TIMESTAMP '2023-09-01 12:00:00', 19.99, 100, DATE '2023-09-01', 101)


CREATE TABLE blinkit_iceberg_staging.de_etls.merge_query_target_table (
    id INTEGER,
    name VARCHAR,
    is_active BOOLEAN,
    created_at TIMESTAMP,
    price DOUBLE,
    quantity BIGINT,
    date DATE,
    outlet_id INTEGER
) WITH (
    format = 'PARQUET', 
    partitioning = ARRAY['date', 'outlet_id']
) 


INSERT INTO blinkit_iceberg_staging.de_etls.merge_query_target_table (id, name, is_active, created_at, price, quantity, date, outlet_id) VALUES
(1, 'Product A', TRUE, TIMESTAMP '2023-09-01 12:00:00', 19.99, 100, DATE '2023-09-01', 101),
(2, 'Product B', FALSE, TIMESTAMP '2023-08-15 08:30:00', 29.99, 50, DATE '2023-08-15', 102),
(3, 'Product C', TRUE, TIMESTAMP '2023-07-20 09:45:00', 39.99, 75, DATE '2023-07-20', 103),
(4, 'Product D', TRUE, TIMESTAMP '2023-06-10 14:20:00', 49.99, 200, DATE '2023-06-10', 101),
(5, 'Product E', FALSE, TIMESTAMP '2023-05-05 11:00:00', 59.99, 60, DATE '2023-05-05', 102),
(6, 'Product F', TRUE, TIMESTAMP '2023-04-15 16:30:00', 69.99, 90, DATE '2023-04-15', 103),
(7, 'Product G', FALSE, TIMESTAMP '2023-03-25 13:15:00', 79.99, 120, DATE '2023-03-25', 101),
(8, 'Product H', TRUE, TIMESTAMP '2023-02-10 10:00:00', 89.99, 30, DATE '2023-02-10', 102),
(9, 'Product I', TRUE, TIMESTAMP '2023-01-05 09:30:00', 99.99, 85, DATE '2023-01-05', 103),
(10, 'Product J', FALSE, TIMESTAMP '2023-09-02 12:00:00', 19.99, 45, DATE '2023-09-02', 101),
(11, 'Product K', TRUE, TIMESTAMP '2023-08-16 08:30:00', 29.99, 95, DATE '2023-08-16', 102),
(12, 'Product L', FALSE, TIMESTAMP '2023-07-21 09:45:00', 39.99, 65, DATE '2023-07-21', 103),
(13, 'Product M', TRUE, TIMESTAMP '2023-06-11 14:20:00', 49.99, 110, DATE '2023-06-11', 101),
(14, 'Product N', FALSE, TIMESTAMP '2023-05-06 11:00:00', 59.99, 55, DATE '2023-05-06', 102),
(15, 'Product O', TRUE, TIMESTAMP '2023-04-16 16:30:00', 69.99, 80, DATE '2023-04-16', 103),
(16, 'Product P', FALSE, TIMESTAMP '2023-03-26 13:15:00', 79.99, 100, DATE '2023-03-26', 101),
(17, 'Product Q', TRUE, TIMESTAMP '2023-02-11 10:00:00', 89.99, 40, DATE '2023-02-11', 102),
(18, 'Product R', FALSE, TIMESTAMP '2023-01-06 09:30:00', 99.99, 70, DATE '2023-01-06', 103),
(19, 'Product S', TRUE, TIMESTAMP '2023-09-03 12:00:00', 19.99, 50, DATE '2023-09-03', 101),
(20, 'Product T', FALSE, TIMESTAMP '2023-08-17 08:30:00', 29.99, 85, DATE '2023-08-17', 102) 



MERGE INTO blinkit_iceberg_staging.de_etls.merge_query_target_table AS t
                USING (SELECT id,name,is_active,created_at,price,quantity,date,outlet_id FROM  blinkit_iceberg_staging.de_etls.merge_query_source_table    where date IS NOT NULL) AS s
                    ON ((s.id = t.id) AND (s.outlet_id = t.outlet_id) AND (s.date = t.date))  
                    WHEN MATCHED
                        THEN UPDATE SET id=s.id,name=s.name,is_active=s.is_active,created_at=s.created_at,price=s.price,quantity=s.quantity,date=s.date,outlet_id=s.outlet_id
                    WHEN NOT MATCHED
                        THEN INSERT (id,name,is_active,created_at,price,quantity,date,outlet_id)
                        VALUES (s.id,s.name,s.is_active,s.created_at,s.price,s.quantity,s.date,s.outlet_id)

20 rows getting scanned from the target table even thought the source table just had 1 row. So according to dynamic filtering only a single row should have been scanned.

EXPLAIN TYPE(LOGICAL )

Trino version: 455 Output[columnNames = [rows]] │ Layout: [rows:bigint] │ Estimates: {rows: ? (?), cpu: 0, memory: 0B, network: 0B} └─ TableCommit[target = blinkit_iceberg_staging:de_etls.merge_query_target_table$data@2591632568254666277 constraint on [date]] │ Layout: [rows:bigint] └─ LocalExchange[partitioning = SINGLE] │ Layout: [partialrows:bigint, fragment:varbinary] │ Estimates: {rows: ? (?), cpu: 0, memory: 0B, network: 0B} └─ RemoteExchange[type = GATHER] │ Layout: [partialrows:bigint, fragment:varbinary] │ Estimates: {rows: ? (?), cpu: 0, memory: 0B, network: ?} └─ MergeWriter[table = blinkit_iceberg_staging:de_etls.merge_query_target_table$data@2591632568254666277 constraint on [date]] │ Layout: [partialrows:bigint, fragment:varbinary] └─ LocalExchange[partitioning = MERGE [insert = HASH, update = blinkit_iceberg_staging:INSTANCE], arguments = [operation::tinyint, date::date, outlet_id::integer, field::row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar)]] │ Layout: [id:integer, name:varchar, is_active:boolean, created_at:timestamp(6), price:double, quantity:bigint, date:date, outlet_id:integer, operation:tinyint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), insert_from_update:tinyint] │ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B} └─ RemoteExchange[type = REPARTITION] │ Layout: [id:integer, name:varchar, is_active:boolean, created_at:timestamp(6), price:double, quantity:bigint, date:date, outlet_id:integer, operation:tinyint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), insert_from_update:tinyint] │ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} └─ MergeProcessor[] │ Layout: [id:integer, name:varchar, is_active:boolean, created_at:timestamp(6), price:double, quantity:bigint, date:date, outlet_id:integer, operation:tinyint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), insert_from_update:tinyint] │ target: blinkit_iceberg_staging:de_etls.merge_query_target_table$data@2591632568254666277 │ merge row column: merge_row │ row id column: field │ redistribution columns: [date, outlet_id] │ data columns: [id, name, is_active, created_at, price, quantity, date, outlet_id] └─ Filter[filterPredicate = (CASE WHEN ($not(is_distinct) AND $not((unique_id IS NULL))) THEN CAST(fail(integer '119', varchar 'One MERGE target table row matched more than one source row') AS boolean) ELSE boolean 'true' END)] │ Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer), case_number:integer, is_distinct:boolean] │ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B} └─ MarkDistinct[distinct = [unique_id:bigint, case_number:integer], marker = is_distinct] │ Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer), case_number:integer, is_distinct:boolean] └─ LocalExchange[partitioning = HASH, arguments = [unique_id::bigint, case_number::integer]] │ Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer), case_number:integer] │ Estimates: {rows: 0 (65B), cpu: 65, memory: 0B, network: 0B} └─ RemoteExchange[type = REPARTITION] │ Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer), case_number:integer] │ Estimates: {rows: 0 (65B), cpu: 65, memory: 0B, network: 65B} └─ Project[] │ Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer), case_number:integer] │ Estimates: {rows: 0 (65B), cpu: 65, memory: 0B, network: 0B} │ case_number := merge_row.10 └─ Project[] │ Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer)] │ Estimates: {rows: 0 (63B), cpu: 63, memory: 0B, network: 0B} │ merge_row := (CASE WHEN present THEN ROW (id_0, name_1, is_active_2, created_at_3, price_4, quantity_5, date_6, outlet_id_7, $not((present IS NULL)), tinyint '3', integer '0') WHEN (present IS NULL) THEN ROW (id_0, name_1, is_active_2, created_at_3, price_4, quantity_5, date_6, outlet_id_7, $not((present IS NULL)), tinyint '1', integer '1') ELSE ROW (null::integer, null::varchar, null::boolean, null::timestamp(6), null::double, null::bigint, null::date, null::integer, $not((present IS NULL)), tinyint '-1', integer '-1') END) └─ RightJoin[criteria = (id = id_0) AND (outlet_id = outlet_id_7) AND (date = date_6), distribution = PARTITIONED] │ Layout: [date:date, outlet_id:integer, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), unique_id:bigint, present:boolean, id_0:integer, name_1:varchar, is_active_2:boolean, created_at_3:timestamp(6), price_4:double, quantity_5:bigint, date_6:date, outlet_id_7:integer] │ Estimates: {rows: 0 (110B), cpu: 2.00k, memory: 151B, network: 0B} │ Distribution: PARTITIONED │ dynamicFilterAssignments = {id_0 -> #df_891, outlet_id_7 -> #df_892, date_6 -> #df_893} ├─ RemoteExchange[type = REPARTITION] │ │ Layout: [id:integer, date:date, outlet_id:integer, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), unique_id:bigint, present:boolean] │ │ Estimates: {rows: 22 (1.74kB), cpu: 1.74k, memory: 0B, network: 1.74kB} │ └─ Project[] │ │ Layout: [id:integer, date:date, outlet_id:integer, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), unique_id:bigint, present:boolean] │ │ Estimates: {rows: 22 (1.74kB), cpu: 1.74k, memory: 0B, network: 0B} │ │ present := boolean 'true' │ └─ AssignUniqueId[] │ │ Layout: [id:integer, date:date, outlet_id:integer, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), unique_id:bigint] │ │ Estimates: {rows: 22 (1.70kB), cpu: 198, memory: 0B, network: 0B} │ └─ ScanFilter[table = blinkit_iceberg_staging:de_etls.merge_query_target_table$data@2591632568254666277 constraint on [date], dynamicFilters = {id = #df_891, outlet_id = #df_892, date = #df_893}] │ Layout: [id:integer, date:date, outlet_id:integer, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar)] │ Estimates: {rows: 22 (1.50kB), cpu: 1.50k, memory: 0B, network: 0B}/{rows: 22 (1.50kB), cpu: 1.50k, memory: 0B, network: 0B} │ id := 1:id:integer │ date := 7:date:date │ :: [(, )] │ field := -2147483647:$row_id:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar) │ outlet_id := 8:outlet_id:integer └─ LocalExchange[partitioning = SINGLE] │ Layout: [id_0:integer, name_1:varchar, is_active_2:boolean, created_at_3:timestamp(6), price_4:double, quantity_5:bigint, date_6:date, outlet_id_7:integer] │ Estimates: {rows: 1 (151B), cpu: 0, memory: 0B, network: 0B} └─ RemoteExchange[type = REPARTITION] │ Layout: [id_0:integer, name_1:varchar, is_active_2:boolean, created_at_3:timestamp(6), price_4:double, quantity_5:bigint, date_6:date, outlet_id_7:integer] │ Estimates: {rows: 1 (151B), cpu: 151, memory: 0B, network: 151B} └─ TableScan[table = blinkit_iceberg_staging:de_etls.merge_query_source_table$data@8453901305866021703 constraint on [date]] Layout: [id_0:integer, name_1:varchar, is_active_2:boolean, created_at_3:timestamp(6), price_4:double, quantity_5:bigint, date_6:date, outlet_id_7:integer] Estimates: {rows: 1 (151B), cpu: 151, memory: 0B, network: 0B} name_1 := 2:name:varchar id_0 := 1:id:integer outlet_id_7 := 8:outlet_id:integer created_at_3 := 4:created_at:timestamp(6) is_active_2 := 3:is_active:boolean quantity_5 := 6:quantity:bigint price_4 := 5:price:double date_6 := 7:date:date :: [(, )]

EXPLAIN ANALYZE

Trino version: 455 Queued: 148.83us, Analysis: 176.45ms, Planning: 149.41ms, Execution: 1.44s Fragment 1 [COORDINATOR_ONLY] CPU: 48.69ms, Scheduled: 693.27ms, Blocked 4.71s (Input: 4.18s, Output: 0.00ns), Input: 418 rows (8.46kB); per task: avg.: 418.00 std.dev.: 0.00, Output: 1 row (9B) Output layout: [rows] Output partitioning: SINGLE [] TableCommit[target = blinkit_iceberg_staging:de_etls.merge_query_target_table$data@2591632568254666277 constraint on [date]] │ Layout: [rows:bigint] │ CPU: 32.00ms (4.37%), Scheduled: 654.00ms (25.20%), Blocked: 0.00ns (0.00%), Output: 1 row (9B) │ Input avg.: 0.00 rows, Input std.dev.: ?% └─ LocalExchange[partitioning = SINGLE] │ Layout: [partialrows:bigint, fragment:varbinary] │ Estimates: {rows: ? (?), cpu: 0, memory: 0B, network: 0B} │ CPU: 4.00ms (0.55%), Scheduled: 4.00ms (0.15%), Blocked: 524.00ms (0.29%), Output: 418 rows (8.46kB) │ Input avg.: 52.25 rows, Input std.dev.: 130.27% └─ RemoteSource[sourceFragmentIds = [2]] Layout: [partialrows:bigint, fragment:varbinary] CPU: 10.00ms (1.36%), Scheduled: 31.00ms (1.19%), Blocked: 4.18s (2.32%), Output: 418 rows (8.46kB) Input avg.: 52.25 rows, Input std.dev.: 130.27% Fragment 2 [MERGE [insert = HASH, update = blinkit_iceberg_staging:INSTANCE]] CPU: 410.95ms, Scheduled: 741.64ms, Blocked 1.89m (Input: 42.29s, Output: 0.00ns), Input: 2 rows (403B); per task: avg.: 0.13 std.dev.: 0.34, Output: 418 rows (8.46kB) Amount of input data processed by the workers for this stage might be skewed Output layout: [partialrows, fragment] Output partitioning: SINGLE [] MergeWriter[table = blinkit_iceberg_staging:de_etls.merge_query_target_table$data@2591632568254666277 constraint on [date]] │ Layout: [partialrows:bigint, fragment:varbinary] │ CPU: 343.00ms (46.79%), Scheduled: 493.00ms (19.00%), Blocked: 0.00ns (0.00%), Output: 418 rows (8.46kB) │ Input avg.: 0.00 rows, Input std.dev.: 1438.75% └─ LocalExchange[partitioning = MERGE [insert = HASH, update = blinkit_iceberg_staging:INSTANCE], arguments = [operation::tinyint, date::date, outlet_id::integer, field::row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar)]] │ Layout: [id:integer, name:varchar, is_active:boolean, created_at:timestamp(6), price:double, quantity:bigint, date:date, outlet_id:integer, operation:tinyint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), insert_from_update:tinyint] │ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B} │ CPU: 11.00ms (1.50%), Scheduled: 21.00ms (0.81%), Blocked: 33.82s (18.81%), Output: 2 rows (403B) │ Input avg.: 0.01 rows, Input std.dev.: 1090.87% └─ RemoteSource[sourceFragmentIds = [3]] Layout: [id:integer, name:varchar, is_active:boolean, created_at:timestamp(6), price:double, quantity:bigint, date:date, outlet_id:integer, operation:tinyint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), insert_from_update:tinyint] CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Blocked: 42.29s (23.52%), Output: 2 rows (403B) Input avg.: 0.01 rows, Input std.dev.: 1090.87% Fragment 3 [HASH] CPU: 71.89ms, Scheduled: 209.37ms, Blocked 1.26m (Input: 37.99s, Output: 0.00ns), Input: 1 row (355B); per task: avg.: 0.07 std.dev.: 0.25, Output: 2 rows (411B) Amount of input data processed by the workers for this stage might be skewed Output layout: [id, name, is_active, created_at, price, quantity, date, outlet_id, operation, field, insert_from_update] Output partitioning: MERGE [insert = HASH, update = blinkit_iceberg_staging:INSTANCE] [operation, date, outlet_id, field] MergeProcessor[] │ Layout: [id:integer, name:varchar, is_active:boolean, created_at:timestamp(6), price:double, quantity:bigint, date:date, outlet_id:integer, operation:tinyint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), insert_from_update:tinyint] │ CPU: 8.00ms (1.09%), Scheduled: 38.00ms (1.46%), Blocked: 0.00ns (0.00%), Output: 2 rows (411B) │ Input avg.: 0.00 rows, Input std.dev.: 1545.96% │ target: blinkit_iceberg_staging:de_etls.merge_query_target_table$data@2591632568254666277 │ merge row column: merge_row │ row id column: field │ redistribution columns: [date, outlet_id] │ data columns: [id, name, is_active, created_at, price, quantity, date, outlet_id] └─ Filter[filterPredicate = (CASE WHEN ($not(is_distinct) AND $not((unique_id IS NULL))) THEN CAST(fail(integer '119', varchar 'One MERGE target table row matched more than one source row') AS boolean) ELSE boolean 'true' END)] │ Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer), case_number:integer, is_distinct:boolean] │ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B} │ CPU: 2.00ms (0.27%), Scheduled: 12.00ms (0.46%), Blocked: 0.00ns (0.00%), Output: 1 row (357B) │ Input avg.: 0.00 rows, Input std.dev.: 1545.96% └─ MarkDistinct[distinct = [unique_id:bigint, case_number:integer], marker = is_distinct] │ Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer), case_number:integer, is_distinct:boolean] │ CPU: 0.00ns (0.00%), Scheduled: 1.00ms (0.04%), Blocked: 0.00ns (0.00%), Output: 1 row (357B) │ Input avg.: 0.00 rows, Input std.dev.: 1545.96% └─ LocalExchange[partitioning = HASH, arguments = [unique_id::bigint, case_number::integer]] │ Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer), case_number:integer] │ Estimates: {rows: 0 (65B), cpu: 65, memory: 0B, network: 0B} │ CPU: 6.00ms (0.82%), Scheduled: 11.00ms (0.42%), Blocked: 12.60s (7.01%), Output: 1 row (355B) │ Input avg.: 0.00 rows, Input std.dev.: 1545.96% └─ RemoteSource[sourceFragmentIds = [4]] Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer), case_number:integer] CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Blocked: 37.99s (21.13%), Output: 1 row (355B) Input avg.: 0.00 rows, Input std.dev.: 1545.96% Fragment 4 [HASH] CPU: 113.30ms, Scheduled: 196.29ms, Blocked 50.57s (Input: 45.84s, Output: 0.00ns), Input: 21 rows (5.70kB); per task: avg.: 1.40 std.dev.: 1.08, Output: 1 row (363B) Amount of input data processed by the workers for this stage might be skewed Output layout: [date, outlet_id, unique_id, field, merge_row, case_number] Output partitioning: HASH [unique_id, case_number] Project[] │ Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer), case_number:integer] │ Estimates: {rows: 0 (65B), cpu: 65, memory: 0B, network: 0B} │ CPU: 16.00ms (2.18%), Scheduled: 19.00ms (0.73%), Blocked: 0.00ns (0.00%), Output: 1 row (363B) │ Input avg.: 0.00 rows, Input std.dev.: 1593.74% │ case_number := merge_row.10 └─ Project[] │ Layout: [date:date, outlet_id:integer, unique_id:bigint, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), merge_row:row(integer, varchar, boolean, timestamp(6), double, bigint, date, integer, boolean, tinyint, integer)] │ Estimates: {rows: 0 (63B), cpu: 63, memory: 0B, network: 0B} │ CPU: 4.00ms (0.55%), Scheduled: 4.00ms (0.15%), Blocked: 0.00ns (0.00%), Output: 1 row (350B) │ Input avg.: 0.00 rows, Input std.dev.: 1593.74% │ merge_row := (CASE WHEN present THEN ROW (id_0, name_1, is_active_2, created_at_3, price_4, quantity_5, date_6, outlet_id_7, $not((present IS NULL)), tinyint '3', integer '0') WHEN (present IS NULL) THEN ROW (id_0, name_1, is_active_2, created_at_3, price_4, quantity_5, date_6, outlet_id_7, $not((present IS NULL)), tinyint '1', integer '1') ELSE ROW (null::integer, null::varchar, null::boolean, null::timestamp(6), null::double, null::bigint, null::date, null::integer, $not((present IS NULL)), tinyint '-1', integer '-1') END) └─ RightJoin[criteria = (id = id_0) AND (outlet_id = outlet_id_7) AND (date = date_6), distribution = PARTITIONED] │ Layout: [date:date, outlet_id:integer, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), unique_id:bigint, present:boolean, id_0:integer, name_1:varchar, is_active_2:boolean, created_at_3:timestamp(6), price_4:double, quantity_5:bigint, date_6:date, outlet_id_7:integer] │ Estimates: {rows: 0 (110B), cpu: 2.00k, memory: 151B, network: 0B} │ CPU: 16.00ms (2.18%), Scheduled: 17.00ms (0.66%), Blocked: 1.89s (1.05%), Output: 1 row (342B) │ Left (probe) Input avg.: 0.08 rows, Input std.dev.: 331.66% │ Right (build) Input avg.: 0.07 rows, Input std.dev.: 374.17% │ Distribution: PARTITIONED │ dynamicFilterAssignments = {id_0 -> #df_950, outlet_id_7 -> #df_951, date_6 -> #df_952} ├─ RemoteSource[sourceFragmentIds = [5]] │ Layout: [id:integer, date:date, outlet_id:integer, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), unique_id:bigint, present:boolean] │ CPU: 1.00ms (0.14%), Scheduled: 1.00ms (0.04%), Blocked: 36.04s (20.04%), Output: 20 rows (5.64kB) │ Input avg.: 0.08 rows, Input std.dev.: 331.66% └─ LocalExchange[partitioning = SINGLE] │ Layout: [id_0:integer, name_1:varchar, is_active_2:boolean, created_at_3:timestamp(6), price_4:double, quantity_5:bigint, date_6:date, outlet_id_7:integer] │ Estimates: {rows: 1 (151B), cpu: 0, memory: 0B, network: 0B} │ CPU: 1.00ms (0.14%), Scheduled: 3.00ms (0.12%), Blocked: 670.00ms (0.37%), Output: 1 row (58B) │ Input avg.: 0.00 rows, Input std.dev.: 1545.96% └─ RemoteSource[sourceFragmentIds = [6]] Layout: [id_0:integer, name_1:varchar, is_active_2:boolean, created_at_3:timestamp(6), price_4:double, quantity_5:bigint, date_6:date, outlet_id_7:integer] CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Blocked: 9.80s (5.45%), Output: 1 row (58B) Input avg.: 0.00 rows, Input std.dev.: 1545.96% Fragment 5 [SOURCE] CPU: 278.08ms, Scheduled: 1.25s, Blocked 0.00ns (Input: 0.00ns, Output: 0.00ns), Input: 20 rows (5.43kB); per task: avg.: 1.33 std.dev.: 0.47, Output: 20 rows (5.72kB) Output layout: [id, date, outlet_id, field, unique_id, present] Output partitioning: HASH [id, outlet_id, date] Project[] │ Layout: [id:integer, date:date, outlet_id:integer, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), unique_id:bigint, present:boolean] │ Estimates: {rows: 22 (1.74kB), cpu: 1.74k, memory: 0B, network: 0B} │ CPU: 2.00ms (0.27%), Scheduled: 2.00ms (0.08%), Blocked: 0.00ns (0.00%), Output: 20 rows (5.72kB) │ Input avg.: 0.91 rows, Input std.dev.: 31.62% │ present := boolean 'true' └─ AssignUniqueId[] │ Layout: [id:integer, date:date, outlet_id:integer, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar), unique_id:bigint] │ Estimates: {rows: 22 (1.70kB), cpu: 198, memory: 0B, network: 0B} │ CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Blocked: 0.00ns (0.00%), Output: 20 rows (5.61kB) │ Input avg.: 0.91 rows, Input std.dev.: 31.62% └─ ScanFilter[table = blinkit_iceberg_staging:de_etls.merge_query_target_table$data@2591632568254666277 constraint on [date], dynamicFilters = {id = #df_950, outlet_id = #df_951, date = #df_952}] Layout: [id:integer, date:date, outlet_id:integer, field:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar)] Estimates: {rows: 22 (1.50kB), cpu: 1.50k, memory: 0B, network: 0B}/{rows: 22 (1.50kB), cpu: 1.50k, memory: 0B, network: 0B} CPU: 267.00ms (36.43%), Scheduled: 1.24s (47.63%), Blocked: 0.00ns (0.00%), Output: 20 rows (5.43kB) Input avg.: 0.91 rows, Input std.dev.: 31.62% id := 1:id:integer date := 7:date:date :: [(, )] field := -2147483647:$row_id:row(_file varchar, _pos bigint, partition_spec_id integer, partition_data varchar) outlet_id := 8:outlet_id:integer Input: 20 rows (5.43kB), Filtered: 0.00%, Physical input: 23.43kB, Physical input time: 1.07s Dynamic filters: - df_950, [ SortedRangeSet[type=integer, ranges=1, {[1]}] ], collection time=93.42ms - df_951, [ SortedRangeSet[type=integer, ranges=1, {[101]}] ], collection time=92.81ms - df_952, [ SortedRangeSet[type=date, ranges=1, {[2023-09-01]}] ], collection time=93.36ms Fragment 6 [SOURCE] CPU: 10.26ms, Scheduled: 48.77ms, Blocked 0.00ns (Input: 0.00ns, Output: 0.00ns), Input: 1 row (58B); per task: avg.: 1.00 std.dev.: 0.00, Output: 1 row (58B) Output layout: [id_0, name_1, is_active_2, created_at_3, price_4, quantity_5, date_6, outlet_id_7] Output partitioning: HASH [id_0, outlet_id_7, date_6] TableScan[table = blinkit_iceberg_staging:de_etls.merge_query_source_table$data@8453901305866021703 constraint on [date]] Layout: [id_0:integer, name_1:varchar, is_active_2:boolean, created_at_3:timestamp(6), price_4:double, quantity_5:bigint, date_6:date, outlet_id_7:integer] Estimates: {rows: 1 (151B), cpu: 151, memory: 0B, network: 0B} CPU: 10.00ms (1.36%), Scheduled: 48.00ms (1.85%), Blocked: 0.00ns (0.00%), Output: 1 row (58B) Input avg.: 1.00 rows, Input std.dev.: 0.00% name_1 := 2:name:varchar id_0 := 1:id:integer outlet_id_7 := 8:outlet_id:integer created_at_3 := 4:created_at:timestamp(6) is_active_2 := 3:is_active:boolean quantity_5 := 6:quantity:bigint price_4 := 5:price:double date_6 := 7:date:date :: [(, )] Input: 1 row (58B), Physical input: 1.06kB, Physical input time: 43.94ms

[ratuldawar11]

Hey @raunaqmorarka any follow up on this, would be really appreciated !

[raunaqmorarka]

For small data sets like that the query will finish too quickly for dynamic filtering to have an effect.
Try setting iceberg.dynamic-filtering.wait-timeout=10s

[ratuldawar11]

Thanks @raunaqmorarka , it was set to zero as default which was lead to target table getting scanned completely as no dynamic filters were being applied. Thanks for the support. 

[ratuldawar11]

Just out of curiosity, @raunaqmorarka why is it set to zero, this means that dynamic filtering would never take place according to my understanding.

[raunaqmorarka]

For large tables this won't matter much as we will apply the dynamic filters when they become available. We don't wait by default as it would slow down some queries where using the dynamic filter wouldn't provide much benefit.

[ratuldawar11]

Thanks for the reply ! really appreciated