[GLUTEN-1632][CH]Daily Update Clickhouse Version (20250629) (#10080)

kyligence-git · kyligence-git · Chang chen · web-flow · commit 748a27245b7c · 2025-06-29T21:48:42.000+08:00
* [GLUTEN-1632][CH]Daily Update Clickhouse Version (20250629) * Fix Build due to ClickHouse/ClickHouse#80931 * Fix Build due to ClickHouse/ClickHouse#81976 * Fix Build due to ClickHouse/ClickHouse#82508 * Try to Fix issue caused by ClickHouse/ClickHouse#81754 see ClickHouse/ClickHouse#82379 * Fix UT due to ClickHouse/ClickHouse#82358 --------- Co-authored-by: kyligence-git <gluten@kyligence.io> Co-authored-by: Chang chen <chenchang@apache.com>
diff --git a/cpp-ch/clickhouse.version b/cpp-ch/clickhouse.version
@@ -1,3 +1,3 @@
 CH_ORG=Kyligence
-CH_BRANCH=rebase_ch/20250621
-CH_COMMIT=8960b38d0bf
+CH_BRANCH=rebase_ch/20250629
+CH_COMMIT=addbf00cfd7
diff --git a/cpp-ch/local-engine/Common/DebugUtils.cpp b/cpp-ch/local-engine/Common/DebugUtils.cpp
@@ -361,7 +361,7 @@ void dumpMessage(const google::protobuf::Message & message, const char * type, b
 
     if (!force && !logger->debug())
         return;
-    pb_util::JsonOptions options;
+    pb_util::JsonPrintOptions options;
     std::string json;
     if (auto s = MessageToJsonString(message, &json, options); !s.ok())
         throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Can not convert {} to Json", type);
diff --git a/cpp-ch/local-engine/Disks/registerGlutenDisks.cpp b/cpp-ch/local-engine/Disks/registerGlutenDisks.cpp
@@ -87,7 +87,7 @@ void registerGlutenDisks(bool global_skip_access_check)
             config_prefix,
             object_storage_creator);
 
-        disk->startup(context, skip_access_check);
+        disk->startup(skip_access_check);
         return disk;
     };
 
@@ -134,7 +134,7 @@ void registerGlutenDisks(bool global_skip_access_check)
             config_prefix,
             object_storage_creator);
 
-        disk->startup(context, skip_access_check);
+        disk->startup(skip_access_check);
         return disk;
     };
 
diff --git a/cpp-ch/local-engine/Join/StorageJoinFromReadBuffer.cpp b/cpp-ch/local-engine/Join/StorageJoinFromReadBuffer.cpp
@@ -101,13 +101,13 @@ StorageJoinFromReadBuffer::StorageJoinFromReadBuffer(
         collectAllInputs(data, right_sample_block);
 }
 
-void StorageJoinFromReadBuffer::buildJoin(Blocks & data, const Block header, std::shared_ptr<DB::TableJoin> analyzed_join)
+void StorageJoinFromReadBuffer::buildJoin(const Blocks & data, const Block & header, std::shared_ptr<DB::TableJoin> analyzed_join)
 {
     auto build_join = [&]
     {
-        join = std::make_shared<HashJoin>(analyzed_join, header, overwrite, row_count);
-        for (Block block : data)
-            join->addBlockToJoin(std::move(block), true);
+        join = std::make_shared<HashJoin>(analyzed_join, header, overwrite, row_count, "", false, true);
+        for (const Block& block : data)
+            join->addBlockToJoin(block, true);
     };
     /// Record memory usage in Total Memory Tracker
     ThreadFromGlobalPoolNoTracingContextPropagation thread(build_join);
@@ -132,7 +132,7 @@ void StorageJoinFromReadBuffer::buildJoinLazily(DB::Block header, std::shared_pt
         std::unique_lock lock(join_mutex);
         if (join)
             return;
-        join = std::make_shared<HashJoin>(analyzed_join, header, overwrite, row_count);
+        join = std::make_shared<HashJoin>(analyzed_join, header, overwrite, row_count, "", false, true);
         while (!input_blocks.empty())
         {
             auto & block = *input_blocks.begin();
@@ -154,11 +154,11 @@ void StorageJoinFromReadBuffer::buildJoinLazily(DB::Block header, std::shared_pt
 }
 
 
-/// The column names of 'rgiht_header' could be different from the ones in `input_blocks`, and we must
+/// The column names of 'right_header' could be different from the ones in `input_blocks`, and we must
 /// use 'right_header' to build the HashJoin. Otherwise, it will cause exceptions with name mismatches.
 ///
 /// In most cases, 'getJoinLocked' is called only once, and the input_blocks should not be too large.
-/// This is will be OK.
+/// This will be OK.
 DB::JoinPtr StorageJoinFromReadBuffer::getJoinLocked(std::shared_ptr<DB::TableJoin> analyzed_join, DB::ContextPtr /*context*/)
 {
     if ((analyzed_join->forceNullableRight() && !use_nulls)
diff --git a/cpp-ch/local-engine/Join/StorageJoinFromReadBuffer.h b/cpp-ch/local-engine/Join/StorageJoinFromReadBuffer.h
@@ -72,7 +72,7 @@ class StorageJoinFromReadBuffer
     bool is_null_aware_anti_join;
 
     void readAllBlocksFromInput(DB::ReadBuffer & in);
-    void buildJoin(DB::Blocks & data, const DB::Block header, std::shared_ptr<DB::TableJoin> analyzed_join);
+    void buildJoin(const DB::Blocks & data, const DB::Block & header, std::shared_ptr<DB::TableJoin> analyzed_join);
     void collectAllInputs(DB::Blocks & data, const DB::Block header);
     void buildJoinLazily(DB::Block header, std::shared_ptr<DB::TableJoin> analyzed_join);
 };
diff --git a/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.cpp b/cpp-ch/local-engine/Parser/RelParsers/CrossRelParser.cpp
@@ -171,10 +171,6 @@ DB::QueryPlanPtr CrossRelParser::parseJoin(const substrait::CrossRel & join, DB:
     QueryPlanPtr query_plan;
     if (storage_join)
     {
-        /// FIXME: There is mistake in HashJoin::needUsedFlagsForPerRightTableRow which returns true when
-        /// join clauses is empty. But in fact there should not be any join clause in cross join.
-        table_join->addDisjunct();
-
         auto broadcast_hash_join = storage_join->getJoinLocked(table_join, context);
         // table_join->resetKeys();
         QueryPlanStepPtr join_step = std::make_unique<FilledJoinStep>(left->getCurrentHeader(), broadcast_hash_join, 8192);
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
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/SubstraitFileSourceStep.h b/cpp-ch/local-engine/Storages/SubstraitSource/SubstraitFileSourceStep.h
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.cpp b/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.cpp
diff --git a/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.h b/cpp-ch/local-engine/Storages/SubstraitSource/TextFormatFile.h
diff --git a/cpp-ch/local-engine/tests/gtest_parquet_read.cpp b/cpp-ch/local-engine/tests/gtest_parquet_read.cpp

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`