diff --git a/src/DataStreams/TTLBlockInputStream.cpp b/src/DataStreams/TTLBlockInputStream.cpp
index 692d2c0d5cf..5064e2236f8 100644
--- a/src/DataStreams/TTLBlockInputStream.cpp
+++ b/src/DataStreams/TTLBlockInputStream.cpp
@@ -110,10 +110,11 @@ void TTLBlockInputStream::readSuffixImpl()
 
 void TTLBlockInputStream::removeRowsWithExpiredTableTTL(Block & block)
 {
-    storage.rows_ttl_entry.expression->execute(block);
+    const auto & rows_ttl = storage.getRowsTTL();
+    rows_ttl.expression->execute(block);
 
     const IColumn * ttl_column =
-        block.getByName(storage.rows_ttl_entry.result_column).column.get();
+        block.getByName(rows_ttl.result_column).column.get();
 
     const auto & column_names = header.getNames();
     MutableColumns result_columns;
@@ -152,7 +153,8 @@ void TTLBlockInputStream::removeValuesWithExpiredColumnTTL(Block & block)
     }
 
     std::vector<String> columns_to_remove;
-    for (const auto & [name, ttl_entry] : storage.column_ttl_entries_by_name)
+    const auto & column_ttl_entries_by_name = storage.getColumnTTLs();
+    for (const auto & [name, ttl_entry] : column_ttl_entries_by_name)
     {
         /// If we read not all table columns. E.g. while mutation.
         if (!block.has(name))
@@ -212,8 +214,9 @@ void TTLBlockInputStream::removeValuesWithExpiredColumnTTL(Block & block)
 
 void TTLBlockInputStream::updateMovesTTL(Block & block)
 {
+    const auto & move_ttl_entries = storage.getMoveTTLs();
     std::vector<String> columns_to_remove;
-    for (const auto & ttl_entry : storage.move_ttl_entries)
+    for (const auto & ttl_entry : move_ttl_entries)
     {
         auto & new_ttl_info = new_ttl_infos.moves_ttl[ttl_entry.result_column];
 
diff --git a/src/Storages/IStorage.cpp b/src/Storages/IStorage.cpp
index 64e06a2534e..f8f2599184f 100644
--- a/src/Storages/IStorage.cpp
+++ b/src/Storages/IStorage.cpp
@@ -541,4 +541,50 @@ Names IStorage::getColumnsRequiredForSampling() const
     return {};
 }
 
+
+const StorageMetadataTTLField & IStorage::getRowsTTL() const
+{
+    return table_ttl.rows_ttl;
+}
+
+bool IStorage::hasRowsTTL() const
+{
+    return table_ttl.rows_ttl.expression != nullptr;
+}
+
+const StorageMetadataTTLFields & IStorage::getMoveTTLs() const
+{
+    return table_ttl.move_ttl;
+}
+
+bool IStorage::hasAnyMoveTTL() const
+{
+    return !table_ttl.move_ttl.empty();
+}
+
+const StorageMetadataTTLColumnEntries & IStorage::getColumnTTLs() const
+{
+    return column_ttls_by_name;
+}
+
+void IStorage::setColumnTTLs(const StorageMetadataTTLColumnEntries & column_ttls_by_name_)
+{
+    column_ttls_by_name = column_ttls_by_name_;
+}
+
+const StorageMetadataTableTTL & IStorage::getTableTTLs() const
+{
+    return table_ttl;
+}
+
+void IStorage::setTableTTLs(const StorageMetadataTableTTL & table_ttl_)
+{
+    table_ttl = table_ttl_;
+}
+
+bool IStorage::hasAnyTableTTL() const
+{
+    return hasAnyMoveTTL() || hasRowsTTL();
+}
+
 }
diff --git a/src/Storages/IStorage.h b/src/Storages/IStorage.h
index f707b3a4b5c..416d8ffbbb1 100644
--- a/src/Storages/IStorage.h
+++ b/src/Storages/IStorage.h
@@ -130,10 +130,7 @@ public:
     virtual bool hasEvenlyDistributedRead() const { return false; }
 
     /// Returns true if there is set table TTL, any column TTL or any move TTL.
-    virtual bool hasAnyTTL() const { return false; }
-
-    /// Returns true if there is set TTL for rows.
-    virtual bool hasRowsTTL() const { return false; }
+    virtual bool hasAnyTTL() const { return hasRowsTTL() || hasAnyTableTTL(); }
 
     /// Optional size information of each physical column.
     /// Currently it's only used by the MergeTree family for query optimizations.
@@ -205,6 +202,9 @@ private:
     StorageMetadataKeyField sorting_key;
     StorageMetadataKeyField sampling_key;
 
+    StorageMetadataTTLColumnEntries column_ttls_by_name;
+    StorageMetadataTableTTL table_ttl;
+
 private:
     RWLockImpl::LockHolder tryLockTimed(
         const RWLock & rwlock, RWLockImpl::Type type, const String & query_id, const SettingSeconds & acquire_timeout) const;
@@ -514,6 +514,22 @@ public:
     /// Returns storage policy if storage supports it
     virtual StoragePolicyPtr getStoragePolicy() const { return {}; }
 
+    /// Returns true if there is set TTL for rows.
+    const StorageMetadataTTLField & getRowsTTL() const;
+    bool hasRowsTTL() const;
+
+    const StorageMetadataTTLFields & getMoveTTLs() const;
+    bool hasAnyMoveTTL() const;
+
+    const StorageMetadataTableTTL & getTableTTLs() const;
+    void setTableTTLs(const StorageMetadataTableTTL & table_ttl_);
+    bool hasAnyTableTTL() const;
+
+    const StorageMetadataTTLColumnEntries & getColumnTTLs() const;
+    void setColumnTTLs(const StorageMetadataTTLColumnEntries & column_ttls_by_name_);
+    bool hasAnyColumnTTL() const { return !column_ttls_by_name.empty(); }
+
+
     /// If it is possible to quickly determine exact number of rows in the table at this moment of time, then return it.
     /// Used for:
     /// - Simple count() opimization
diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp
index 142d503569f..a9ba8c871b0 100644
--- a/src/Storages/MergeTree/MergeTreeData.cpp
+++ b/src/Storages/MergeTree/MergeTreeData.cpp
@@ -580,55 +580,17 @@ void MergeTreeData::initPartitionKey(ASTPtr partition_by_ast)
     setPartitionKey(new_partition_key);
 }
 
