create table data (key Int, value String) engine=Null();
create table dist (key Int, value String) engine=Distributed(test_shard_localhost, currentDatabase(), data);
system stop distributed sends dist;

-- insert into Null() directly, just foward blocks to Null engine.
insert into dist select number, reinterpretAsString(number) from system.numbers limit 1e9 settings prefer_localhost_replica=1;
-- 0 rows in set. Elapsed: 16.888 sec. Processed 1.00 billion rows, 8.00 GB (59.23 million rows/s., 473.85 MB/s.)

-- async INSERT, with storing blocks on disk
insert into dist select number, reinterpretAsString(number) from system.numbers limit 1e9 settings prefer_localhost_replica=0;
-- 0 rows in set. Elapsed: 44.010 sec. Processed 1.00 billion rows, 8.00 GB (22.73 million rows/s., 181.83 MB/s.)

-- distributed_directory_monitor_batch_inserts=0
system flush distributed dist;
-- pending *.bin size is 7.5G
-- 0 rows in set. Elapsed: 22.445 sec.

-- distributed_directory_monitor_batch_inserts=1
system flush distributed dist;
-- first run
-- 0 rows in set. Elapsed: 50.845 sec.;
-- second run
-- 0 rows in set. Elapsed: 39.750 sec.;

-- distributed_directory_monitor_batch_inserts=0
system flush distributed dist;
-- 0 rows in set. Elapsed: 21.459 sec.

-- distributed_directory_monitor_batch_inserts=1
system flush distributed dist;
-- 0 rows in set. Elapsed: 45.488 sec.