Support orc filter push down (file + stripe + rowgroup level)

src/Core/Settings.h

@@ -895,6 +895,7 @@ class IColumn;
     M(Bool, input_format_allow_seeks, true, "Allow seeks while reading in ORC/Parquet/Arrow input formats", 0) \
     M(Bool, input_format_orc_allow_missing_columns, false, "Allow missing columns while reading ORC input formats", 0) \
     M(Bool, input_format_orc_use_fast_decoder, true, "Use a faster ORC decoder implementation.", 0) \
+    M(Bool, input_format_orc_filter_push_down, true, "When reading ORC files, skip whole stripes or row groups based on the WHERE/PREWHERE expressions, min/max statistics or bloom filter in the ORC metadata.", 0) \
     M(Bool, input_format_parquet_allow_missing_columns, false, "Allow missing columns while reading Parquet input formats", 0) \
     M(UInt64, input_format_parquet_local_file_min_bytes_for_seek, 8192, "Min bytes required for local read (file) to do seek, instead of read with ignore in Parquet input format", 0) \
     M(Bool, input_format_arrow_allow_missing_columns, false, "Allow missing columns while reading Arrow input formats", 0) \

src/Formats/FormatFactory.cpp

@@ -193,6 +193,7 @@ FormatSettings getFormatSettings(ContextPtr context, const Settings & settings)
     format_settings.orc.output_string_as_string = settings.output_format_orc_string_as_string;
     format_settings.orc.output_compression_method = settings.output_format_orc_compression_method;
     format_settings.orc.use_fast_decoder = settings.input_format_orc_use_fast_decoder;
+    format_settings.orc.filter_push_down = settings.input_format_orc_filter_push_down;
     format_settings.defaults_for_omitted_fields = settings.input_format_defaults_for_omitted_fields;
     format_settings.capn_proto.enum_comparing_mode = settings.format_capn_proto_enum_comparising_mode;
     format_settings.capn_proto.skip_fields_with_unsupported_types_in_schema_inference = settings.input_format_capn_proto_skip_fields_with_unsupported_types_in_schema_inference;

src/Formats/FormatSettings.h

@@ -362,6 +362,7 @@ struct FormatSettings
         bool output_string_as_string = false;
         ORCCompression output_compression_method = ORCCompression::NONE;
         bool use_fast_decoder = true;
+        bool filter_push_down = true;
     } orc;
 
     /// For capnProto format we should determine how to

src/Interpreters/Set.h

@@ -193,7 +193,7 @@ using FunctionPtr = std::shared_ptr<IFunction>;
   */
 struct FieldValue
 {
-    FieldValue(MutableColumnPtr && column_) : column(std::move(column_)) {}
+    explicit FieldValue(MutableColumnPtr && column_) : column(std::move(column_)) {}
     void update(const Field & x);
 
     bool isNormal() const { return !value.isPositiveInfinity() && !value.isNegativeInfinity(); }
@@ -225,6 +225,8 @@ class MergeTreeSetIndex
 
     size_t size() const { return ordered_set.at(0)->size(); }
 
+    const Columns & getOrderedSet() const { return ordered_set; }
+
     bool hasMonotonicFunctionsChain() const;
 
     BoolMask checkInRange(const std::vector<Range> & key_ranges, const DataTypes & data_types, bool single_point = false) const;

src/Processors/Formats/IInputFormat.cpp

@@ -6,7 +6,7 @@ namespace DB
 {
 
 IInputFormat::IInputFormat(Block header, ReadBuffer * in_)
-    : ISource(std::move(header)), in(in_)
+    : SourceWithKeyCondition(std::move(header)), in(in_)
 {
     column_mapping = std::make_shared<ColumnMapping>();
 }

src/Processors/Formats/IInputFormat.h

@@ -1,10 +1,11 @@
 #pragma once
 
-#include <Processors/Formats/InputFormatErrorsLogger.h>
-#include <Processors/ISource.h>
+#include <Formats/ColumnMapping.h>
 #include <IO/ReadBuffer.h>
 #include <Interpreters/Context.h>
-#include <Formats/ColumnMapping.h>
+#include <Processors/Formats/InputFormatErrorsLogger.h>
+#include <Processors/SourceWithKeyCondition.h>
+#include <Storages/MergeTree/KeyCondition.h>
 
 
 namespace DB
@@ -16,7 +17,7 @@ using ColumnMappingPtr = std::shared_ptr<ColumnMapping>;
 
 /** Input format is a source, that reads data from ReadBuffer.
   */
-class IInputFormat : public ISource
+class IInputFormat : public SourceWithKeyCondition
 {
 protected:
 
@@ -26,10 +27,6 @@ class IInputFormat : public ISource
     /// ReadBuffer can be nullptr for random-access formats.
     IInputFormat(Block header, ReadBuffer * in_);
 
-    /// If the format is used by a SELECT query, this method may be called.
-    /// The format may use it for filter pushdown.
-    virtual void setQueryInfo(const SelectQueryInfo &, ContextPtr) {}
-
     /** In some usecase (hello Kafka) we need to read a lot of tiny streams in exactly the same format.
      * The recreating of parser for each small stream takes too long, so we introduce a method
      * resetParser() which allow to reset the state of parser to continue reading of