[Scaling Writers] Close Idle Writers

[gaurav8297]

Description

In this PR there are following changes:

1. Closing of idle writers in a single writer thread based on some threshold. This change will help reduce memory footprint in long-running partitioned writes. Thus, we can scale more to increase the write performance.
2. Remove max scaling from SkewedPartitionRebalancer.

Additional context and related issues

Benchmarks:

Before:
With memory constraints:

trino:lstec> insert into trips_data_big_gaurav_orc_test_3 select * from trips_data_big_gaurav;

Query 20231114_024556_00005_tc9mv, FAILED, 9 nodes
Splits: 28,887 total, 27,454 done (95.04%)
126:08 [21.9B rows, 968GB] [2.89M rows/s, 131MB/s]

Query aborted by user

peak memory used: ~128GB

Without memory constraints:

trino:lstec> set session max_memory_per_partition_writer='1MB';
trino:lstec> insert into trips_data_big_gaurav_orc_test_3 select * from trips_data_big_gaurav;
INSERT: 27676827574 rows

Query 20231114_052350_00009_tc9mv, FINISHED, 9 nodes
Splits: 35,725 total, 35,725 done (100.00%)
39:14 [27.7B rows, 1.2TB] [11.8M rows/s, 532MB/s]

peak memory used: ~700GB

After:

trino:lstec> insert into trips_data_big_gaurav_orc_test_3 select * from trips_data_big_gaurav;
INSERT: 27676827574 rows

Query 20231113_181736_00006_wkhru, FINISHED, 9 nodes
Splits: 35,739 total, 35,739 done (100.00%)
23:15 [27.7B rows, 1.2TB] [19.8M rows/s, 898MB/s]

peak memory used: ~450GB

Another benchmark with limited max memory per node:

trino:insert_demo> set session query_max_memory_per_node = '8GB';
SET SESSION
trino:insert_demo> insert into lineitem_few_skewed_partitions select orderkey, partkey, suppkey, linenumber, quantity, extendedprice, discount, tax, returnflag, linestatus, c
ommitdate, receiptdate, shipinstruct, comment, shipdate, shipmode from tpch_sf1000_orc_part.lineitem;
INSERT: 5999989709 rows

Query 20231113_193710_00012_wkhru, FINISHED, 9 nodes
Splits: 5,402 total, 5,402 done (100.00%)
7:33 [6B rows, 159GB] [13.3M rows/s, 359MB/s]

Release notes

( ) This is not user-visible or is docs only, and no release notes are required.
( ) Release notes are required. Please propose a release note for me.
( ) Release notes are required, with the following suggested text:

# Section
* Fix some things. ({issue}`issuenumber`)

[sopel39]

some commits lgtm. Consider extracting it as separate PRs

[sopel39]

thanks!

[sopel39]

cc @findepi @alexjo2144

[findepi]

a high level question at this point

Closing of idle writers in a single writer thread based on some threshold. This change will help reduce memory footprint in long-running partitioned writes. Thus, we can scale more to increase the write performance.

This is always a good idea to reclaim unused resources!
What about peak memory needs? Does this PR change number of open partitions per writing node?
how does this relate to iceberg.max-partitions-per-writer = 100?

cc @sopel39 @gaurav8297 @alexjo2144

[gaurav8297]

What about peak memory needs?

Peak memory usage will reduce in the case of idle writers since we are closing the writers and reclaiming their resources early on.

Does this PR change number of open partitions per writing node? how does this relate to iceberg.max-partitions-per-writer = 100?

We do not change anything wrt the maximum number of open partitions (or writers). Currently, when we close the idle writer, we set its index to null in HivePageSink.writers. However, the check for the number of open partitions consider a null writer as an open writer (code). Maybe it's okay for now. But we can improve this by only considering non-null writers during this check.

cc @sopel39 @findepi

[sopel39]

small comments