Merge branch 'master' into less-number-of-threads-in-builder

contrib/capnproto-cmake/CMakeLists.txt

@@ -78,8 +78,9 @@ if (COMPILER_GCC)
         -Wno-deprecated-declarations -Wno-class-memaccess)
 elseif (COMPILER_CLANG)
     set (SUPPRESS_WARNINGS -Wno-non-virtual-dtor -Wno-sign-compare -Wno-strict-aliasing -Wno-deprecated-declarations)
+    set (CAPNP_PRIVATE_CXX_FLAGS -fno-char8_t)
 endif ()
 
-target_compile_options(kj PRIVATE ${SUPPRESS_WARNINGS})
-target_compile_options(capnp PRIVATE ${SUPPRESS_WARNINGS})
-target_compile_options(capnpc PRIVATE ${SUPPRESS_WARNINGS})
+target_compile_options(kj PRIVATE ${SUPPRESS_WARNINGS} ${CAPNP_PRIVATE_CXX_FLAGS})
+target_compile_options(capnp PRIVATE ${SUPPRESS_WARNINGS} ${CAPNP_PRIVATE_CXX_FLAGS})
+target_compile_options(capnpc PRIVATE ${SUPPRESS_WARNINGS} ${CAPNP_PRIVATE_CXX_FLAGS})

src/Core/MySQL/MySQLReplication.cpp

@@ -13,6 +13,7 @@ namespace DB
 namespace ErrorCodes
 {
     extern const int UNKNOWN_EXCEPTION;
+    extern const int LOGICAL_ERROR;
 }
 
 namespace MySQLReplication
@@ -103,19 +104,19 @@ namespace MySQLReplication
             = header.event_size - EVENT_HEADER_LENGTH - 4 - 4 - 1 - 2 - 2 - status_len - schema_len - 1 - CHECKSUM_CRC32_SIGNATURE_LENGTH;
         query.resize(len);
         payload.readStrict(reinterpret_cast<char *>(query.data()), len);
-        if (query.rfind("BEGIN", 0) == 0 || query.rfind("COMMIT") == 0)
+        if (query.starts_with("BEGIN") || query.starts_with("COMMIT"))
         {
             typ = QUERY_EVENT_MULTI_TXN_FLAG;
         }
-        else if (query.rfind("XA", 0) == 0)
+        else if (query.starts_with("XA"))
         {
-            if (query.rfind("XA ROLLBACK", 0) == 0)
-                throw ReplicationError("ParseQueryEvent: Unsupported query event:" + query, ErrorCodes::UNKNOWN_EXCEPTION);
+            if (query.starts_with("XA ROLLBACK"))
+                throw ReplicationError("ParseQueryEvent: Unsupported query event:" + query, ErrorCodes::LOGICAL_ERROR);
             typ = QUERY_EVENT_XA;
         }
-        else if (query.rfind("SAVEPOINT", 0) == 0)
+        else if (query.starts_with("SAVEPOINT"))
         {
-            throw ReplicationError("ParseQueryEvent: Unsupported query event:" + query, ErrorCodes::UNKNOWN_EXCEPTION);
+            throw ReplicationError("ParseQueryEvent: Unsupported query event:" + query, ErrorCodes::LOGICAL_ERROR);
         }
     }
 

src/Dictionaries/BucketCache.h

@@ -30,13 +30,12 @@ struct Int64Hasher
 };
 
 
