Fix Build due to ClickHouse/ClickHouse#80931

Chang chen · Chang chen · commit 02d12f65b0ee · 2025-06-29T14:53:21.000+08:00
diff --git a/cpp-ch/local-engine/Storages/Parquet/VectorizedParquetRecordReader.h b/cpp-ch/local-engine/Storages/Parquet/VectorizedParquetRecordReader.h
@@ -214,9 +214,6 @@ class VectorizedParquetBlockInputFormat final : public DB::IInputFormat
 protected:
     void onCancel() noexcept override { is_stopped = 1; }
 
-    // TODO: create ColumnIndexFilter here, currently disable it now.
-    void setKeyCondition(const std::shared_ptr<const DB::KeyCondition> & key_condition_) override { }
-
 public:
     VectorizedParquetBlockInputFormat(
         DB::ReadBuffer & in_,
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/ExcelTextFormatFile.cpp b/cpp-ch/local-engine/Storages/SubstraitSource/ExcelTextFormatFile.cpp
@@ -62,7 +62,8 @@ bool ExcelTextFormatFile::useThis(const DB::ContextPtr & context)
     return settingsEqual(context->getSettingsRef(), USE_EXCEL_PARSER, "true");
 }
 
-FormatFile::InputFormatPtr ExcelTextFormatFile::createInputFormat(const DB::Block & header)
+FormatFile::InputFormatPtr
+ExcelTextFormatFile::createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & /*filter_actions_dag*/)
 {
     auto read_buffer = read_buffer_builder->build(file_info);
 
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/ExcelTextFormatFile.h b/cpp-ch/local-engine/Storages/SubstraitSource/ExcelTextFormatFile.h
@@ -47,7 +47,8 @@ class ExcelTextFormatFile : public FormatFile
 
     ~ExcelTextFormatFile() override = default;
 
-    FormatFile::InputFormatPtr createInputFormat(const DB::Block & header) override;
+    FormatFile::InputFormatPtr
+    createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag = nullptr) override;
 
     bool supportSplit() const override { return true; }
     String getFileFormat() const override { return "ExcelText"; }
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/FileReader.cpp b/cpp-ch/local-engine/Storages/SubstraitSource/FileReader.cpp
@@ -274,17 +274,12 @@ std::unique_ptr<NormalFileReader> createNormalFileReader(
     const FormatFilePtr & file,
     const DB::Block & to_read_header_,
     const DB::Block & output_header_,
-    const std::shared_ptr<const DB::KeyCondition> & key_condition = nullptr,
+    const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag = nullptr,
     const ColumnIndexFilterPtr & column_index_filter = nullptr)
 {
     file->initialize(column_index_filter);
     auto createInputFormat = [&](const DB::Block & new_read_header_) -> FormatFile::InputFormatPtr
-    {
-        auto input_format = file->createInputFormat(new_read_header_);
-        if (key_condition && input_format)
-            input_format->inputFormat().setKeyCondition(key_condition);
-        return input_format;
-    };
+    { return file->createInputFormat(new_read_header_, filter_actions_dag); };
 
     if (file->getFileInfo().has_iceberg())
         return iceberg::IcebergReader::create(file, to_read_header_, output_header_, createInputFormat);
@@ -316,11 +311,13 @@ std::unique_ptr<NormalFileReader> createNormalFileReader(
     return std::make_unique<NormalFileReader>(file, to_read_header_, output_header_, input_format);
 }
 }
+
+/// TODO Remove ColumnIndexFilterPtr
 std::unique_ptr<BaseReader> BaseReader::create(
     const FormatFilePtr & current_file,
     const DB::Block & readHeader,
     const DB::Block & outputHeader,
-    const std::shared_ptr<const DB::KeyCondition> & key_condition,
+    const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag,
     const ColumnIndexFilterPtr & column_index_filter)
 {
     if (!readHeader)
@@ -335,7 +332,7 @@ std::unique_ptr<BaseReader> BaseReader::create(
         }
     }
 
