Merge pull request ClickHouse#78429 from amosbird/projection-index-1

Projection Index Step 1: Support _part_offset in normal projections

Author: KochetovNicolai

ci/jobs/scripts/check_style/aspell-ignore/en/aspell-dict.txt

@@ -2235,6 +2235,7 @@ mebibytes
 memtable
 memtables
 mergeTreeIndex
+mergeTreeProjection
 mergeable
 mergetree
 messageID

docs/en/sql-reference/statements/alter/projection.md

@@ -137,6 +137,49 @@ As mentioned before, we could review the `system.query_log` table. On the `proje
 SELECT query, projections FROM system.query_log WHERE query_id='<query_id>'
 ```
 
+## Normal projection with `_part_offset` field {#normal-projection-with-part-offset-field}
+
+Creating a table with a normal projection that utilizes the `_part_offset` field:
+
+```sql
+CREATE TABLE events
+(
+    `event_time` DateTime,
+    `event_id` UInt64,
+    `user_id` UInt64,
+    `huge_string` String,
+    PROJECTION order_by_user_id
+    (
+        SELECT
+            _part_offset
+        ORDER BY user_id
+    )
+)
+ENGINE = MergeTree()
+ORDER BY (event_id);
+```
+
+Inserting some sample data:
+
+```sql
+INSERT INTO events SELECT * FROM generateRandom() LIMIT 100000;
+```
+
+### Using `_part_offset` as a secondary index {#normal-projection-secondary-index}
+
+The `_part_offset` field preserves its value through merges and mutations, making it valuable for secondary indexing. We can leverage this in queries:
+
+```sql
+SELECT
+    count()
+FROM events
+WHERE (_part, _part_offset) IN (
+    SELECT _part, _part_offset
+    FROM events
+    WHERE user_id = 42
+)
+```
+
 # Manipulating Projections
 
 The following operations with [projections](/engines/table-engines/mergetree-family/mergetree.md/#projections) are available:

docs/en/sql-reference/table-functions/mergeTreeProjection.md

@@ -0,0 +1,70 @@
+---
+description: 'Represents the contents of some projection in MergeTree tables.
+  It can be used for introspection.'
+sidebar_label: 'mergeTreeProjection'
+sidebar_position: 77
+slug: /sql-reference/table-functions/mergeTreeProjection
+title: 'mergeTreeProjection'
+---
+
+# mergeTreeProjection Table Function
+
+Represents the contents of some projection in MergeTree tables. It can be used for introspection.
+
+```sql
+mergeTreeProjection(database, table, projection)
+```
+
+**Arguments**
+
+- `database`- The database name to read projection from.
+- `table`- The table name to read projection from.
+- `projection` - The projection to read from.
+
+**Returned Value**
+
+A table object with columns provided by given projection.
+
+## Usage Example {#usage-example}
+
+```sql
+CREATE TABLE test
+(
+    `user_id` UInt64,
+    `item_id` UInt64,
+    PROJECTION order_by_item_id
+    (
+        SELECT _part_offset
+        ORDER BY item_id
+    )
+)
+ENGINE = MergeTree
+ORDER BY user_id;
+
+INSERT INTO test SELECT number, 100 - number FROM numbers(5);
+```
+
+```sql
+SELECT *, _part_offset FROM mergeTreeProjection(currentDatabase(), test, order_by_item_id);
+```
+
+```text
+   ┌─item_id─┬─_parent_part_offset─┬─_part_offset─┐
+1. │      96 │                   4 │            0 │
+2. │      97 │                   3 │            1 │
+3. │      98 │                   2 │            2 │
+4. │      99 │                   1 │            3 │
+5. │     100 │                   0 │            4 │
+   └─────────┴─────────────────────┴──────────────┘
+```
+
+```sql
+DESCRIBE mergeTreeProjection(currentDatabase(), test, order_by_item_id) SETTINGS describe_compact_output = 1;
+```
+
+```text
+   ┌─name────────────────┬─type───┐
+1. │ item_id             │ UInt64 │
+2. │ _parent_part_offset │ UInt64 │
+   └─────────────────────┴────────┘
+```

src/Interpreters/MutationsInterpreter.cpp

@@ -1285,6 +1285,8 @@ void MutationsInterpreter::Source::read(
             storage_snapshot,
             part,
             alter_conversions,
+            nullptr,
+            0,
             required_columns,
             nullptr,
             apply_deleted_mask_,

src/Processors/QueryPlan/Optimizations/optimizeUseNormalProjection.cpp

@@ -1,18 +1,15 @@
-#include <Processors/QueryPlan/Optimizations/Optimizations.h>
-#include <Processors/QueryPlan/Optimizations/projectionsCommon.h>
+#include <Core/Settings.h>
 #include <Processors/QueryPlan/ExpressionStep.h>
 #include <Processors/QueryPlan/FilterStep.h>
+#include <Processors/QueryPlan/Optimizations/Optimizations.h>
+#include <Processors/QueryPlan/Optimizations/projectionsCommon.h>
 #include <Processors/QueryPlan/ReadFromMergeTree.h>
-#include <Processors/QueryPlan/UnionStep.h>
 #include <Processors/QueryPlan/ReadFromPreparedSource.h>
+#include <Processors/QueryPlan/UnionStep.h>
 #include <Processors/Sources/NullSource.h>
-#include <Common/logger_useful.h>
-#include <Core/Settings.h>
+#include <Storages/MergeTree/MergeTreeDataSelectExecutor.h>
 #include <Storages/ProjectionsDescription.h>
 #include <Storages/SelectQueryInfo.h>
-#include <Storages/MergeTree/MergeTreeDataSelectExecutor.h>
-
-#include <algorithm>
 
 namespace DB
 {
@@ -72,18 +69,6 @@ static std::optional<ActionsDAG> makeMaterializingDAG(const Block & proj_header,
     return dag;
 }
 
-static bool hasAllRequiredColumns(const ProjectionDescription * projection, const Names & required_columns)
-{
-    for (const auto & col : required_columns)
-    {
-        if (!projection->sample_block.has(col))
-            return false;
-    }
-
-    return true;
-}
-
-
 std::optional<String> optimizeUseNormalProjections(Stack & stack, QueryPlan::Nodes & nodes)
 {
     const auto & frame = stack.back();
@@ -100,8 +85,7 @@ std::optional<String> optimizeUseNormalProjections(Stack & stack, QueryPlan::Nod
     {
         iter = std::next(iter);
 
-        if (!typeid_cast<FilterStep *>(iter->node->step.get()) &&
-            !typeid_cast<ExpressionStep *>(iter->node->step.get()))
+        if (!typeid_cast<FilterStep *>(iter->node->step.get()) && !typeid_cast<ExpressionStep *>(iter->node->step.get()))
             break;
     }
 
@@ -124,7 +108,8 @@ std::optional<String> optimizeUseNormalProjections(Stack & stack, QueryPlan::Nod
     auto it = std::find_if(
         normal_projections.begin(),
         normal_projections.end(),
-        [&](const auto * projection) { return projection->name == context->getSettingsRef()[Setting::preferred_optimize_projection_name].value; });
+        [&](const auto * projection)
+        { return projection->name == context->getSettingsRef()[Setting::preferred_optimize_projection_name].value; });
 
     if (it != normal_projections.end())
     {
@@ -133,20 +118,52 @@ std::optional<String> optimizeUseNormalProjections(Stack & stack, QueryPlan::Nod
         normal_projections.push_back(preferred_projection);
     }
 
+    Names required_columns = reading->getAllColumnNames();
+
+    /// If `with_parent_part_offset` is true and the required columns include `_part_offset`,
+    /// we need to remap it to `_parent_part_offset`. This ensures that the projection's
+    /// ActionsDAG reads from the correct column and generates `_part_offset` in the output.
+    bool with_parent_part_offset = std::any_of(
+        normal_projections.begin(), normal_projections.end(), [](const auto & projection) { return projection->with_parent_part_offset; });
+    bool need_parent_part_offset = false;
+    if (with_parent_part_offset)
+    {
+        for (auto & name : required_columns)
+        {
+            if (name == "_part_offset")
+            {
+                name = "_parent_part_offset";
+                need_parent_part_offset = true;
+            }
+        }
+    }
+
     QueryDAG query;
     {
         auto & child = iter->node->children[iter->next_child - 1];
         if (!query.build(*child))
             return {};
 
+        if (need_parent_part_offset)
+        {
+            ActionsDAG rename_dag;
+            const auto * node = &rename_dag.addInput("_parent_part_offset", std::make_shared<DataTypeUInt64>());
+            node = &rename_dag.addAlias(*node, "_part_offset");
+            rename_dag.getOutputs() = {node};
+
+            if (query.dag)
+                query.dag = ActionsDAG::merge(std::move(rename_dag), *std::move(query.dag));
+            else
+                query.dag = std::move(rename_dag);
+        }
+
         if (query.dag)
             query.dag->removeUnusedActions();
     }
 
     std::list<NormalProjectionCandidate> candidates;
     NormalProjectionCandidate * best_candidate = nullptr;
 
-    const Names & required_columns = reading->getAllColumnNames();
     const auto & query_info = reading->getQueryInfo();
     MergeTreeDataSelectExecutor reader(reading->getMergeTreeData());
 
@@ -168,9 +185,21 @@ std::optional<String> optimizeUseNormalProjections(Stack & stack, QueryPlan::Nod
 
     auto logger = getLogger("optimizeUseNormalProjections");
 
+    auto projection_virtuals = reading->getMergeTreeData().getProjectionVirtualsPtr();
+    auto has_all_required_columns = [&](const ProjectionDescription * projection)
+    {
+        for (const auto & col : required_columns)
+        {
+            if (!projection->sample_block.has(col) && !projection_virtuals->has(col))
+                return false;
+        }
+
+        return true;
+    };
+
     for (const auto * projection : normal_projections)
     {
-        if (!hasAllRequiredColumns(projection, required_columns))
+        if (!has_all_required_columns(projection))
             continue;
 
         auto & candidate = candidates.emplace_back();
@@ -237,7 +266,7 @@ std::optional<String> optimizeUseNormalProjections(Stack & stack, QueryPlan::Nod
     query_info_copy.prewhere_info = nullptr;
 
     auto projection_reading = reader.readFromParts(
-        /*parts=*/ {},
+        /*parts=*/{},
         reading->getMutationsSnapshot()->cloneEmpty(),
         required_columns,
         proj_snapshot,
@@ -280,15 +309,11 @@ std::optional<String> optimizeUseNormalProjections(Stack & stack, QueryPlan::Nod
         if (query.filter_node)
         {
             expr_or_filter_node.step = std::make_unique<FilterStep>(
-                projection_reading_node.step->getOutputHeader(),
-                std::move(*query.dag),
-                query.filter_node->result_name,
-                true);
+                projection_reading_node.step->getOutputHeader(), std::move(*query.dag), query.filter_node->result_name, true);
         }
         else
-            expr_or_filter_node.step = std::make_unique<ExpressionStep>(
-                projection_reading_node.step->getOutputHeader(),
-                std::move(*query.dag));
+            expr_or_filter_node.step
+                = std::make_unique<ExpressionStep>(projection_reading_node.step->getOutputHeader(), std::move(*query.dag));
 
         expr_or_filter_node.children.push_back(&projection_reading_node);
         next_node = &expr_or_filter_node;