Merge pull request #11162 from azat/data-skip-index-merging-params-fix

alexey-milovidov · KochetovNicolai · commit 459559b5c9e3 · 2020-06-08T15:39:04.000+03:00
[RFC] Fix data skipping indexes for columns with additional actions during merge (cherry picked from commit 8accde7)
diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp
@@ -406,6 +406,7 @@ void MergeTreeData::setProperties(const StorageInMemoryMetadata & metadata, bool
     }
 
     ASTPtr skip_indices_with_primary_key_expr_list = new_primary_key_expr_list->clone();
+    ASTPtr skip_indices_expr_list = new_primary_key_expr_list->clone();
     ASTPtr skip_indices_with_sorting_key_expr_list = new_sorting_key_expr_list->clone();
 
     MergeTreeIndices new_indices;
@@ -435,6 +436,7 @@ void MergeTreeData::setProperties(const StorageInMemoryMetadata & metadata, bool
             {
                 skip_indices_with_primary_key_expr_list->children.push_back(expr->clone());
                 skip_indices_with_sorting_key_expr_list->children.push_back(expr->clone());
+                skip_indices_expr_list->children.push_back(expr->clone());
             }
 
             indices_names.insert(new_indices.back()->name);
@@ -445,7 +447,12 @@ void MergeTreeData::setProperties(const StorageInMemoryMetadata & metadata, bool
     auto new_indices_with_primary_key_expr = ExpressionAnalyzer(
             skip_indices_with_primary_key_expr_list, syntax_primary, global_context).getActions(false);
 
-    auto syntax_sorting = SyntaxAnalyzer(global_context, {}).analyze(
+    auto syntax_indices = SyntaxAnalyzer(global_context, {}).analyze(
+            skip_indices_with_primary_key_expr_list, all_columns);
+    auto new_indices_expr = ExpressionAnalyzer(
+            skip_indices_expr_list, syntax_indices, global_context).getActions(false);
+
+    auto syntax_sorting = SyntaxAnalyzer(global_context).analyze(
             skip_indices_with_sorting_key_expr_list, all_columns);
     auto new_indices_with_sorting_key_expr = ExpressionAnalyzer(
             skip_indices_with_sorting_key_expr_list, syntax_sorting, global_context).getActions(false);
@@ -471,6 +478,7 @@ void MergeTreeData::setProperties(const StorageInMemoryMetadata & metadata, bool
 
         setConstraints(metadata.constraints);
 
+        skip_indices_expr = new_indices_expr;
         primary_key_and_skip_indices_expr = new_indices_with_primary_key_expr;
         sorting_key_and_skip_indices_expr = new_indices_with_sorting_key_expr;
     }
diff --git a/src/Storages/MergeTree/MergeTreeData.h b/src/Storages/MergeTree/MergeTreeData.h
@@ -708,6 +708,7 @@ class MergeTreeData : public IStorage
     /// Secondary (data skipping) indices for MergeTree
     MergeTreeIndices skip_indices;
 
+    ExpressionActionsPtr skip_indices_expr;
     ExpressionActionsPtr primary_key_and_skip_indices_expr;
     ExpressionActionsPtr sorting_key_and_skip_indices_expr;
 
diff --git a/src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp b/src/Storages/MergeTree/MergeTreeDataMergerMutator.cpp
@@ -683,7 +683,7 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataMergerMutator::mergePartsToTempor
             MergeProgressCallback(merge_entry, watch_prev_elapsed, horizontal_stage_progress));
 
         BlockInputStreamPtr stream = std::move(input);
-        if (data.hasPrimaryKey() || data.hasSkipIndices())
+        if (data.hasSortingKey())
             stream = std::make_shared<MaterializingBlockInputStream>(
                     std::make_shared<ExpressionBlockInputStream>(stream, data.sorting_key_and_skip_indices_expr));
 
@@ -756,6 +756,12 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataMergerMutator::mergePartsToTempor
     if (need_remove_expired_values)
         merged_stream = std::make_shared<TTLBlockInputStream>(merged_stream, data, new_data_part, time_of_merge, force_ttl);
 
+    if (data.hasSkipIndices())
+    {
+        merged_stream = std::make_shared<ExpressionBlockInputStream>(merged_stream, data.skip_indices_expr);
+        merged_stream = std::make_shared<MaterializingBlockInputStream>(merged_stream);
+    }
+
     MergedBlockOutputStream to{
         data,
         new_part_tmp_path,
diff --git a/tests/queries/0_stateless/01285_data_skip_index_over_aggregation.reference b/tests/queries/0_stateless/01285_data_skip_index_over_aggregation.reference
@@ -0,0 +1,20 @@
+INSERT
+1	0
+1	1
+1	1
+INSERT
+1	0
+1	1
+1	0
+1	1
+1	2
+1	3
+1	1
+1	1
+1	3
+OPTIMIZE
+1	3
+1	3
+OPTIMIZE
+1	3
+1	3
diff --git a/tests/queries/0_stateless/01285_data_skip_index_over_aggregation.sql b/tests/queries/0_stateless/01285_data_skip_index_over_aggregation.sql
@@ -0,0 +1,34 @@
+DROP TABLE IF EXISTS data_01285;
+
+SET max_threads=1;
+
+
+CREATE TABLE data_01285 (
+    key   Int,
+    value SimpleAggregateFunction(max, Nullable(Int)),
+    INDEX value_idx assumeNotNull(value) TYPE minmax GRANULARITY 1
+)
+ENGINE=AggregatingMergeTree()
+ORDER BY key;
+
+SELECT 'INSERT';
+INSERT INTO data_01285 SELECT 1, number FROM numbers(2);
+SELECT * FROM data_01285;
+SELECT * FROM data_01285 WHERE assumeNotNull(value) = 1;
+SELECT 'INSERT';
+INSERT INTO data_01285 SELECT 1, number FROM numbers(4);
+SELECT * FROM data_01285;
+SELECT * FROM data_01285 WHERE assumeNotNull(value) = 1;
+SELECT * FROM data_01285 WHERE assumeNotNull(value) = 3;
+SELECT 'OPTIMIZE';
+OPTIMIZE TABLE data_01285 FINAL;
+SELECT * FROM data_01285;
+-- before the fix value_idx contains one range {0, 0}
+-- and hence cannot find these record.
+SELECT * FROM data_01285 WHERE assumeNotNull(value) = 3;
+-- one more time just in case
+SELECT 'OPTIMIZE';
+OPTIMIZE TABLE data_01285 FINAL;
+SELECT * FROM data_01285;
+-- and this passes even without fix.
+SELECT * FROM data_01285 WHERE assumeNotNull(value) = 3;

-Original file line number
+Diff line change
@@ @@ -0,0 +1,20 @@ @@
 +INSERT
 +1	0
 +1	1
 +1	1
 +INSERT
 +1	0
 +1	1
 +1	0
 +1	1
 +1	2
 +1	3
 +1	1
 +1	1
 +1	3
 +OPTIMIZE
 +1	3
 +1	3
 +OPTIMIZE
 +1	3
 +1	3