-    return createNormalFileReader(current_file, readHeader, outputHeader, key_condition, column_index_filter);
+    return createNormalFileReader(current_file, readHeader, outputHeader, filter_actions_dag, column_index_filter);
 }
 
 
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/FileReader.h b/cpp-ch/local-engine/Storages/SubstraitSource/FileReader.h
@@ -77,7 +77,7 @@ class BaseReader
         const FormatFilePtr & current_file,
         const DB::Block & readHeader,
         const DB::Block & outputHeader,
-        const std::shared_ptr<const DB::KeyCondition> & key_condition,
+        const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag,
         const ColumnIndexFilterPtr & column_index_filter);
 };
 
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/FormatFile.h b/cpp-ch/local-engine/Storages/SubstraitSource/FormatFile.h
@@ -30,6 +30,7 @@
 
 namespace DB
 {
+class ActionsDAG;
 namespace ErrorCodes
 {
 extern const int NOT_IMPLEMENTED;
@@ -115,7 +116,9 @@ class FormatFile
     virtual ~FormatFile() = default;
 
     /// Create a new input format for reading this file
-    virtual InputFormatPtr createInputFormat(const DB::Block & header) = 0;
+    virtual InputFormatPtr
+    createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag = nullptr)
+        = 0;
 
     /// Spark would split a large file into small segements and read in different tasks
     /// If this file doesn't support the split feacture, only the task with offset 0 will generate data.
@@ -149,7 +152,6 @@ class FormatFile
     std::map<String, String> partition_values;
     /// partition keys are normalized to lower cases for partition column case-insensitive matching
     std::map<String, String> normalized_partition_values;
-    std::shared_ptr<const DB::KeyCondition> key_condition;
     const FileMetaColumns meta_columns;
 
     /// Currently, it is used to read an iceberg format, and initialized in the constructor of child class
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/Iceberg/IcebergReader.h b/cpp-ch/local-engine/Storages/SubstraitSource/Iceberg/IcebergReader.h
@@ -18,6 +18,12 @@
 
 #include <Storages/SubstraitSource/FileReader.h>
 
+namespace DB
+{
+class ExpressionActions;
+using ExpressionActionsPtr = std::shared_ptr<ExpressionActions>;
+}
+
 namespace local_engine
 {
 class DeltaDVRoaringBitmapArray;
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/JSONFormatFile.cpp b/cpp-ch/local-engine/Storages/SubstraitSource/JSONFormatFile.cpp
@@ -31,7 +31,8 @@ JSONFormatFile::JSONFormatFile(
 {
 }
 
-FormatFile::InputFormatPtr JSONFormatFile::createInputFormat(const DB::Block & header)
+FormatFile::InputFormatPtr
+JSONFormatFile::createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & /*filter_actions_dag*/)
 {
     auto read_buffer = read_buffer_builder->buildWithCompressionWrapper(file_info);
 
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/JSONFormatFile.h b/cpp-ch/local-engine/Storages/SubstraitSource/JSONFormatFile.h
@@ -28,7 +28,8 @@ class JSONFormatFile : public FormatFile
 
     bool supportSplit() const override { return true; }
 
-    FormatFile::InputFormatPtr createInputFormat(const DB::Block & header) override;
+    FormatFile::InputFormatPtr
+    createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag = nullptr) override;
 
     String getFileFormat() const override { return "JSONEachRow"; }
 };
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/ORCFormatFile.cpp b/cpp-ch/local-engine/Storages/SubstraitSource/ORCFormatFile.cpp
@@ -36,7 +36,8 @@ ORCFormatFile::ORCFormatFile(
 {
 }
 
-FormatFile::InputFormatPtr ORCFormatFile::createInputFormat(const DB::Block & header)
+FormatFile::InputFormatPtr
+ORCFormatFile::createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag)
 {
     auto read_buffer = read_buffer_builder->build(file_info);
 
@@ -70,7 +71,8 @@ FormatFile::InputFormatPtr ORCFormatFile::createInputFormat(const DB::Block & he
         format_settings.orc.reader_time_zone_name = mapped_timezone;
     }
     //TODO: support prefetch
-    auto input_format = std::make_shared<DB::NativeORCBlockInputFormat>(*read_buffer, header, format_settings, false, 0);
+    auto parser_group = std::make_shared<DB::FormatParserGroup>(context->getSettingsRef(), 1, filter_actions_dag, context);
+    auto input_format = std::make_shared<DB::NativeORCBlockInputFormat>(*read_buffer, header, format_settings, false, 0, parser_group);
     return std::make_shared<InputFormat>(std::move(read_buffer), input_format);
 }
 
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/ORCFormatFile.h b/cpp-ch/local-engine/Storages/SubstraitSource/ORCFormatFile.h
@@ -44,7 +44,8 @@ class ORCFormatFile : public FormatFile
         DB::ContextPtr context_, const substrait::ReadRel::LocalFiles::FileOrFiles & file_info_, ReadBufferBuilderPtr read_buffer_builder_);
     ~ORCFormatFile() override = default;
 