-namespace
-{
-
-void checkTTLExpression(const ExpressionActionsPtr & ttl_expression, const String & result_column_name)
-{
-    for (const auto & action : ttl_expression->getActions())
-    {
-        if (action.type == ExpressionAction::APPLY_FUNCTION)
-        {
-            IFunctionBase & func = *action.function_base;
-            if (!func.isDeterministic())
-                throw Exception("TTL expression cannot contain non-deterministic functions, "
-                    "but contains function " + func.getName(), ErrorCodes::BAD_ARGUMENTS);
-        }
-    }
-
-    const auto & result_column = ttl_expression->getSampleBlock().getByName(result_column_name);
-
-    if (!typeid_cast<const DataTypeDateTime *>(result_column.type.get())
-        && !typeid_cast<const DataTypeDate *>(result_column.type.get()))
-    {
-        throw Exception("TTL expression result column should have DateTime or Date type, but has "
-            + result_column.type->getName(), ErrorCodes::BAD_TTL_EXPRESSION);
-    }
-}
-
-}
 
 
 void MergeTreeData::setTTLExpressions(const ColumnsDescription & new_columns,
         const ASTPtr & new_ttl_table_ast, bool only_check)
 {
 
-    auto new_column_ttls = new_columns.getColumnTTLs();
+    auto new_column_ttl_asts = new_columns.getColumnTTLs();
 
-    auto create_ttl_entry = [this, &new_columns](ASTPtr ttl_ast)
-    {
-        TTLEntry result;
+    StorageMetadataTTLColumnEntries new_column_ttl_by_name = getColumnTTLs();
 
-        auto syntax_result = SyntaxAnalyzer(global_context).analyze(ttl_ast, new_columns.getAllPhysical());
-        result.expression = ExpressionAnalyzer(ttl_ast, syntax_result, global_context).getActions(false);
-        result.destination_type = DataDestinationType::DELETE;
-        result.result_column = ttl_ast->getColumnName();
-
-        checkTTLExpression(result.expression, result.result_column);
-        return result;
-    };
-
-    if (!new_column_ttls.empty())
+    if (!new_column_ttl_asts.empty())
     {
         NameSet columns_ttl_forbidden;
 
@@ -640,23 +602,24 @@ void MergeTreeData::setTTLExpressions(const ColumnsDescription & new_columns,
             for (const auto & col : getColumnsRequiredForSortingKey())
                 columns_ttl_forbidden.insert(col);
 
-        for (const auto & [name, ast] : new_column_ttls)
+        for (const auto & [name, ast] : new_column_ttl_asts)
         {
             if (columns_ttl_forbidden.count(name))
                 throw Exception("Trying to set TTL for key column " + name, ErrorCodes::ILLEGAL_COLUMN);
             else
             {
-                auto new_ttl_entry = create_ttl_entry(ast);
-                if (!only_check)
-                    column_ttl_entries_by_name[name] = new_ttl_entry;
+                auto new_ttl_entry = StorageMetadataTTLField::getTTLFromAST(ast, new_columns, global_context);
+                new_column_ttl_by_name[name] = new_ttl_entry;
             }
         }
+        if (!only_check)
+            setColumnTTLs(new_column_ttl_by_name);
     }
 
     if (new_ttl_table_ast)
     {
-        std::vector<TTLEntry> update_move_ttl_entries;
-        TTLEntry update_rows_ttl_entry;
+        StorageMetadataTTLFields update_move_ttl_entries;
+        StorageMetadataTTLField update_rows_ttl_entry;
 
         bool seen_delete_ttl = false;
         for (const auto & ttl_element_ptr : new_ttl_table_ast->children)
@@ -672,20 +635,16 @@ void MergeTreeData::setTTLExpressions(const ColumnsDescription & new_columns,
                     throw Exception("More than one DELETE TTL expression is not allowed", ErrorCodes::BAD_TTL_EXPRESSION);
                 }
 
-                auto new_rows_ttl_entry = create_ttl_entry(ttl_element->children[0]);
-                if (!only_check)
-                    update_rows_ttl_entry = new_rows_ttl_entry;
+                auto new_rows_ttl_entry = StorageMetadataTTLField::getTTLFromAST(ttl_element_ptr, new_columns, global_context);
+                update_rows_ttl_entry = new_rows_ttl_entry;
 
                 seen_delete_ttl = true;
             }
             else
             {
-                auto new_ttl_entry = create_ttl_entry(ttl_element->children[0]);
+                auto new_ttl_entry = StorageMetadataTTLField::getTTLFromAST(ttl_element_ptr, new_columns, global_context);
 
-                new_ttl_entry.entry_ast = ttl_element_ptr;
-                new_ttl_entry.destination_type = ttl_element->destination_type;
-                new_ttl_entry.destination_name = ttl_element->destination_name;
-                if (!new_ttl_entry.getDestination(getStoragePolicy()))
+                if (!getDestinationForTTL(new_ttl_entry))
                 {
                     String message;
                     if (new_ttl_entry.destination_type == DataDestinationType::DISK)
@@ -695,18 +654,21 @@ void MergeTreeData::setTTLExpressions(const ColumnsDescription & new_columns,
                     throw Exception(message, ErrorCodes::BAD_TTL_EXPRESSION);
                 }
 
-                if (!only_check)
-                    update_move_ttl_entries.emplace_back(std::move(new_ttl_entry));
+                update_move_ttl_entries.emplace_back(std::move(new_ttl_entry));
             }
         }
 
         if (!only_check)
         {
-            rows_ttl_entry = update_rows_ttl_entry;
-            ttl_table_ast = new_ttl_table_ast;
+            StorageMetadataTableTTL new_table_ttl
+            {
+                .definition_ast = new_ttl_table_ast,
+                .rows_ttl = update_rows_ttl_entry,
+                .move_ttl = update_move_ttl_entries,
+            };
 
             auto move_ttl_entries_lock = std::lock_guard<std::mutex>(move_ttl_entries_mutex);
-            move_ttl_entries = update_move_ttl_entries;
+            setTableTTLs(new_table_ttl);
         }
     }
 }
