More refactoring.

src/Interpreters/InterpreterInsertQuery.cpp

@@ -304,7 +304,7 @@ BlockIO InterpreterInsertQuery::execute()
 
             /// Actually we don't know structure of input blocks from query/table,
             /// because some clients break insertion protocol (columns != header)
-            out.addSource(std::make_shared<ExpressionTransform>(query_sample_block, adding_missing_defaults_actions));
+            out.addSource(std::make_shared<ConvertingTransform>(query_sample_block, adding_missing_defaults_actions));
 
             /// It's important to squash blocks as early as possible (before other transforms),
             ///  because other transforms may work inefficient if block size is small.

src/Processors/Transforms/ExpressionTransform.cpp

@@ -10,7 +10,7 @@ Block ExpressionTransform::transformHeader(Block header, const ActionsDAG & expr
 
 
 ExpressionTransform::ExpressionTransform(const Block & header_, ExpressionActionsPtr expression_)
-    : ExceptionKeepingTransform(header_, transformHeader(header_, expression_->getActionsDAG()))
+    : ISimpleTransform(header_, transformHeader(header_, expression_->getActionsDAG()), false)
     , expression(std::move(expression_))
 {
 }
@@ -25,4 +25,20 @@ void ExpressionTransform::transform(Chunk & chunk)
     chunk.setColumns(block.getColumns(), num_rows);
 }
 
+ConvertingTransform::ConvertingTransform(const Block & header_, ExpressionActionsPtr expression_)
+    : ExceptionKeepingTransform(header_, ExpressionTransform::transformHeader(header_, expression_->getActionsDAG()))
+    , expression(std::move(expression_))
+{
+}
+
+void ConvertingTransform::transform(Chunk & chunk)
+{
+    size_t num_rows = chunk.getNumRows();
+    auto block = getInputPort().getHeader().cloneWithColumns(chunk.detachColumns());
+
+    expression->execute(block, num_rows);
+
+    chunk.setColumns(block.getColumns(), num_rows);
+}
+
 }

src/Processors/Transforms/ExpressionTransform.h

@@ -1,5 +1,6 @@
 #pragma once
 #include <Processors/Transforms/ExceptionKeepingTransform.h>
+#include <Processors/ISimpleTransform.h>
 
 namespace DB
 {
@@ -14,7 +15,7 @@ class ActionsDAG;
   * For example: hits * 2 + 3, url LIKE '%yandex%'
   * The expression processes each row independently of the others.
   */
-class ExpressionTransform final : public ExceptionKeepingTransform
+class ExpressionTransform final : public ISimpleTransform
 {
 public:
     ExpressionTransform(
@@ -32,4 +33,20 @@ private:
     ExpressionActionsPtr expression;
 };
 
+class ConvertingTransform final : public ExceptionKeepingTransform
+{
+public:
+    ConvertingTransform(
+            const Block & header_,
+            ExpressionActionsPtr expression_);
+
+    String getName() const override { return "ConvertingTransform"; }
+
+protected:
+    void transform(Chunk & chunk) override;
+
+private:
+    ExpressionActionsPtr expression;
+};
+
 }