-    FormatFile::InputFormatPtr createInputFormat(const DB::Block & header) override;
+    FormatFile::InputFormatPtr
+    createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag = nullptr) override;
 
     std::optional<size_t> getTotalRows() override;
 
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.cpp b/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.cpp
@@ -142,7 +142,8 @@ void ParquetFormatFile::initialize(const ColumnIndexFilterPtr & filter)
         file_schema = ParquetMetaBuilder::collectFileSchema(context, *read_buffer_);
 }
 
-FormatFile::InputFormatPtr ParquetFormatFile::createInputFormat(const Block & header)
+FormatFile::InputFormatPtr
+ParquetFormatFile::createInputFormat(const Block & header, const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag)
 {
     assert(read_buffer_);
 
@@ -205,14 +206,8 @@ FormatFile::InputFormatPtr ParquetFormatFile::createInputFormat(const Block & he
             // We need to disable fiter push down and read all row groups, so that we can get correct row index.
             format_settings.parquet.filter_push_down = false;
         }
-
-        auto input = std::make_shared<ParquetBlockInputFormat>(
-            *read_buffer_,
-            read_header,
-            format_settings,
-            settings[Setting::max_parsing_threads],
-            settings[Setting::max_download_threads],
-            8192);
+        auto parser_group = std::make_shared<FormatParserGroup>(context->getSettingsRef(), 1, filter_actions_dag, context);
+        auto input = std::make_shared<ParquetBlockInputFormat>(*read_buffer_, read_header, format_settings, parser_group, 8192);
         return std::make_shared<ParquetInputFormat>(
             std::move(read_buffer_), input, std::move(provider), std::move(read_header), std::move(output_header));
     };
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.h b/cpp-ch/local-engine/Storages/SubstraitSource/ParquetFormatFile.h
@@ -35,7 +35,8 @@ class ParquetFormatFile : public FormatFile
         bool use_local_format_);
     ~ParquetFormatFile() override = default;
 
