From 69b2b2a159e18931f3d3f553908a934ce92ad2cd Mon Sep 17 00:00:00 2001
From: Azat Khuzhin <a3at.mail@gmail.com>
Date: Thu, 11 Mar 2021 21:41:27 +0300
Subject: [PATCH] Fix fsync_part_directory for horizontal merge

---
 .../MergeTree/MergeTreeDataMergerMutator.cpp        |  9 +++++----
 .../01643_merge_tree_fsync_smoke.reference          |  2 ++
 .../0_stateless/01643_merge_tree_fsync_smoke.sql    | 13 +++++++++++++
 3 files changed, 20 insertions(+), 4 deletions(-)

diff --git a/src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp b/src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp
index f2f8172837c..b1a3e108013 100644
--- a/src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp
+++ b/src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp
@@ -757,6 +757,8 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataMergerMutator::mergePartsToTempor
     std::unique_ptr<WriteBuffer> rows_sources_write_buf;
     std::optional<ColumnSizeEstimator> column_sizes;
 
+    SyncGuardPtr sync_guard;
+
     if (chosen_merge_algorithm == MergeAlgorithm::Vertical)
     {
         tmp_disk->createDirectories(new_part_tmp_path);
@@ -769,6 +771,9 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataMergerMutator::mergePartsToTempor
             part->accumulateColumnSizes(merged_column_to_size);
 
         column_sizes = ColumnSizeEstimator(merged_column_to_size, merging_column_names, gathering_column_names);
+
+        if (data.getSettings()->fsync_part_directory)
+            sync_guard = disk->getDirectorySyncGuard(new_part_tmp_path);
     }
     else
     {
@@ -778,10 +783,6 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataMergerMutator::mergePartsToTempor
         gathering_column_names.clear();
     }
 
-    SyncGuardPtr sync_guard;
-    if (data.getSettings()->fsync_part_directory)
-        sync_guard = disk->getDirectorySyncGuard(new_part_tmp_path);
-
     /** Read from all parts, merge and write into a new one.
       * In passing, we calculate expression for sorting.
       */
diff --git a/tests/queries/0_stateless/01643_merge_tree_fsync_smoke.reference b/tests/queries/0_stateless/01643_merge_tree_fsync_smoke.reference
index 654db9dbc86..f57d5df6efd 100644
--- a/tests/queries/0_stateless/01643_merge_tree_fsync_smoke.reference
+++ b/tests/queries/0_stateless/01643_merge_tree_fsync_smoke.reference
@@ -10,3 +10,5 @@ wide fsync_after_insert,fsync_part_directory
 1
 memory in_memory_parts_insert_sync
 1
+wide fsync_part_directory,vertical
+1
diff --git a/tests/queries/0_stateless/01643_merge_tree_fsync_smoke.sql b/tests/queries/0_stateless/01643_merge_tree_fsync_smoke.sql
index 21ebb607693..644cf063a33 100644
--- a/tests/queries/0_stateless/01643_merge_tree_fsync_smoke.sql
+++ b/tests/queries/0_stateless/01643_merge_tree_fsync_smoke.sql
@@ -4,34 +4,47 @@ select 'default';
 create table data_01643 (key Int) engine=MergeTree() order by key;
 insert into data_01643 values (1);
 select * from data_01643;
+optimize table data_01643 final;
 drop table data_01643;
 
 select 'compact fsync_after_insert';
 create table data_01643 (key Int) engine=MergeTree() order by key settings min_rows_for_wide_part=2, fsync_after_insert=1;
 insert into data_01643 values (1);
 select * from data_01643;
+optimize table data_01643 final;
 drop table data_01643;
 
 select 'compact fsync_after_insert,fsync_part_directory';
 create table data_01643 (key Int) engine=MergeTree() order by key settings min_rows_for_wide_part=2, fsync_after_insert=1, fsync_part_directory=1;
 insert into data_01643 values (1);
 select * from data_01643;
+optimize table data_01643 final;
 drop table data_01643;
 
 select 'wide fsync_after_insert';
 create table data_01643 (key Int) engine=MergeTree() order by key settings min_bytes_for_wide_part=0, fsync_after_insert=1;
 insert into data_01643 values (1);
 select * from data_01643;
+optimize table data_01643 final;
 drop table data_01643;
 
 select 'wide fsync_after_insert,fsync_part_directory';
 create table data_01643 (key Int) engine=MergeTree() order by key settings min_bytes_for_wide_part=0, fsync_after_insert=1, fsync_part_directory=1;
 insert into data_01643 values (1);
 select * from data_01643;
+optimize table data_01643 final;
 drop table data_01643;
 
 select 'memory in_memory_parts_insert_sync';
 create table data_01643 (key Int) engine=MergeTree() order by key settings min_rows_for_compact_part=2, in_memory_parts_insert_sync=1, fsync_after_insert=1, fsync_part_directory=1;
 insert into data_01643 values (1);
 select * from data_01643;
+optimize table data_01643 final;
+drop table data_01643;
+
+select 'wide fsync_part_directory,vertical';
+create table data_01643 (key Int) engine=MergeTree() order by key settings min_bytes_for_wide_part=0, fsync_part_directory=1, enable_vertical_merge_algorithm=1, vertical_merge_algorithm_min_rows_to_activate=1, vertical_merge_algorithm_min_columns_to_activate=1;
+insert into data_01643 values (1);
+select * from data_01643;
+optimize table data_01643 final;
 drop table data_01643;