-/**
-  * Class for storing cache index.
-  * It consists of two arrays.
-  * The first one is split into buckets (each stores 8 elements (cells)) determined by hash of the element key.
-  * The second one is split into 4bit numbers, which are positions in bucket for next element write
-  * (So cache uses FIFO eviction algorithm inside each bucket).
-  */
+/*
+    Class for storing cache index.
+    It consists of two arrays.
+    The first one is split into buckets (each stores 8 elements (cells)) determined by hash of the element key.
+    The second one is split into 4bit numbers, which are positions in bucket for next element write (So cache uses FIFO eviction algorithm inside each bucket).
+*/
 template <typename K, typename V, typename Hasher, typename Deleter = EmptyDeleter>
 class BucketCacheIndex
 {

src/Interpreters/TranslateQualifiedNamesVisitor.cpp

@@ -18,6 +18,7 @@
 #include <Parsers/ASTLiteral.h>
 #include <Parsers/ASTFunction.h>
 #include <Parsers/ASTColumnsMatcher.h>
+#include <Parsers/ASTColumnsTransformers.h>
 
 
 namespace DB
@@ -135,8 +136,8 @@ void TranslateQualifiedNamesMatcher::visit(ASTFunction & node, const ASTPtr &, D
 
 void TranslateQualifiedNamesMatcher::visit(const ASTQualifiedAsterisk &, const ASTPtr & ast, Data & data)
 {
-    if (ast->children.size() != 1)
-        throw Exception("Logical error: qualified asterisk must have exactly one child", ErrorCodes::LOGICAL_ERROR);
+    if (ast->children.empty())
+        throw Exception("Logical error: qualified asterisk must have children", ErrorCodes::LOGICAL_ERROR);
 
     auto & ident = ast->children[0];
 
@@ -242,6 +243,10 @@ void TranslateQualifiedNamesMatcher::visit(ASTExpressionList & node, const ASTPt
 
                 first_table = false;
             }
+            for (const auto & transformer : asterisk->children)
+            {
+                IASTColumnsTransformer::transform(transformer, node.children);
+            }
         }
         else if (const auto * asterisk_pattern = child->as<ASTColumnsMatcher>())
         {
@@ -258,6 +263,11 @@ void TranslateQualifiedNamesMatcher::visit(ASTExpressionList & node, const ASTPt
 
                 first_table = false;
             }
+            // ColumnsMatcher's transformers start to appear at child 1
+            for (auto it = asterisk_pattern->children.begin() + 1; it != asterisk_pattern->children.end(); ++it)
+            {
+                IASTColumnsTransformer::transform(*it, node.children);
+            }
         }
         else if (const auto * qualified_asterisk = child->as<ASTQualifiedAsterisk>())
         {
@@ -274,6 +284,11 @@ void TranslateQualifiedNamesMatcher::visit(ASTExpressionList & node, const ASTPt
                     break;
                 }
             }
+            // QualifiedAsterisk's transformers start to appear at child 1
+            for (auto it = qualified_asterisk->children.begin() + 1; it != qualified_asterisk->children.end(); ++it)
+            {
+                IASTColumnsTransformer::transform(*it, node.children);
+            }
         }
         else
             node.children.emplace_back(child);

src/Parsers/ASTAsterisk.cpp

@@ -13,9 +13,14 @@ ASTPtr ASTAsterisk::clone() const
 
 void ASTAsterisk::appendColumnName(WriteBuffer & ostr) const { ostr.write('*'); }
 
-void ASTAsterisk::formatImpl(const FormatSettings & settings, FormatState &, FormatStateStacked) const
+void ASTAsterisk::formatImpl(const FormatSettings & settings, FormatState & state, FormatStateStacked frame) const
 {
     settings.ostr << "*";
+    for (const auto & child : children)
+    {
+        settings.ostr << ' ';
+        child->formatImpl(settings, state, frame);
+    }
 }
 
 }

src/Parsers/ASTAsterisk.h

@@ -9,6 +9,9 @@ namespace DB
 struct AsteriskSemantic;
 struct AsteriskSemanticImpl;
 
+/** SELECT * is expanded to all visible columns of the source table.
+  * Optional transformers can be attached to further manipulate these expanded columns.
+  */
 class ASTAsterisk : public IAST
 {
 public:

src/Parsers/ASTColumnsMatcher.cpp

@@ -28,10 +28,15 @@ void ASTColumnsMatcher::updateTreeHashImpl(SipHash & hash_state) const
     IAST::updateTreeHashImpl(hash_state);
 }
 