-    InputFormatPtr createInputFormat(const DB::Block & header) override;
+    InputFormatPtr
+    createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag = nullptr) override;
 
     std::optional<size_t> getTotalRows() override;
 
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/SubstraitFileSource.cpp b/cpp-ch/local-engine/Storages/SubstraitSource/SubstraitFileSource.cpp
@@ -54,7 +54,7 @@ static DB::Block initReadHeader(const DB::Block & block, const FormatFiles & fil
 
 SubstraitFileSource::SubstraitFileSource(
     const DB::ContextPtr & context_, const DB::Block & outputHeader_, const substrait::ReadRel::LocalFiles & file_infos)
-    : DB::SourceWithKeyCondition(BaseReader::buildRowCountHeader(outputHeader_), false)
+    : DB::ISource(BaseReader::buildRowCountHeader(outputHeader_), false)
     , files(initializeFiles(file_infos, context_))
     , outputHeader(outputHeader_)
     , readHeader(initReadHeader(outputHeader, files))
@@ -63,11 +63,11 @@ SubstraitFileSource::SubstraitFileSource(
 
 SubstraitFileSource::~SubstraitFileSource() = default;
 
-void SubstraitFileSource::setKeyCondition(const std::optional<DB::ActionsDAG> & filter_actions_dag, DB::ContextPtr context_)
+void SubstraitFileSource::setKeyCondition(const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag_, DB::ContextPtr context_)
 {
-    setKeyConditionImpl(filter_actions_dag, context_, readHeader);
-    if (filter_actions_dag)
-        column_index_filter = std::make_shared<ColumnIndexFilter>(filter_actions_dag.value(), context_);
+    assert(filter_actions_dag_);
+    filter_actions_dag = filter_actions_dag_;
+    column_index_filter = std::make_shared<ColumnIndexFilter>(*filter_actions_dag, context_);
 }
 
 DB::Chunk SubstraitFileSource::generate()
@@ -105,7 +105,7 @@ bool SubstraitFileSource::tryPrepareReader()
         if (!current_file->supportSplit() && current_file->getStartOffset())
             continue;
 
-        file_reader = BaseReader::create(current_file, readHeader, outputHeader, key_condition, column_index_filter);
+        file_reader = BaseReader::create(current_file, readHeader, outputHeader, filter_actions_dag, column_index_filter);
         if (file_reader)
             return true;
     }
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/SubstraitFileSource.h b/cpp-ch/local-engine/Storages/SubstraitSource/SubstraitFileSource.h
@@ -17,9 +17,15 @@
 #pragma once
 
 #include <memory>
-#include <Processors/SourceWithKeyCondition.h>
+#include <Interpreters/Context_fwd.h>
+#include <Processors/ISource.h>
 #include <substrait/algebra.pb.h>
 
+namespace DB
+{
+class ActionsDAG;
+}
+
 namespace local_engine
 {
 class ColumnIndexFilter;
@@ -29,15 +35,15 @@ class FormatFile;
 using FormatFilePtr = std::shared_ptr<FormatFile>;
 using FormatFiles = std::vector<FormatFilePtr>;
 
-class SubstraitFileSource : public DB::SourceWithKeyCondition
+class SubstraitFileSource : public DB::ISource
 {
 public:
     SubstraitFileSource(const DB::ContextPtr & context_, const DB::Block & header_, const substrait::ReadRel::LocalFiles & file_infos);
     ~SubstraitFileSource() override;
 
     String getName() const override { return "SubstraitFileSource"; }
 
-    void setKeyCondition(const std::optional<DB::ActionsDAG> & filter_actions_dag, DB::ContextPtr context_) override;
+    void setKeyCondition(const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag_, DB::ContextPtr context_);
 
 protected:
     DB::Chunk generate() override;
@@ -54,5 +60,6 @@ class SubstraitFileSource : public DB::SourceWithKeyCondition
 
     std::unique_ptr<BaseReader> file_reader;
     ColumnIndexFilterPtr column_index_filter;
+    std::shared_ptr<const DB::ActionsDAG> filter_actions_dag;
 };
 }
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/SubstraitFileSourceStep.cpp b/cpp-ch/local-engine/Storages/SubstraitSource/SubstraitFileSourceStep.cpp
@@ -15,7 +15,6 @@
  * limitations under the License.
  */
 
-#include <Interpreters/Context_fwd.h>
 #include <Processors/QueryPlan/IQueryPlanStep.h>
 #include <QueryPipeline/Pipe.h>
 #include <QueryPipeline/QueryPipelineBuilder.h>
@@ -66,9 +65,12 @@ void SubstraitFileSourceStep::initializePipeline(DB::QueryPipelineBuilder & pipe
 
 void SubstraitFileSourceStep::applyFilters(const DB::ActionDAGNodes added_filter_nodes)
 {
-    filter_actions_dag = DB::ActionsDAG::buildFilterActionsDAG(added_filter_nodes.nodes);
-    for (const auto & processor : pipe.getProcessors())
-        if (auto * source = dynamic_cast<DB::SourceWithKeyCondition *>(processor.get()))
-            source->setKeyCondition(filter_actions_dag, context);
+    SourceStepWithFilter::applyFilters(std::move(added_filter_nodes));
+    if (filter_actions_dag)
+    {
+        for (const auto & processor : pipe.getProcessors())
+            if (auto * source = dynamic_cast<SubstraitFileSource *>(processor.get()))
+                source->setKeyCondition(filter_actions_dag, context);
+    }
 }
 }
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/SubstraitFileSourceStep.h b/cpp-ch/local-engine/Storages/SubstraitSource/SubstraitFileSourceStep.h
@@ -17,11 +17,10 @@
 
 #pragma once
 
+#include <Interpreters/Context_fwd.h>
 #include <Processors/QueryPlan/ReadFromPreparedSource.h>
 #include <Processors/QueryPlan/SourceStepWithFilter.h>
-#include <Storages/MergeTree/KeyCondition.h>
-#include <Interpreters/Context_fwd.h>
-#include <Core/NamesAndTypes.h>
+
 
 namespace local_engine
 {
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.cpp b/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.cpp
@@ -34,7 +34,8 @@ TextFormatFile::TextFormatFile(
 {
 }
 
-FormatFile::InputFormatPtr TextFormatFile::createInputFormat(const DB::Block & header)
+FormatFile::InputFormatPtr
+TextFormatFile::createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & /*filter_actions_dag*/)
 {
     auto read_buffer = read_buffer_builder->buildWithCompressionWrapper(file_info);
 
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.h b/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.h
@@ -32,7 +32,8 @@ class TextFormatFile : public FormatFile
         DB::ContextPtr context_, const substrait::ReadRel::LocalFiles::FileOrFiles & file_info_, ReadBufferBuilderPtr read_buffer_builder_);
     ~TextFormatFile() override = default;
 
-    FormatFile::InputFormatPtr createInputFormat(const DB::Block & header) override;
+    FormatFile::InputFormatPtr
+    createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag = nullptr) override;
 
     DB::NamesAndTypesList getSchema() const
     {
diff --git a/cpp-ch/local-engine/tests/gtest_parquet_read.cpp b/cpp-ch/local-engine/tests/gtest_parquet_read.cpp
@@ -45,6 +45,7 @@
 #include <tests/utils/gluten_test_util.h>
 #include <Common/BlockTypeUtils.h>
 #include <Common/DebugUtils.h>
+#include <Common/QueryContext.h>
 
 using namespace DB;
 using namespace local_engine;
@@ -111,8 +112,10 @@ void readData(const String & path, const std::map<String, Field> & fields)
     ReadBufferFromFile in(full_path);
 
     InputFormatPtr format;
+    auto parser_group
+        = std::make_shared<FormatParserGroup>(QueryContext::globalContext()->getSettingsRef(), 1, nullptr, QueryContext::globalContext());
     if constexpr (std::is_same_v<InputFormat, DB::ParquetBlockInputFormat>)
-        format = std::make_shared<InputFormat>(in, header, settings, 1, 1, 8192);
+        format = std::make_shared<InputFormat>(in, header, settings, parser_group, 8192);
     else
         format = std::make_shared<InputFormat>(in, header, settings);
 

Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,8 @@ bool ExcelTextFormatFile::useThis(const DB::ContextPtr & context)`
`62`	`62`	`return settingsEqual(context->getSettingsRef(), USE_EXCEL_PARSER, "true");`
`63`	`63`	`}`
`64`	`64`
`65`		`-FormatFile::InputFormatPtr ExcelTextFormatFile::createInputFormat(const DB::Block & header)`
	`65`	`+FormatFile::InputFormatPtr`
	`66`	`+ExcelTextFormatFile::createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & /filter_actions_dag/)`
`66`	`67`	`{`
`67`	`68`	`auto read_buffer = read_buffer_builder->build(file_info);`
`68`	`69`
Original file line number	Diff line number	Diff line change
`@@ -274,17 +274,12 @@ std::unique_ptr<NormalFileReader> createNormalFileReader(`
`274`	`274`	`const FormatFilePtr & file,`
`275`	`275`	`const DB::Block & to_read_header_,`
`276`	`276`	`const DB::Block & output_header_,`
`277`		`- const std::shared_ptr<const DB::KeyCondition> & key_condition = nullptr,`
	`277`	`+ const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag = nullptr,`
`278`	`278`	`const ColumnIndexFilterPtr & column_index_filter = nullptr)`
`279`	`279`	`{`
`280`	`280`	`file->initialize(column_index_filter);`
`281`	`281`	`auto createInputFormat = [&](const DB::Block & new_read_header_) -> FormatFile::InputFormatPtr`
`282`		`- {`
`283`		`- auto input_format = file->createInputFormat(new_read_header_);`
`284`		`- if (key_condition && input_format)`
`285`		`- input_format->inputFormat().setKeyCondition(key_condition);`
`286`		`- return input_format;`
`287`		`- };`
	`282`	`+ { return file->createInputFormat(new_read_header_, filter_actions_dag); };`
`288`	`283`
`289`	`284`	`if (file->getFileInfo().has_iceberg())`
`290`	`285`	`return iceberg::IcebergReader::create(file, to_read_header_, output_header_, createInputFormat);`
`@@ -316,11 +311,13 @@ std::unique_ptr<NormalFileReader> createNormalFileReader(`
`316`	`311`	`return std::make_unique<NormalFileReader>(file, to_read_header_, output_header_, input_format);`
`317`	`312`	`}`
`318`	`313`	`}`
	`314`	`+`
	`315`	`+/// TODO Remove ColumnIndexFilterPtr`
`319`	`316`	`std::unique_ptr<BaseReader> BaseReader::create(`
`320`	`317`	`const FormatFilePtr & current_file,`
`321`	`318`	`const DB::Block & readHeader,`
`322`	`319`	`const DB::Block & outputHeader,`
`323`		`- const std::shared_ptr<const DB::KeyCondition> & key_condition,`
	`320`	`+ const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag,`
`324`	`321`	`const ColumnIndexFilterPtr & column_index_filter)`
`325`	`322`	`{`
`326`	`323`	`if (!readHeader)`
`@@ -335,7 +332,7 @@ std::unique_ptr<BaseReader> BaseReader::create(`
`335`	`332`	`}`
`336`	`333`	`}`
`337`	`334`
`338`		`- return createNormalFileReader(current_file, readHeader, outputHeader, key_condition, column_index_filter);`
	`335`	`+ return createNormalFileReader(current_file, readHeader, outputHeader, filter_actions_dag, column_index_filter);`
`339`	`336`	`}`
`340`	`337`
`341`	`338`
Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,8 @@ JSONFormatFile::JSONFormatFile(`
`31`	`31`	`{`
`32`	`32`	`}`
`33`	`33`
`34`		`-FormatFile::InputFormatPtr JSONFormatFile::createInputFormat(const DB::Block & header)`
	`34`	`+FormatFile::InputFormatPtr`
	`35`	`+JSONFormatFile::createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & /filter_actions_dag/)`
`35`	`36`	`{`
`36`	`37`	`auto read_buffer = read_buffer_builder->buildWithCompressionWrapper(file_info);`
`37`	`38`
Original file line number	Diff line number	Diff line change
`@@ -36,7 +36,8 @@ ORCFormatFile::ORCFormatFile(`
`36`	`36`	`{`
`37`	`37`	`}`
`38`	`38`
`39`		`-FormatFile::InputFormatPtr ORCFormatFile::createInputFormat(const DB::Block & header)`
	`39`	`+FormatFile::InputFormatPtr`
	`40`	`+ORCFormatFile::createInputFormat(const DB::Block & header, const std::shared_ptr<const DB::ActionsDAG> & filter_actions_dag)`
`40`	`41`	`{`
`41`	`42`	`auto read_buffer = read_buffer_builder->build(file_info);`
`42`	`43`
`@@ -70,7 +71,8 @@ FormatFile::InputFormatPtr ORCFormatFile::createInputFormat(const DB::Block & he`
`70`	`71`	`format_settings.orc.reader_time_zone_name = mapped_timezone;`
`71`	`72`	`}`
`72`	`73`	`//TODO: support prefetch`
`73`		`- auto input_format = std::make_shared<DB::NativeORCBlockInputFormat>(*read_buffer, header, format_settings, false, 0);`
	`74`	`+ auto parser_group = std::make_shared<DB::FormatParserGroup>(context->getSettingsRef(), 1, filter_actions_dag, context);`
	`75`	`+ auto input_format = std::make_shared<DB::NativeORCBlockInputFormat>(*read_buffer, header, format_settings, false, 0, parser_group);`
`74`	`76`	`return std::make_shared<InputFormat>(std::move(read_buffer), input_format);`
`75`	`77`	`}`
`76`	`78`