@@ -2918,7 +2880,7 @@ ReservationPtr MergeTreeData::tryReserveSpacePreferringTTLRules(UInt64 expected_
     auto ttl_entry = selectTTLEntryForTTLInfos(ttl_infos, time_of_move);
     if (ttl_entry)
     {
-        SpacePtr destination_ptr = ttl_entry->getDestination(getStoragePolicy());
+        SpacePtr destination_ptr = getDestinationForTTL(*ttl_entry);
         if (!destination_ptr)
         {
             if (ttl_entry->destination_type == DataDestinationType::VOLUME)
@@ -2952,37 +2914,39 @@ ReservationPtr MergeTreeData::tryReserveSpacePreferringTTLRules(UInt64 expected_
     return reservation;
 }
 
-SpacePtr MergeTreeData::TTLEntry::getDestination(StoragePolicyPtr policy) const
+SpacePtr MergeTreeData::getDestinationForTTL(const StorageMetadataTTLField & ttl) const
 {
-    if (destination_type == DataDestinationType::VOLUME)
-        return policy->getVolumeByName(destination_name);
-    else if (destination_type == DataDestinationType::DISK)
-        return policy->getDiskByName(destination_name);
+    auto policy = getStoragePolicy();
+    if (ttl.destination_type == DataDestinationType::VOLUME)
+        return policy->getVolumeByName(ttl.destination_name);
+    else if (ttl.destination_type == DataDestinationType::DISK)
+        return policy->getDiskByName(ttl.destination_name);
     else
         return {};
 }
 
-bool MergeTreeData::TTLEntry::isPartInDestination(StoragePolicyPtr policy, const IMergeTreeDataPart & part) const
+bool MergeTreeData::isPartInTTLDestination(const StorageMetadataTTLField & ttl, const IMergeTreeDataPart & part) const
 {
-    if (destination_type == DataDestinationType::VOLUME)
+    auto policy = getStoragePolicy();
+    if (ttl.destination_type == DataDestinationType::VOLUME)
     {
-        for (const auto & disk : policy->getVolumeByName(destination_name)->getDisks())
+        for (const auto & disk : policy->getVolumeByName(ttl.destination_name)->getDisks())
             if (disk->getName() == part.volume->getDisk()->getName())
                 return true;
     }
-    else if (destination_type == DataDestinationType::DISK)
-        return policy->getDiskByName(destination_name)->getName() == part.volume->getDisk()->getName();
+    else if (ttl.destination_type == DataDestinationType::DISK)
+        return policy->getDiskByName(ttl.destination_name)->getName() == part.volume->getDisk()->getName();
     return false;
 }
 
-std::optional<MergeTreeData::TTLEntry> MergeTreeData::selectTTLEntryForTTLInfos(
-        const IMergeTreeDataPart::TTLInfos & ttl_infos,
-        time_t time_of_move) const
+std::optional<StorageMetadataTTLField>
+MergeTreeData::selectTTLEntryForTTLInfos(const IMergeTreeDataPart::TTLInfos & ttl_infos, time_t time_of_move) const
 {
     time_t max_max_ttl = 0;
-    std::vector<DB::MergeTreeData::TTLEntry>::const_iterator best_entry_it;
+    std::vector<StorageMetadataTTLField>::const_iterator best_entry_it;
 
     auto lock = std::lock_guard(move_ttl_entries_mutex);
+    const auto & move_ttl_entries = getMoveTTLs();
     for (auto ttl_entry_it = move_ttl_entries.begin(); ttl_entry_it != move_ttl_entries.end(); ++ttl_entry_it)
     {
         auto ttl_info_it = ttl_infos.moves_ttl.find(ttl_entry_it->result_column);
@@ -2996,7 +2960,7 @@ std::optional<MergeTreeData::TTLEntry> MergeTreeData::selectTTLEntryForTTLInfos(
         }
     }
 
-    return max_max_ttl ? *best_entry_it : std::optional<MergeTreeData::TTLEntry>();
+    return max_max_ttl ? *best_entry_it : std::optional<StorageMetadataTTLField>();
 }
 
 MergeTreeData::DataParts MergeTreeData::getDataParts(const DataPartStates & affordable_states) const
@@ -3416,7 +3380,7 @@ bool MergeTreeData::areBackgroundMovesNeeded() const
     if (policy->getVolumes().size() > 1)
         return true;
 
-    return policy->getVolumes().size() == 1 && policy->getVolumes()[0]->getDisks().size() > 1 && !move_ttl_entries.empty();
+    return policy->getVolumes().size() == 1 && policy->getVolumes()[0]->getDisks().size() > 1 && hasAnyMoveTTL();
 }
 
 bool MergeTreeData::movePartsToSpace(const DataPartsVector & parts, SpacePtr space)
@@ -3555,7 +3519,7 @@ ColumnDependencies MergeTreeData::getColumnDependencies(const NameSet & updated_
 
     if (hasRowsTTL())
     {
-        if (add_dependent_columns(rows_ttl_entry.expression, required_ttl_columns))
+        if (add_dependent_columns(getRowsTTL().expression, required_ttl_columns))
         {
             /// Filter all columns, if rows TTL expression have to be recalculated.
             for (const auto & column : getColumns().getAllPhysical())
@@ -3563,13 +3527,13 @@ ColumnDependencies MergeTreeData::getColumnDependencies(const NameSet & updated_
         }
     }
 
-    for (const auto & [name, entry] : column_ttl_entries_by_name)
+    for (const auto & [name, entry] : getColumnTTLs())
     {
         if (add_dependent_columns(entry.expression, required_ttl_columns))
             updated_ttl_columns.insert(name);
     }
 
-    for (const auto & entry : move_ttl_entries)
+    for (const auto & entry : getMoveTTLs())
         add_dependent_columns(entry.expression, required_ttl_columns);
 
     for (const auto & column : indices_columns)
diff --git a/src/Storages/MergeTree/MergeTreeData.h b/src/Storages/MergeTree/MergeTreeData.h
index 2b5b75110f1..b1e14188e21 100644
--- a/src/Storages/MergeTree/MergeTreeData.h
+++ b/src/Storages/MergeTree/MergeTreeData.h
@@ -518,11 +518,6 @@ public:
 
     bool hasSkipIndices() const { return !skip_indices.empty(); }
 
-    bool hasAnyColumnTTL() const { return !column_ttl_entries_by_name.empty(); }
-    bool hasAnyMoveTTL() const { return !move_ttl_entries.empty(); }
-    bool hasRowsTTL() const override { return !rows_ttl_entry.isEmpty(); }
-    bool hasAnyTTL() const override { return hasRowsTTL() || hasAnyMoveTTL() || hasAnyColumnTTL(); }
-
     /// Check that the part is not broken and calculate the checksums for it if they are not present.
     MutableDataPartPtr loadPartAndFixMetadata(const VolumePtr & volume, const String & relative_path) const;
 
@@ -624,6 +619,13 @@ public:
 
     /// Return alter conversions for part which must be applied on fly.
     AlterConversions getAlterConversionsForPart(const MergeTreeDataPartPtr part) const;
+    /// Returns destination disk or volume for the TTL rule according to current
+    /// storage policy
+    SpacePtr getDestinationForTTL(const StorageMetadataTTLField & ttl) const;
+
+    /// Checks if given part already belongs destination disk or volume for the
+    /// TTL rule.
+    bool isPartInTTLDestination(const StorageMetadataTTLField & ttl, const IMergeTreeDataPart & part) const;
 
     MergeTreeDataFormatVersion format_version;
 
@@ -646,39 +648,13 @@ public:
     ExpressionActionsPtr primary_key_and_skip_indices_expr;
     ExpressionActionsPtr sorting_key_and_skip_indices_expr;
 
-    struct TTLEntry
-    {
-        ExpressionActionsPtr expression;
-        String result_column;
+    std::optional<StorageMetadataTTLField> selectTTLEntryForTTLInfos(const IMergeTreeDataPart::TTLInfos & ttl_infos, time_t time_of_move) const;
 
-        /// Name and type of a destination are only valid in table-level context.
-        DataDestinationType destination_type;
-        String destination_name;
-
-        ASTPtr entry_ast;
-
-        /// Returns destination disk or volume for this rule.
-        SpacePtr getDestination(StoragePolicyPtr policy) const;
-
-        /// Checks if given part already belongs destination disk or volume for this rule.
-        bool isPartInDestination(StoragePolicyPtr policy, const IMergeTreeDataPart & part) const;
-
-        bool isEmpty() const { return expression == nullptr; }
-    };
-
-    std::optional<TTLEntry> selectTTLEntryForTTLInfos(const IMergeTreeDataPart::TTLInfos & ttl_infos, time_t time_of_move) const;
-
-    using TTLEntriesByName = std::unordered_map<String, TTLEntry>;
-    TTLEntriesByName column_ttl_entries_by_name;
-
-    TTLEntry rows_ttl_entry;
-
-    /// This mutex is required for background move operations which do not obtain global locks.
+    /// This mutex is required for background move operations which do not
+    /// obtain global locks.
+    /// TODO (alesap) It will be removed after metadata became atomic
     mutable std::mutex move_ttl_entries_mutex;
 
-    /// Vector rw operations have to be done under "move_ttl_entries_mutex".
-    std::vector<TTLEntry> move_ttl_entries;
-
     /// Limiting parallel sends per one table, used in DataPartsExchange
     std::atomic_uint current_table_sends {0};
 
diff --git a/src/Storages/MergeTree/MergeTreeDataWriter.cpp b/src/Storages/MergeTree/MergeTreeDataWriter.cpp
index 48dbf66e72b..0209e6cc23d 100644
--- a/src/Storages/MergeTree/MergeTreeDataWriter.cpp
+++ b/src/Storages/MergeTree/MergeTreeDataWriter.cpp
@@ -77,10 +77,12 @@ void buildScatterSelector(
 }
 
 /// Computes ttls and updates ttl infos
-void updateTTL(const MergeTreeData::TTLEntry & ttl_entry,
+void updateTTL(
+    const StorageMetadataTTLField & ttl_entry,
     IMergeTreeDataPart::TTLInfos & ttl_infos,
     DB::MergeTreeDataPartTTLInfo & ttl_info,
-    Block & block, bool update_part_min_max_ttls)
+    Block & block,
+    bool update_part_min_max_ttls)
 {
     bool remove_column = false;
     if (!block.has(ttl_entry.result_column))
@@ -228,7 +230,8 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataWriter::writeTempPart(BlockWithPa
     size_t expected_size = block.bytes();
 
     DB::IMergeTreeDataPart::TTLInfos move_ttl_infos;
-    for (const auto & ttl_entry : data.move_ttl_entries)
+    const auto & move_ttl_entries = data.getMoveTTLs();
+    for (const auto & ttl_entry : move_ttl_entries)
         updateTTL(ttl_entry, move_ttl_infos, move_ttl_infos.moves_ttl[ttl_entry.result_column], block, false);
 
     NamesAndTypesList columns = data.getColumns().getAllPhysical().filter(block.getNames());
@@ -287,9 +290,9 @@ MergeTreeData::MutableDataPartPtr MergeTreeDataWriter::writeTempPart(BlockWithPa
     }
 
     if (data.hasRowsTTL())
-        updateTTL(data.rows_ttl_entry, new_data_part->ttl_infos, new_data_part->ttl_infos.table_ttl, block, true);
+        updateTTL(data.getRowsTTL(), new_data_part->ttl_infos, new_data_part->ttl_infos.table_ttl, block, true);
 
-    for (const auto & [name, ttl_entry] : data.column_ttl_entries_by_name)
+    for (const auto & [name, ttl_entry] : data.getColumnTTLs())
         updateTTL(ttl_entry, new_data_part->ttl_infos, new_data_part->ttl_infos.columns_ttl[name], block, true);
 
     new_data_part->ttl_infos.update(move_ttl_infos);
diff --git a/src/Storages/MergeTree/MergeTreePartsMover.cpp b/src/Storages/MergeTree/MergeTreePartsMover.cpp
index 1609e1e517e..a2c2099a5f5 100644
--- a/src/Storages/MergeTree/MergeTreePartsMover.cpp
+++ b/src/Storages/MergeTree/MergeTreePartsMover.cpp
@@ -128,14 +128,14 @@ bool MergeTreePartsMover::selectPartsForMove(
         if (!can_move(part, &reason))
             continue;
 
-        auto ttl_entry = part->storage.selectTTLEntryForTTLInfos(part->ttl_infos, time_of_move);
+        auto ttl_entry = data->selectTTLEntryForTTLInfos(part->ttl_infos, time_of_move);
         auto to_insert = need_to_move.find(part->volume->getDisk());
         ReservationPtr reservation;
         if (ttl_entry)
         {
-            auto destination = ttl_entry->getDestination(policy);
-            if (destination && !ttl_entry->isPartInDestination(policy, *part))
-                reservation = part->storage.tryReserveSpace(part->getBytesOnDisk(), ttl_entry->getDestination(policy));
+            auto destination = data->getDestinationForTTL(*ttl_entry);
+            if (destination && !data->isPartInTTLDestination(*ttl_entry, *part))
+                reservation = data->tryReserveSpace(part->getBytesOnDisk(), data->getDestinationForTTL(*ttl_entry));
         }
 
         if (reservation) /// Found reservation by TTL rule.
diff --git a/src/Storages/MergeTree/ReplicatedMergeTreeTableMetadata.cpp b/src/Storages/MergeTree/ReplicatedMergeTreeTableMetadata.cpp
index 5925c4a6935..12bec501181 100644
--- a/src/Storages/MergeTree/ReplicatedMergeTreeTableMetadata.cpp
+++ b/src/Storages/MergeTree/ReplicatedMergeTreeTableMetadata.cpp
@@ -53,7 +53,7 @@ ReplicatedMergeTreeTableMetadata::ReplicatedMergeTreeTableMetadata(const MergeTr
     if (data.format_version >= MERGE_TREE_DATA_MIN_FORMAT_VERSION_WITH_CUSTOM_PARTITIONING)
         partition_key = formattedAST(data.getPartitionKey().expression_list_ast);
 
-    ttl_table = formattedAST(data.ttl_table_ast);
+    ttl_table = formattedAST(data.getTableTTLs().definition_ast);
 
     skip_indices = data.getIndices().toString();
     if (data.canUseAdaptiveGranularity())
diff --git a/src/Storages/MergeTree/StorageFromMergeTreeDataPart.h b/src/Storages/MergeTree/StorageFromMergeTreeDataPart.h
index a25a3a25337..77fdfd60b46 100644
--- a/src/Storages/MergeTree/StorageFromMergeTreeDataPart.h
+++ b/src/Storages/MergeTree/StorageFromMergeTreeDataPart.h
@@ -39,8 +39,7 @@ public:
         return part->storage.mayBenefitFromIndexForIn(left_in_operand, query_context);
     }
 
-    bool hasAnyTTL() const override { return part->storage.hasAnyTTL(); }
-    bool hasRowsTTL() const override { return part->storage.hasRowsTTL(); }
+    //bool hasAnyTTL() const override { return part->storage.hasAnyTTL(); }
 
     ColumnDependencies getColumnDependencies(const NameSet & updated_columns) const override
     {
@@ -65,6 +64,8 @@ protected:
         setColumns(part_->storage.getColumns());
         setIndices(part_->storage.getIndices());
         setSortingKey(part_->storage.getSortingKey());
+        setColumnTTLs(part->storage.getColumnTTLs());
+        setTableTTLs(part->storage.getTableTTLs());
     }
 
 private:
diff --git a/src/Storages/StorageInMemoryMetadata.cpp b/src/Storages/StorageInMemoryMetadata.cpp
index 6c13033da1e..61982dcf46f 100644
--- a/src/Storages/StorageInMemoryMetadata.cpp
+++ b/src/Storages/StorageInMemoryMetadata.cpp
@@ -6,10 +6,21 @@
 #include <Parsers/ASTExpressionList.h>
 #include <Parsers/ASTFunction.h>
 #include <Parsers/queryToString.h>
+#include <Parsers/ASTTTLElement.h>
+#include <Functions/IFunction.h>
+
+#include <DataTypes/DataTypeDateTime.h>
+#include <DataTypes/DataTypeDate.h>
 
 namespace DB
 {
 
+
+namespace ErrorCodes
+{
+    extern const int BAD_TTL_EXPRESSION;
+};
+
 StorageInMemoryMetadata::StorageInMemoryMetadata(
     const ColumnsDescription & columns_,
     const IndicesDescription & indices_,
@@ -138,4 +149,63 @@ StorageMetadataKeyField StorageMetadataKeyField::getKeyFromAST(const ASTPtr & de
     return result;
 }
 
+
+namespace
+{
+
+void checkTTLExpression(const ExpressionActionsPtr & ttl_expression, const String & result_column_name)
+{
+    for (const auto & action : ttl_expression->getActions())
+    {
+        if (action.type == ExpressionAction::APPLY_FUNCTION)
+        {
+            IFunctionBase & func = *action.function_base;
+            if (!func.isDeterministic())
+                throw Exception(
+                    "TTL expression cannot contain non-deterministic functions, "
+                    "but contains function "
+                        + func.getName(),
+                    ErrorCodes::BAD_ARGUMENTS);
+        }
+    }
+
+    const auto & result_column = ttl_expression->getSampleBlock().getByName(result_column_name);
+
+    if (!typeid_cast<const DataTypeDateTime *>(result_column.type.get())
+        && !typeid_cast<const DataTypeDate *>(result_column.type.get()))
+    {
+        throw Exception(
+            "TTL expression result column should have DateTime or Date type, but has " + result_column.type->getName(),
+            ErrorCodes::BAD_TTL_EXPRESSION);
+    }
+}
+
+}
+
+StorageMetadataTTLField StorageMetadataTTLField::getTTLFromAST(const ASTPtr & definition_ast, const ColumnsDescription & columns, const Context & context)
+{
+    StorageMetadataTTLField result;
+    const auto * ttl_element = definition_ast->as<ASTTTLElement>();
+
+    /// First child is expression, like `TTL expr TO DISK`
+    if (ttl_element != nullptr)
+        result.definition_ast = ttl_element->children.front()->clone();
+    else
+        result.definition_ast = definition_ast->clone();
+
+    auto ttl_ast = result.definition_ast->clone();
+    auto syntax_result = SyntaxAnalyzer(context).analyze(ttl_ast, columns.getAllPhysical());
+    result.expression = ExpressionAnalyzer(ttl_ast, syntax_result, context).getActions(false);
+
+    if (ttl_element != nullptr)
+    {
+        result.destination_type = ttl_element->destination_type;
+        result.destination_name = ttl_element->destination_name;
+    }
+
+    result.result_column = ttl_ast->getColumnName();
+
+    checkTTLExpression(result.expression, result.result_column);
+    return result;
+}
 }
diff --git a/src/Storages/StorageInMemoryMetadata.h b/src/Storages/StorageInMemoryMetadata.h
index 3faaed8d680..76720248fbd 100644
--- a/src/Storages/StorageInMemoryMetadata.h
+++ b/src/Storages/StorageInMemoryMetadata.h
@@ -84,6 +84,20 @@ struct StorageMetadataTTLField
     DataDestinationType destination_type;
 
     String destination_name;
+
+    static StorageMetadataTTLField getTTLFromAST(const ASTPtr & definition_ast, const ColumnsDescription & columns, const Context & context);
+};
+
+using StorageMetadataTTLColumnEntries = std::unordered_map<String, StorageMetadataTTLField>;
+using StorageMetadataTTLFields = std::vector<StorageMetadataTTLField>;
+
+struct StorageMetadataTableTTL
+{
+    ASTPtr definition_ast;
+
+    StorageMetadataTTLField rows_ttl;
+
+    StorageMetadataTTLFields move_ttl;
 };
 
 }