Data Skipping Indices fix

dbms/src/Interpreters/InterpreterCreateQuery.cpp

@@ -460,14 +460,21 @@ ColumnsDescription InterpreterCreateQuery::setColumns(
     ASTCreateQuery & create, const Block & as_select_sample, const StoragePtr & as_storage) const
 {
     ColumnsDescription res;
+    IndicesDescription indices;
 
-    if (create.columns_list && create.columns_list->columns)
+    if (create.columns_list)
     {
-        res = getColumnsDescription(*create.columns_list->columns, context);
+        if (create.columns_list->columns)
+            res = getColumnsDescription(*create.columns_list->columns, context);
+        if (create.columns_list->indices)
+            for (const auto & index : create.columns_list->indices->children)
+                indices.indices.push_back(
+                        std::dynamic_pointer_cast<ASTIndexDeclaration>(index->clone()));
     }
     else if (!create.as_table.empty())
     {
         res = as_storage->getColumns();
+        indices = as_storage->getIndicesDescription();
     }
     else if (create.select)
     {
@@ -479,6 +486,8 @@ ColumnsDescription InterpreterCreateQuery::setColumns(
 
     /// Even if query has list of columns, canonicalize it (unfold Nested columns).
     ASTPtr new_columns = formatColumns(res);
+    ASTPtr new_indices = formatIndices(indices);
+
     if (!create.columns_list)
     {
         auto new_columns_list = std::make_shared<ASTColumns>();
@@ -490,6 +499,11 @@ ColumnsDescription InterpreterCreateQuery::setColumns(
     else
         create.columns_list->set(create.columns_list->columns, new_columns);
 
+    if (new_indices && create.columns_list->indices)
+        create.columns_list->replace(create.columns_list->indices, new_indices);
+    else if (new_indices)
+        create.columns_list->set(create.columns_list->indices, new_indices);
+
     /// Check for duplicates
     std::set<String> all_columns;
     auto check_column_already_exists = [&all_columns](const NameAndTypePair & column_name_and_type)

dbms/src/Parsers/ASTIndexDeclaration.h

@@ -3,6 +3,7 @@
 #include <Core/Field.h>
 #include <Core/Types.h>
 #include <Common/FieldVisitors.h>
+#include <IO/WriteHelpers.h>
 #include <Parsers/ASTExpressionList.h>
 #include <Parsers/ASTFunction.h>
 #include <Parsers/IAST.h>
@@ -21,7 +22,7 @@ class ASTIndexDeclaration : public IAST
     String name;
     IAST * expr;
     ASTFunction * type;
-    Field granularity;
+    UInt64 granularity;
 
     /** Get the text that identifies this element. */
     String getID(char) const override { return "Index"; }
@@ -52,7 +53,7 @@ class ASTIndexDeclaration : public IAST
         s.ostr << (s.hilite ? hilite_keyword : "") << " TYPE " << (s.hilite ? hilite_none : "");
         type->formatImpl(s, state, frame);
         s.ostr << (s.hilite ? hilite_keyword : "") << " GRANULARITY " << (s.hilite ? hilite_none : "");
-        s.ostr << applyVisitor(FieldVisitorToString(), granularity);
+        s.ostr << toString(granularity);
     }
 };
 

dbms/src/Parsers/ParserCreateQuery.cpp

@@ -127,7 +127,7 @@ bool ParserIndexDeclaration::parseImpl(Pos & pos, ASTPtr & node, Expected & expe
 
     auto index = std::make_shared<ASTIndexDeclaration>();
     index->name = typeid_cast<const ASTIdentifier &>(*name).name;
-    index->granularity = typeid_cast<const ASTLiteral &>(*granularity).value;
+    index->granularity = typeid_cast<const ASTLiteral &>(*granularity).value.get<UInt64>();
     index->set(index->expr, expr);
     index->set(index->type, type);
     node = index;

dbms/src/Storages/IndicesDescription.h

@@ -6,11 +6,11 @@
 namespace DB
 {
 
-using IndicesAsts = std::vector<std::shared_ptr<ASTIndexDeclaration>>;
+using IndicesASTs = std::vector<std::shared_ptr<ASTIndexDeclaration>>;
 
 struct IndicesDescription
 {
-    IndicesAsts indices;
+    IndicesASTs indices;
 
     IndicesDescription() = default;
 

dbms/src/Storages/MergeTree/MergeTreeData.cpp

@@ -1148,7 +1148,7 @@ void MergeTreeData::checkAlter(const AlterCommands & commands)
 }
 
 void MergeTreeData::createConvertExpression(const DataPartPtr & part, const NamesAndTypesList & old_columns, const NamesAndTypesList & new_columns,
-    const IndicesAsts & old_indices, const IndicesAsts & new_indices, ExpressionActionsPtr & out_expression,
+    const IndicesASTs & old_indices, const IndicesASTs & new_indices, ExpressionActionsPtr & out_expression,
     NameToNameMap & out_rename_map, bool & out_force_update_metadata) const
 {
     out_expression = nullptr;
@@ -1308,7 +1308,7 @@ void MergeTreeData::createConvertExpression(const DataPartPtr & part, const Name
 MergeTreeData::AlterDataPartTransactionPtr MergeTreeData::alterDataPart(
     const DataPartPtr & part,
     const NamesAndTypesList & new_columns,
-    const IndicesAsts & new_indices,
+    const IndicesASTs & new_indices,
     bool skip_sanity_checks)
 {
     ExpressionActionsPtr expression;

dbms/src/Storages/MergeTree/MergeTreeData.h

@@ -490,7 +490,7 @@ class MergeTreeData : public ITableDeclaration
     AlterDataPartTransactionPtr alterDataPart(
         const DataPartPtr & part,
         const NamesAndTypesList & new_columns,
-        const IndicesAsts & new_indices,
+        const IndicesASTs & new_indices,
         bool skip_sanity_checks);
 
     /// Freezes all parts.
@@ -746,7 +746,7 @@ class MergeTreeData : public ITableDeclaration
     /// Files to be deleted are mapped to an empty string in out_rename_map.
     /// If part == nullptr, just checks that all type conversions are possible.
     void createConvertExpression(const DataPartPtr & part, const NamesAndTypesList & old_columns, const NamesAndTypesList & new_columns,
-                                 const IndicesAsts & old_indices, const IndicesAsts & new_indices,
+                                 const IndicesASTs & old_indices, const IndicesASTs & new_indices,
                                  ExpressionActionsPtr & out_expression, NameToNameMap & out_rename_map, bool & out_force_update_metadata) const;
 
     /// Calculates column sizes in compressed form for the current state of data_parts. Call with data_parts mutex locked.

dbms/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp

@@ -520,6 +520,14 @@ BlockInputStreams MergeTreeDataSelectExecutor::readFromParts(
 
     RangesInDataParts parts_with_ranges;
 
+    std::vector<std::pair<MergeTreeIndexPtr, IndexConditionPtr>> useful_indices;
+    for (const auto & index : data.skip_indices)
+    {
+        auto condition = index->createIndexCondition(query_info, context);
+        if (!condition->alwaysUnknownOrTrue())
+            useful_indices.emplace_back(index, condition);
+    }
+
     /// Let's find what range to read from each part.
     size_t sum_marks = 0;
     size_t sum_ranges = 0;
@@ -532,16 +540,9 @@ BlockInputStreams MergeTreeDataSelectExecutor::readFromParts(
         else
             ranges.ranges = MarkRanges{MarkRange{0, part->marks_count}};
 
-        /// It can be done in multiple threads (one thread for each part).
-        /// Maybe it should be moved to BlockInputStream, but it can cause some problems.
-        for (const auto & index : data.skip_indices)
-        {
-            auto condition = index->createIndexCondition(query_info, context);
-            if (!condition->alwaysUnknownOrTrue())
-            {
-                ranges.ranges = filterMarksUsingIndex(index, condition, part, ranges.ranges, settings);
-            }
-        }
+        for (const auto & index_and_condition : useful_indices)
+            ranges.ranges = filterMarksUsingIndex(
+                    index_and_condition.first, index_and_condition.second, part, ranges.ranges, settings);
 
         if (!ranges.ranges.empty())
         {

dbms/src/Storages/MergeTree/MergeTreeIndices.cpp

@@ -26,7 +26,7 @@ void MergeTreeIndexFactory::registerIndex(const std::string &name, Creator creat
                         ErrorCodes::LOGICAL_ERROR);
 }
 
-std::unique_ptr<MergeTreeIndex> MergeTreeIndexFactory::get(
+std::unique_ptr<IMergeTreeIndex> MergeTreeIndexFactory::get(
     const NamesAndTypesList & columns,
     std::shared_ptr<ASTIndexDeclaration> node,
     const Context & context) const
@@ -54,4 +54,21 @@ std::unique_ptr<MergeTreeIndex> MergeTreeIndexFactory::get(
     return it->second(columns, node, context);
 }
 
+
+std::unique_ptr<IMergeTreeIndex> minmaxIndexCreator(
+        const NamesAndTypesList & columns,
+        std::shared_ptr<ASTIndexDeclaration> node,
+        const Context & context);
+
+std::unique_ptr<IMergeTreeIndex> setIndexCreator(
+        const NamesAndTypesList & columns,
+        std::shared_ptr<ASTIndexDeclaration> node,
+        const Context & context);
+
+MergeTreeIndexFactory::MergeTreeIndexFactory()
+{
+    registerIndex("minmax", minmaxIndexCreator);
+    registerIndex("set", setIndexCreator);
+}
+
 }

dbms/src/Storages/MergeTree/MergeTreeIndices.h

@@ -18,48 +18,51 @@ namespace DB
 {
 
 class MergeTreeData;
-class MergeTreeIndex;
+class IMergeTreeIndex;
 
-using MergeTreeIndexPtr = std::shared_ptr<const MergeTreeIndex>;
-using MutableMergeTreeIndexPtr = std::shared_ptr<MergeTreeIndex>;
+using MergeTreeIndexPtr = std::shared_ptr<const IMergeTreeIndex>;
+using MutableMergeTreeIndexPtr = std::shared_ptr<IMergeTreeIndex>;
 
 
-struct MergeTreeIndexGranule
+/// Stores some info about a single block of data.
+struct IMergeTreeIndexGranule
 {
-    virtual ~MergeTreeIndexGranule() = default;
+    virtual ~IMergeTreeIndexGranule() = default;
 
     virtual void serializeBinary(WriteBuffer & ostr) const = 0;
     virtual void deserializeBinary(ReadBuffer & istr) = 0;
 
-    virtual String toString() const = 0;
     virtual bool empty() const = 0;
 
+    /// Updates the stored info using rows of the specified block.
+    /// Reads no more than `limit` rows.
+    /// After finishing updating `pos` will store the position of the first row which was not read.
     virtual void update(const Block & block, size_t * pos, size_t limit) = 0;
 };
 
-
-using MergeTreeIndexGranulePtr = std::shared_ptr<MergeTreeIndexGranule>;
+using MergeTreeIndexGranulePtr = std::shared_ptr<IMergeTreeIndexGranule>;
 using MergeTreeIndexGranules = std::vector<MergeTreeIndexGranulePtr>;
 
+
 /// Condition on the index.
-class IndexCondition
+class IIndexCondition
 {
 public:
-    virtual ~IndexCondition() = default;
+    virtual ~IIndexCondition() = default;
     /// Checks if this index is useful for query.
     virtual bool alwaysUnknownOrTrue() const = 0;
 
     virtual bool mayBeTrueOnGranule(MergeTreeIndexGranulePtr granule) const = 0;
 };
 
-using IndexConditionPtr = std::shared_ptr<IndexCondition>;
+using IndexConditionPtr = std::shared_ptr<IIndexCondition>;
 
 
 /// Structure for storing basic index info like columns, expression, arguments, ...
-class MergeTreeIndex
+class IMergeTreeIndex
 {
 public:
-    MergeTreeIndex(
+    IMergeTreeIndex(
         String name,
         ExpressionActionsPtr expr,
         const Names & columns,
@@ -73,7 +76,7 @@ class MergeTreeIndex
         , header(header)
         , granularity(granularity) {}
 
-    virtual ~MergeTreeIndex() = default;
+    virtual ~IMergeTreeIndex() = default;
 
     /// gets filename without extension
     String getFileName() const { return INDEX_FILE_PREFIX + name; }
@@ -91,7 +94,6 @@ class MergeTreeIndex
     size_t granularity;
 };
 
-
 using MergeTreeIndices = std::vector<MutableMergeTreeIndexPtr>;
 
 
@@ -101,12 +103,12 @@ class MergeTreeIndexFactory : public ext::singleton<MergeTreeIndexFactory>
 
 public:
     using Creator = std::function<
-            std::unique_ptr<MergeTreeIndex>(
+            std::unique_ptr<IMergeTreeIndex>(
                     const NamesAndTypesList & columns,
                     std::shared_ptr<ASTIndexDeclaration> node,
                     const Context & context)>;
 
-    std::unique_ptr<MergeTreeIndex> get(
+    std::unique_ptr<IMergeTreeIndex> get(
         const NamesAndTypesList & columns,
         std::shared_ptr<ASTIndexDeclaration> node,
         const Context & context) const;
@@ -116,7 +118,7 @@ class MergeTreeIndexFactory : public ext::singleton<MergeTreeIndexFactory>
     const auto & getAllIndexes() const { return indexes; }
 
 protected:
-    MergeTreeIndexFactory() = default;
+    MergeTreeIndexFactory();
 
 private:
     using Indexes = std::unordered_map<std::string, Creator>;

dbms/src/Storages/MergeTree/MergeTreeMinMaxIndex.cpp

@@ -17,7 +17,7 @@ namespace ErrorCodes
 
 
 MergeTreeMinMaxGranule::MergeTreeMinMaxGranule(const MergeTreeMinMaxIndex & index)
-    : MergeTreeIndexGranule(), index(index), parallelogram()
+    : IMergeTreeIndexGranule(), index(index), parallelogram()
 {
 }
 
@@ -52,22 +52,13 @@ void MergeTreeMinMaxGranule::deserializeBinary(ReadBuffer & istr)
     }
 }
 
-String MergeTreeMinMaxGranule::toString() const
-{
-    String res = "";
-
-    for (size_t i = 0; i < parallelogram.size(); ++i)
-    {
-        res += "["
-                + applyVisitor(FieldVisitorToString(), parallelogram[i].left) + ", "
-                + applyVisitor(FieldVisitorToString(), parallelogram[i].right) + "]";
-    }
-
-    return res;
-}
-
 void MergeTreeMinMaxGranule::update(const Block & block, size_t * pos, size_t limit)
 {
+    if (*pos >= block.rows())
+        throw Exception(
+                "The provided position is not less than the number of block rows. Position: "
+                + toString(*pos) + ", Block rows: " + toString(block.rows()) + ".", ErrorCodes::LOGICAL_ERROR);
+
     size_t rows_read = std::min(limit, block.rows() - *pos);
 
     for (size_t i = 0; i < index.columns.size(); ++i)
@@ -96,7 +87,7 @@ MinMaxCondition::MinMaxCondition(
     const SelectQueryInfo &query,
     const Context &context,
     const MergeTreeMinMaxIndex &index)
-    : IndexCondition(), index(index), condition(query, context, index.columns, index.expr) {}
+    : IIndexCondition(), index(index), condition(query, context, index.columns, index.expr) {}
 
 bool MinMaxCondition::alwaysUnknownOrTrue() const
 {
@@ -109,7 +100,7 @@ bool MinMaxCondition::mayBeTrueOnGranule(MergeTreeIndexGranulePtr idx_granule) c
         = std::dynamic_pointer_cast<MergeTreeMinMaxGranule>(idx_granule);
     if (!granule)
         throw Exception(
-            "Minmax index condition got wrong granule", ErrorCodes::LOGICAL_ERROR);
+            "Minmax index condition got a granule with the wrong type.", ErrorCodes::LOGICAL_ERROR);
 
     return condition.mayBeTrueInParallelogram(granule->parallelogram, index.data_types);
 }
@@ -123,11 +114,11 @@ MergeTreeIndexGranulePtr MergeTreeMinMaxIndex::createIndexGranule() const
 IndexConditionPtr MergeTreeMinMaxIndex::createIndexCondition(
     const SelectQueryInfo & query, const Context & context) const
 {
-return std::make_shared<MinMaxCondition>(query, context, *this);
+    return std::make_shared<MinMaxCondition>(query, context, *this);
 };
 
 
-std::unique_ptr<MergeTreeIndex> MergeTreeMinMaxIndexCreator(
+std::unique_ptr<IMergeTreeIndex> minmaxIndexCreator(
     const NamesAndTypesList & new_columns,
     std::shared_ptr<ASTIndexDeclaration> node,
     const Context & context)
@@ -158,7 +149,7 @@ std::unique_ptr<MergeTreeIndex> MergeTreeMinMaxIndexCreator(
     }
 
     return std::make_unique<MergeTreeMinMaxIndex>(
-        node->name, std::move(minmax_expr), columns, data_types, sample, node->granularity.get<size_t>());
+        node->name, std::move(minmax_expr), columns, data_types, sample, node->granularity);
 }
 
 }