-void ASTColumnsMatcher::formatImpl(const FormatSettings & settings, FormatState &, FormatStateStacked) const
+void ASTColumnsMatcher::formatImpl(const FormatSettings & settings, FormatState & state, FormatStateStacked frame) const
 {
     settings.ostr << (settings.hilite ? hilite_keyword : "") << "COLUMNS" << (settings.hilite ? hilite_none : "") << "("
                   << quoteString(original_pattern) << ")";
+    for (ASTs::const_iterator it = children.begin() + 1; it != children.end(); ++it)
+    {
+        settings.ostr << ' ';
+        (*it)->formatImpl(settings, state, frame);
+    }
 }
 
 void ASTColumnsMatcher::setPattern(String pattern)

src/Parsers/ASTColumnsMatcher.h

@@ -23,6 +23,7 @@ struct AsteriskSemanticImpl;
 
 
 /** SELECT COLUMNS('regexp') is expanded to multiple columns like * (asterisk).
+  * Optional transformers can be attached to further manipulate these expanded columns.
   */
 class ASTColumnsMatcher : public IAST
 {

src/Parsers/ASTColumnsTransformers.cpp

@@ -0,0 +1,159 @@
+#include <map>
+#include "ASTColumnsTransformers.h"
+#include <IO/WriteHelpers.h>
+#include <Parsers/ASTFunction.h>
+#include <Parsers/ASTIdentifier.h>
+#include <Common/SipHash.h>
+#include <Common/quoteString.h>
+
+
+namespace DB
+{
+namespace ErrorCodes
+{
+    extern const int ILLEGAL_TYPE_OF_ARGUMENT;
+}
+
+void IASTColumnsTransformer::transform(const ASTPtr & transformer, ASTs & nodes)
+{
+    if (const auto * apply = transformer->as<ASTColumnsApplyTransformer>())
+    {
+        apply->transform(nodes);
+    }
+    else if (const auto * except = transformer->as<ASTColumnsExceptTransformer>())
+    {
+        except->transform(nodes);
+    }
+    else if (const auto * replace = transformer->as<ASTColumnsReplaceTransformer>())
+    {
+        replace->transform(nodes);
+    }
+}
+
+void ASTColumnsApplyTransformer::formatImpl(const FormatSettings & settings, FormatState &, FormatStateStacked) const
+{
+    settings.ostr << (settings.hilite ? hilite_keyword : "") << "APPLY" << (settings.hilite ? hilite_none : "") << "(" << func_name << ")";
+}
+
+void ASTColumnsApplyTransformer::transform(ASTs & nodes) const
+{
+    for (auto & column : nodes)
+    {
+        column = makeASTFunction(func_name, column);
+    }
+}
+
+void ASTColumnsExceptTransformer::formatImpl(const FormatSettings & settings, FormatState & state, FormatStateStacked frame) const
+{
+    settings.ostr << (settings.hilite ? hilite_keyword : "") << "EXCEPT" << (settings.hilite ? hilite_none : "") << "(";
+
+    for (ASTs::const_iterator it = children.begin(); it != children.end(); ++it)
+    {
+        if (it != children.begin())
+        {
+            settings.ostr << ", ";
+        }
+        (*it)->formatImpl(settings, state, frame);
+    }
+
+    settings.ostr << ")";
+}
+
+void ASTColumnsExceptTransformer::transform(ASTs & nodes) const
+{
+    nodes.erase(
+        std::remove_if(
+            nodes.begin(),
+            nodes.end(),
+            [this](const ASTPtr & node_child)
+            {
+                if (const auto * id = node_child->as<ASTIdentifier>())
+                {
+                    for (const auto & except_child : children)
+                    {
+                        if (except_child->as<const ASTIdentifier &>().name == id->shortName())
+                            return true;
+                    }
+                }
+                return false;
+            }),
+        nodes.end());
+}
+
+void ASTColumnsReplaceTransformer::Replacement::formatImpl(
+    const FormatSettings & settings, FormatState & state, FormatStateStacked frame) const
+{
+    expr->formatImpl(settings, state, frame);
+    settings.ostr << (settings.hilite ? hilite_keyword : "") << " AS " << (settings.hilite ? hilite_none : "") << name;
+}
+
+void ASTColumnsReplaceTransformer::formatImpl(const FormatSettings & settings, FormatState & state, FormatStateStacked frame) const
+{
+    settings.ostr << (settings.hilite ? hilite_keyword : "") << "REPLACE" << (settings.hilite ? hilite_none : "") << "(";
+
+    for (ASTs::const_iterator it = children.begin(); it != children.end(); ++it)
+    {
+        if (it != children.begin())
+        {
+            settings.ostr << ", ";
+        }
+        (*it)->formatImpl(settings, state, frame);
+    }
+
+    settings.ostr << ")";
+}
+
+void ASTColumnsReplaceTransformer::replaceChildren(ASTPtr & node, const ASTPtr & replacement, const String & name)
+{
+    for (auto & child : node->children)
+    {
+        if (const auto * id = child->as<ASTIdentifier>())
+        {
+            if (id->shortName() == name)
+                child = replacement;
+        }
+        else
+            replaceChildren(child, replacement, name);
+    }
+}
+
+void ASTColumnsReplaceTransformer::transform(ASTs & nodes) const
+{
+    std::map<String, ASTPtr> replace_map;
+    for (const auto & replace_child : children)
+    {
+        auto & replacement = replace_child->as<Replacement &>();
+        if (replace_map.find(replacement.name) != replace_map.end())
+            throw Exception(
+                "Expressions in columns transformer REPLACE should not contain the same replacement more than once",
+                ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
+        replace_map.emplace(replacement.name, replacement.expr);
+    }
+
+    for (auto & column : nodes)
+    {
+        if (const auto * id = column->as<ASTIdentifier>())
+        {
+            auto replace_it = replace_map.find(id->shortName());
+            if (replace_it != replace_map.end())
+            {
+                column = replace_it->second;
+                column->setAlias(replace_it->first);
+            }
+        }
+        else if (auto * ast_with_alias = dynamic_cast<ASTWithAlias *>(column.get()))
+        {
+            auto replace_it = replace_map.find(ast_with_alias->alias);
+            if (replace_it != replace_map.end())
+            {
+                auto new_ast = replace_it->second->clone();
+                ast_with_alias->alias = ""; // remove the old alias as it's useless after replace transformation
+                replaceChildren(new_ast, column, replace_it->first);
+                column = new_ast;
+                column->setAlias(replace_it->first);
+            }
+        }
+    }
+}
+
+}

src/Parsers/ASTColumnsTransformers.h

@@ -0,0 +1,85 @@
+#pragma once
+
+#include <Parsers/IAST.h>
+
+namespace DB
+{
+class IASTColumnsTransformer : public IAST
+{
+public:
+    virtual void transform(ASTs & nodes) const = 0;
+    static void transform(const ASTPtr & transformer, ASTs & nodes);
+};
+
+class ASTColumnsApplyTransformer : public IASTColumnsTransformer
+{
+public:
+    String getID(char) const override { return "ColumnsApplyTransformer"; }
+    ASTPtr clone() const override
+    {
+        auto res = std::make_shared<ASTColumnsApplyTransformer>(*this);
+        return res;
+    }
+    void transform(ASTs & nodes) const override;
+    String func_name;
+
+protected:
+    void formatImpl(const FormatSettings & settings, FormatState &, FormatStateStacked) const override;
+};
+
+class ASTColumnsExceptTransformer : public IASTColumnsTransformer
+{
+public:
+    String getID(char) const override { return "ColumnsExceptTransformer"; }
+    ASTPtr clone() const override
+    {
+        auto clone = std::make_shared<ASTColumnsExceptTransformer>(*this);
+        clone->cloneChildren();
+        return clone;
+    }
+    void transform(ASTs & nodes) const override;
+
+protected:
+    void formatImpl(const FormatSettings & settings, FormatState &, FormatStateStacked) const override;
+};
+
+class ASTColumnsReplaceTransformer : public IASTColumnsTransformer
+{
+public:
+    class Replacement : public IAST
+    {
+    public:
+        String getID(char) const override { return "ColumnsReplaceTransformer::Replacement"; }
+        ASTPtr clone() const override
+        {
+            auto replacement = std::make_shared<Replacement>(*this);
+            replacement->name = name;
+            replacement->expr = expr->clone();
+            replacement->children.push_back(replacement->expr);
+            return replacement;
+        }
+
+        String name;
+        ASTPtr expr;
+
+    protected:
+        void formatImpl(const FormatSettings & settings, FormatState &, FormatStateStacked) const override;
+    };
+
+    String getID(char) const override { return "ColumnsReplaceTransformer"; }
+    ASTPtr clone() const override
+    {
+        auto clone = std::make_shared<ASTColumnsReplaceTransformer>(*this);
+        clone->cloneChildren();
+        return clone;
+    }
+    void transform(ASTs & nodes) const override;
+
+protected:
+    void formatImpl(const FormatSettings & settings, FormatState &, FormatStateStacked) const override;
+
+private:
+    static void replaceChildren(ASTPtr & node, const ASTPtr & replacement, const String & name);
+};
+
+}

src/Parsers/ASTQualifiedAsterisk.cpp

@@ -16,6 +16,11 @@ void ASTQualifiedAsterisk::formatImpl(const FormatSettings & settings, FormatSta
     const auto & qualifier = children.at(0);
     qualifier->formatImpl(settings, state, frame);
     settings.ostr << ".*";
+    for (ASTs::const_iterator it = children.begin() + 1; it != children.end(); ++it)
+    {
+        settings.ostr << ' ';
+        (*it)->formatImpl(settings, state, frame);
+    }
 }
 
 }

src/Parsers/ASTQualifiedAsterisk.h

@@ -11,6 +11,7 @@ struct AsteriskSemanticImpl;
 
 /** Something like t.*
   * It will have qualifier as its child ASTIdentifier.
+  * Optional transformers can be attached to further manipulate these expanded columns.
   */
 class ASTQualifiedAsterisk : public IAST
 {

src/Parsers/ExpressionElementParsers.cpp

@@ -22,7 +22,9 @@
 #include <Parsers/ASTSelectQuery.h>
 #include <Parsers/ASTSubquery.h>
 #include <Parsers/ASTFunctionWithKeyValueArguments.h>
+#include <Parsers/ASTColumnsTransformers.h>
 
+#include <Parsers/parseIdentifierOrStringLiteral.h>
 #include <Parsers/parseIntervalKind.h>
 #include <Parsers/ExpressionListParsers.h>
 #include <Parsers/ParserSelectWithUnionQuery.h>
@@ -1204,17 +1206,131 @@ bool ParserColumnsMatcher::parseImpl(Pos & pos, ASTPtr & node, Expected & expect
     auto res = std::make_shared<ASTColumnsMatcher>();
     res->setPattern(regex_node->as<ASTLiteral &>().value.get<String>());
     res->children.push_back(regex_node);
+    ParserColumnsTransformers transformers_p;
+    ASTPtr transformer;
+    while (transformers_p.parse(pos, transformer, expected))
+    {
+        res->children.push_back(transformer);
+    }
     node = std::move(res);
     return true;
 }
 
 
-bool ParserAsterisk::parseImpl(Pos & pos, ASTPtr & node, Expected &)
+bool ParserColumnsTransformers::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
+{
+    ParserKeyword apply("APPLY");
+    ParserKeyword except("EXCEPT");
+    ParserKeyword replace("REPLACE");
+    ParserKeyword as("AS");
+
+    if (apply.ignore(pos, expected))
+    {
+        if (pos->type != TokenType::OpeningRoundBracket)
+            return false;
+        ++pos;
+
+        String func_name;
+        if (!parseIdentifierOrStringLiteral(pos, expected, func_name))
+            return false;
+
+        if (pos->type != TokenType::ClosingRoundBracket)
+            return false;
+        ++pos;
+
+        auto res = std::make_shared<ASTColumnsApplyTransformer>();
+        res->func_name = func_name;
+        node = std::move(res);
+        return true;
+    }
+    else if (except.ignore(pos, expected))
+    {
+        if (pos->type != TokenType::OpeningRoundBracket)
+            return false;
+        ++pos;
+
+        ASTs identifiers;
+        auto parse_id = [&identifiers, &pos, &expected]
+        {
+            ASTPtr identifier;
+            if (!ParserIdentifier().parse(pos, identifier, expected))
+                return false;
+
+            identifiers.emplace_back(std::move(identifier));
+            return true;
+        };
+
+        if (!ParserList::parseUtil(pos, expected, parse_id, false))
+            return false;
+
+        if (pos->type != TokenType::ClosingRoundBracket)
+            return false;
+        ++pos;
+
+        auto res = std::make_shared<ASTColumnsExceptTransformer>();
+        res->children = std::move(identifiers);
+        node = std::move(res);
+        return true;
+    }
+    else if (replace.ignore(pos, expected))
+    {
+        if (pos->type != TokenType::OpeningRoundBracket)
+            return false;
+        ++pos;
+
+        ASTs replacements;
+        ParserExpression element_p;
+        ParserIdentifier ident_p;
+        auto parse_id = [&]
+        {
+            ASTPtr expr;
+
+            if (!element_p.parse(pos, expr, expected))
+                return false;
+            if (!as.ignore(pos, expected))
+                return false;
+
+            ASTPtr ident;
+            if (!ident_p.parse(pos, ident, expected))
+                return false;
+
+            auto replacement = std::make_shared<ASTColumnsReplaceTransformer::Replacement>();
+            replacement->name = getIdentifierName(ident);
+            replacement->expr = std::move(expr);
+            replacements.emplace_back(std::move(replacement));
+            return true;
+        };
+
+        if (!ParserList::parseUtil(pos, expected, parse_id, false))
+            return false;
+
+        if (pos->type != TokenType::ClosingRoundBracket)
+            return false;
+        ++pos;
+
+        auto res = std::make_shared<ASTColumnsReplaceTransformer>();
+        res->children = std::move(replacements);
+        node = std::move(res);
+        return true;
+    }
+
+    return false;
+}
+
+
+bool ParserAsterisk::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
 {
     if (pos->type == TokenType::Asterisk)
     {
         ++pos;
-        node = std::make_shared<ASTAsterisk>();
+        auto asterisk = std::make_shared<ASTAsterisk>();
+        ParserColumnsTransformers transformers_p;
+        ASTPtr transformer;
+        while (transformers_p.parse(pos, transformer, expected))
+        {
+            asterisk->children.push_back(transformer);
+        }
+        node = asterisk;
         return true;
     }
     return false;
@@ -1236,6 +1352,12 @@ bool ParserQualifiedAsterisk::parseImpl(Pos & pos, ASTPtr & node, Expected & exp
 
     auto res = std::make_shared<ASTQualifiedAsterisk>();
     res->children.push_back(node);
+    ParserColumnsTransformers transformers_p;
+    ASTPtr transformer;
+    while (transformers_p.parse(pos, transformer, expected))
+    {
+        res->children.push_back(transformer);
+    }
     node = std::move(res);
     return true;
 }

src/Parsers/ExpressionElementParsers.h

@@ -88,6 +88,15 @@ protected:
     bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected) override;
 };
 
+/** *, t.*, db.table.*, COLUMNS('<regular expression>') APPLY(...) or EXCEPT(...) or REPLACE(...)
+  */
+class ParserColumnsTransformers : public IParserBase
+{
+protected:
+    const char * getName() const override { return "COLUMNS transformers"; }
+    bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected) override;
+};
+
 /** A function, for example, f(x, y + 1, g(z)).
   * Or an aggregate function: sum(x + f(y)), corr(x, y). The syntax is the same as the usual function.
   * Or a parametric aggregate function: quantile(0.9)(x + y).

src/Parsers/ya.make

@@ -10,6 +10,7 @@ SRCS(
     ASTAsterisk.cpp
     ASTColumnDeclaration.cpp
     ASTColumnsMatcher.cpp
+    ASTColumnsTransformers.cpp
     ASTConstraintDeclaration.cpp
     ASTCreateQuery.cpp
     ASTCreateQuotaQuery.cpp

tests/integration/test_materialize_mysql_database/materialize_with_ddl.py

@@ -1,4 +1,5 @@
 import time
+import pymysql.cursors
 
 
 def check_query(clickhouse_node, query, result_set, retry_count=3, interval_seconds=3):
@@ -321,3 +322,35 @@ def alter_rename_table_with_materialize_mysql_database(clickhouse_node, mysql_no
 
     clickhouse_node.query("DROP DATABASE test_database")
     mysql_node.query("DROP DATABASE test_database")
+
+def query_event_with_empty_transaction(clickhouse_node, mysql_node, service_name):
+    mysql_node.query("CREATE DATABASE test_database")
+
+    mysql_node.query("RESET MASTER")
+    mysql_node.query("CREATE TABLE test_database.t1(a INT NOT NULL PRIMARY KEY, b VARCHAR(255) DEFAULT 'BEGIN')")
+    mysql_node.query("INSERT INTO test_database.t1(a) VALUES(1)")
+
+    clickhouse_node.query(
+        "CREATE DATABASE test_database ENGINE = MaterializeMySQL('{}:3306', 'test_database', 'root', 'clickhouse')".format(
+            service_name))
+
+    # Reject one empty GTID QUERY event with 'BEGIN' and 'COMMIT'
+    mysql_cursor = mysql_node.alloc_connection().cursor(pymysql.cursors.DictCursor)
+    mysql_cursor.execute("SHOW MASTER STATUS")
+    (uuid, seqs) = mysql_cursor.fetchall()[0]["Executed_Gtid_Set"].split(":")
+    (seq_begin, seq_end) = seqs.split("-")
+    next_gtid = uuid + ":" + str(int(seq_end) + 1)
+    mysql_node.query("SET gtid_next='" + next_gtid + "'")
+    mysql_node.query("BEGIN")
+    mysql_node.query("COMMIT")
+    mysql_node.query("SET gtid_next='AUTOMATIC'")
+
+    # Reject one 'BEGIN' QUERY event and 'COMMIT' XID event.
+    mysql_node.query("/* start */ begin /* end */")
+    mysql_node.query("INSERT INTO test_database.t1(a) VALUES(2)")
+    mysql_node.query("/* start */ commit /* end */")
+
+    check_query(clickhouse_node, "SELECT * FROM test_database.t1 ORDER BY a FORMAT TSV",
+                "1\tBEGIN\n2\tBEGIN\n")
+    clickhouse_node.query("DROP DATABASE test_database")
+    mysql_node.query("DROP DATABASE test_database")

