Repository: ClickHouse/ClickHouse
File: src/Processors/QueryPlan/ReadFromMergeTree.cpp (C++)

Code before:
        NameSet sorting_key_columns_set(sorting_key_columns.begin(), sorting_key_columns.end());

        auto filter_required_columns = query_info.row_level_filter->actions.getRequiredColumnsNames();

        /// Check if ALL required columns are in sorting key
        /// if not, we must apply after FINAL
        /// if they are, we do not change (more optimal)
        bool all_columns_in_sorting_key = true;
        for (const auto & col : filter_required_columns)
        {
            if (!sorting_key_columns_set.contains(col))
            {
                all_columns_in_sorting_key = false;
                break;
            }
        }

        if (!all_columns_in_sorting_key)
        {
            /// save the filters and clear it from query_info so it won't be applied before final
            deferred_row_level_filter = query_info.row_level_filter;
            deferred_prewhere_info = query_info.prewhere_info;
            query_info.row_level_filter = nullptr;
            query_info.prewhere_info = nullptr;

            /// update output_header since row_level_filter and prewhere was part of its calculation
            output_header = std::make_shared<const Block>(MergeTreeSelectProcessor::transformHeader(
                storage_snapshot->getSampleBlockForColumns(all_column_names),
                lazily_read_info,
                nullptr, // <-- row_level_filter
                nullptr)); // <-- prewhere

            LOG_DEBUG(log, "Deferring row policy filter and prewhere to after FINAL");
        }
    }

    result.checkLimits(local_settings, query_info);
    shared_virtual_fields.emplace("_sample_factor", result.sampling.used_sample_factor);

    LOG_DEBUG(
        log,
        "Selected {}/{} parts by partition key, {} parts by primary key, {}/{} marks by primary key, {} marks to read from {} ranges",
        result.parts_before_pk,
        result.total_parts,
        result.selected_parts,
        result.selected_marks_pk,
        result.total_marks_pk,
        result.selected_marks,
        result.selected_ranges);

    // Adding partition info to QueryAccessInfo.

Review comment (bug):
I think need to retain the columns in the reading step. Else, the query will fail when the row level filter is applied after FINAL processing. e.g if the `SELECT * ` in the testcase is replaced with `SELECT x`, error `Column y not found in block ` will be raised.

Code after:
    }

    const auto & local_settings = context->getSettingsRef();

    /// Check if we should apply row policy and prewhere after FINAL instead of during reading
    /// (for correct behavior with ReplacingMergeTree where row policy should not affect which row "wins" during deduplication)
    /// also PREWHERE must always be executed after row policy, so if row policy is deferred, prewhere must be too
    FilterDAGInfoPtr deferred_row_level_filter;
    PrewhereInfoPtr deferred_prewhere_info;

    if (isQueryWithFinal())
    {
        bool defer_row_policy = local_settings[Setting::apply_row_policy_after_final] && query_info.row_level_filter;
        bool defer_prewhere = local_settings[Setting::apply_prewhere_after_final] && query_info.prewhere_info;

        /// If row policy is deferred and uses non-sorting-key columns, prewhere must also be deferred
        /// to maintain correct execution order (row policy before prewhere).
        if (defer_row_policy && query_info.prewhere_info)
        {
            const auto & sorting_key_columns = storage_snapshot->metadata->getSortingKeyColumns();
            NameSet sorting_key_columns_set(sorting_key_columns.begin(), sorting_key_columns.end());

            auto filter_required_columns = query_info.row_level_filter->actions.getRequiredColumnsNames();
            bool all_columns_in_sorting_key = true;
            for (const auto & col : filter_required_columns)
            {
                if (!sorting_key_columns_set.contains(col))
                {
                    all_columns_in_sorting_key = false;
                    break;
                }
            }

            /// If row policy uses non-sorting-key columns, we must defer prewhere too
            if (!all_columns_in_sorting_key)
                defer_prewhere = true;
        }

        if (defer_row_policy || defer_prewhere)
        {
            /// Save the filters and clear them from query_info so they won't be applied during reading
            if (defer_row_policy)
            {
                deferred_row_level_filter = query_info.row_level_filter;
                query_info.row_level_filter = nullptr;
            }

            if (defer_prewhere)
            {
                deferred_prewhere_info = query_info.prewhere_info;
                query_info.prewhere_info = nullptr;