tests/integration/test_materialize_mysql_database/test.py

@@ -120,3 +120,8 @@ def test_materialize_database_ddl_with_mysql_8_0(started_cluster, started_mysql_
     materialize_with_ddl.alter_rename_column_with_materialize_mysql_database(clickhouse_node, started_mysql_8_0, "mysql8_0")
     materialize_with_ddl.alter_modify_column_with_materialize_mysql_database(clickhouse_node, started_mysql_8_0, "mysql8_0")
 
+def test_materialize_database_ddl_with_empty_transaction_5_7(started_cluster, started_mysql_5_7):
+    materialize_with_ddl.query_event_with_empty_transaction(clickhouse_node, started_mysql_5_7, "mysql5_7")
+
+def test_materialize_database_ddl_with_empty_transaction_8_0(started_cluster, started_mysql_8_0):
+    materialize_with_ddl.query_event_with_empty_transaction(clickhouse_node, started_mysql_8_0, "mysql8_0")

tests/queries/0_stateless/01470_columns_transformers.reference

@@ -0,0 +1,63 @@
+220	18	347
+110	9	173.5
+1970-04-11	1970-01-11	1970-11-21
+2	3
+1	2
+18	347
+110	173.5
+1970-04-11	1970-01-11	1970-11-21
+222	18	347
+111	11	173.5
+1970-04-11	1970-01-11	1970-11-21
+SELECT 
+    sum(i),
+    sum(j),
+    sum(k)
+FROM columns_transformers
+SELECT 
+    avg(i),
+    avg(j),
+    avg(k)
+FROM columns_transformers
+SELECT 
+    toDate(any(i)),
+    toDate(any(j)),
+    toDate(any(k))
+FROM columns_transformers AS a
+SELECT 
+    length(toString(j)),
+    length(toString(k))
+FROM columns_transformers
+SELECT 
+    sum(j),
+    sum(k)
+FROM columns_transformers
+SELECT 
+    avg(i),
+    avg(k)
+FROM columns_transformers
+SELECT 
+    toDate(any(i)),
+    toDate(any(j)),
+    toDate(any(k))
+FROM columns_transformers AS a
+SELECT 
+    sum(i + 1 AS i),
+    sum(j),
+    sum(k)
+FROM columns_transformers
+SELECT 
+    avg(i + 1 AS i),
+    avg(j + 2 AS j),
+    avg(k)
+FROM columns_transformers
+SELECT 
+    toDate(any(i)),
+    toDate(any(j)),
+    toDate(any(k))
+FROM columns_transformers AS a
+SELECT 
+    (i + 1) + 1 AS i,
+    j,
+    k
+FROM columns_transformers

tests/queries/0_stateless/01470_columns_transformers.sql

@@ -0,0 +1,36 @@
+DROP TABLE IF EXISTS columns_transformers;
+
+CREATE TABLE columns_transformers (i Int64, j Int16, k Int64) Engine=TinyLog;
+INSERT INTO columns_transformers VALUES (100, 10, 324), (120, 8, 23);
+
+SELECT * APPLY(sum) from columns_transformers;
+SELECT columns_transformers.* APPLY(avg) from columns_transformers;
+SELECT a.* APPLY(toDate) APPLY(any) from columns_transformers a;
+SELECT COLUMNS('[jk]') APPLY(toString) APPLY(length) from columns_transformers;
+
+SELECT * EXCEPT(i) APPLY(sum) from columns_transformers;
+SELECT columns_transformers.* EXCEPT(j) APPLY(avg) from columns_transformers;
+-- EXCEPT after APPLY will not match anything
+SELECT a.* APPLY(toDate) EXCEPT(i, j) APPLY(any) from columns_transformers a;
+
+SELECT * REPLACE(i + 1 AS i) APPLY(sum) from columns_transformers;
+SELECT columns_transformers.* REPLACE(j + 2 AS j, i + 1 AS i) APPLY(avg) from columns_transformers;
+SELECT columns_transformers.* REPLACE(j + 1 AS j, j + 2 AS j) APPLY(avg) from columns_transformers; -- { serverError 43 }
+-- REPLACE after APPLY will not match anything
+SELECT a.* APPLY(toDate) REPLACE(i + 1 AS i) APPLY(any) from columns_transformers a;
+
+EXPLAIN SYNTAX SELECT * APPLY(sum) from columns_transformers;
+EXPLAIN SYNTAX SELECT columns_transformers.* APPLY(avg) from columns_transformers;
+EXPLAIN SYNTAX SELECT a.* APPLY(toDate) APPLY(any) from columns_transformers a;
+EXPLAIN SYNTAX SELECT COLUMNS('[jk]') APPLY(toString) APPLY(length) from columns_transformers;
+EXPLAIN SYNTAX SELECT * EXCEPT(i) APPLY(sum) from columns_transformers;
+EXPLAIN SYNTAX SELECT columns_transformers.* EXCEPT(j) APPLY(avg) from columns_transformers;
+EXPLAIN SYNTAX SELECT a.* APPLY(toDate) EXCEPT(i, j) APPLY(any) from columns_transformers a;
+EXPLAIN SYNTAX SELECT * REPLACE(i + 1 AS i) APPLY(sum) from columns_transformers;
+EXPLAIN SYNTAX SELECT columns_transformers.* REPLACE(j + 2 AS j, i + 1 AS i) APPLY(avg) from columns_transformers;
+EXPLAIN SYNTAX SELECT a.* APPLY(toDate) REPLACE(i + 1 AS i) APPLY(any) from columns_transformers a;
+
+-- Multiple REPLACE in a row
+EXPLAIN SYNTAX SELECT * REPLACE(i + 1 AS i) REPLACE(i + 1 AS i) from columns_transformers;
+
+DROP TABLE columns_transformers;