Merge pull request #1019 from Altinity/feature/optimize_count_in_datalake

Read optimization using Iceberg metadata

Author: Enmk

src/Core/Range.cpp

@@ -2,13 +2,20 @@
 #include <Core/Range.h>
 #include <IO/Operators.h>
 #include <IO/WriteBufferFromString.h>
+#include <IO/ReadBufferFromString.h>
 #include <Common/FieldVisitorToString.h>
 #include <Common/FieldAccurateComparison.h>
 
 
 namespace DB
 {
 
+namespace ErrorCodes
+{
+extern const int INCORRECT_DATA;
+};
+
+
 FieldRef::FieldRef(ColumnsWithTypeAndName * columns_, size_t row_idx_, size_t column_idx_)
     : Field((*(*columns_)[column_idx_].column)[row_idx_]), columns(columns_), row_idx(row_idx_), column_idx(column_idx_)
 {
@@ -151,6 +158,13 @@ bool Range::isInfinite() const
     return left.isNegativeInfinity() && right.isPositiveInfinity();
 }
 
+/// [x, x]
+bool Range::isPoint() const
+{
+    return fullBounded() && left_included && right_included && equals(left, right)
+        && !left.isNegativeInfinity() && !left.isPositiveInfinity();
+}
+
 bool Range::intersectsRange(const Range & r) const
 {
     /// r to the left of me.
@@ -332,6 +346,29 @@ String Range::toString() const
     return str.str();
 }
 
+String Range::serialize() const
+{
+    WriteBufferFromOwnString str;
+
+    str << left_included << right_included;
+    writeFieldBinary(left, str);
+    writeFieldBinary(right, str);
+
+    return str.str();
+}
+
+void Range::deserialize(const String & range)
+{
+    if (range.empty())
+        throw Exception(ErrorCodes::INCORRECT_DATA, "Empty range dump");
+
+    ReadBufferFromOwnString str(range);
+
+    str >> left_included >> right_included;
+    left = readFieldBinary(str);
+    right = readFieldBinary(str);
+}
+
 Hyperrectangle intersect(const Hyperrectangle & a, const Hyperrectangle & b)
 {
     size_t result_size = std::min(a.size(), b.size());

src/Core/Range.h

@@ -94,6 +94,8 @@ struct Range
 
     bool isBlank() const;
 
+    bool isPoint() const;
+
     bool intersectsRange(const Range & r) const;
 
     bool containsRange(const Range & r) const;
@@ -114,6 +116,9 @@ struct Range
     bool nearByWith(const Range & r) const;
 
     String toString() const;
+
+    String serialize() const;
+    void deserialize(const String & range);
 };
 
 Range intersect(const Range & a, const Range & b);

src/Core/Settings.cpp

@@ -6908,6 +6908,9 @@ Allow retries in cluster request, when one node goes offline
 )", EXPERIMENTAL) \
     DECLARE(Bool, object_storage_remote_initiator, false, R"(
 Execute request to object storage as remote on one of object_storage_cluster nodes.
+)", EXPERIMENTAL) \
+    DECLARE(Bool, allow_experimental_iceberg_read_optimization, true, R"(
+Allow Iceberg read optimization based on Iceberg metadata.
 )", EXPERIMENTAL) \
     \
     /** Experimental timeSeries* aggregate functions. */ \

src/Core/SettingsChangesHistory.cpp

@@ -79,6 +79,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"object_storage_cluster_join_mode", "allow", "allow", "New setting"},
             {"object_storage_remote_initiator", false, false, "New setting."},
             {"allow_experimental_export_merge_tree_part", false, false, "New setting."},
+            {"allow_experimental_iceberg_read_optimization", true, true, "New setting."},
             {"export_merge_tree_part_overwrite_file_if_exists", false, false, "New setting."},
         });
         addSettingsChanges(settings_changes_history, "25.6",

src/Disks/ObjectStorages/IObjectStorage.cpp

@@ -8,6 +8,9 @@
 #include <Common/Exception.h>
 #include <Common/ObjectStorageKeyGenerator.h>
 
+/// TODO: move DataFileInfo into separate file
+#include <Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h>
+
 #include <Poco/JSON/Object.h>
 #include <Poco/JSON/Parser.h>
 #include <Poco/JSON/JSONException.h>
@@ -101,6 +104,28 @@ WriteSettings IObjectStorage::patchSettings(const WriteSettings & write_settings
     return write_settings;
 }
 
+RelativePathWithMetadata::RelativePathWithMetadata(const String & task_string, std::optional<ObjectMetadata> metadata_)
+    : metadata(std::move(metadata_))
+    , command(task_string)
+{
+    if (!command.isParsed())
+        relative_path = task_string;
+    else
+    {
+        auto file_path = command.getFilePath();
+        if (file_path.has_value())
+            relative_path = file_path.value();
+        file_meta_info = command.getFileMetaInfo();
+    }
+}
+
+RelativePathWithMetadata::RelativePathWithMetadata(const DataFileInfo & info, std::optional<ObjectMetadata> metadata_)
+    : metadata(std::move(metadata_))
+{
+    relative_path = info.file_path;
+    file_meta_info = info.file_meta_info;
+}
+
 void RelativePathWithMetadata::loadMetadata(ObjectStoragePtr object_storage, bool ignore_non_existent_file)
 {
     if (!metadata)
@@ -129,20 +154,29 @@ RelativePathWithMetadata::CommandInTaskResponse::CommandInTaskResponse(const std
 
         successfully_parsed = true;
 
+        if (json->has("file_path"))
+            file_path = json->getValue<std::string>("file_path");
         if (json->has("retry_after_us"))
             retry_after_us = json->getValue<size_t>("retry_after_us");
+        if (json->has("meta_info"))
+            file_meta_info = std::make_shared<DataFileMetaInfo>(json->getObject("meta_info"));
     }
     catch (const Poco::JSON::JSONException &)
     { /// Not a JSON
         return;
     }
 }
 
-std::string RelativePathWithMetadata::CommandInTaskResponse::to_string() const
+std::string RelativePathWithMetadata::CommandInTaskResponse::toString() const
 {
     Poco::JSON::Object json;
+
+    if (file_path.has_value())
+        json.set("file_path", file_path.value());
     if (retry_after_us.has_value())
         json.set("retry_after_us", retry_after_us.value());
+    if (file_meta_info.has_value())
+        json.set("meta_info", file_meta_info.value()->toJson());
 
     std::ostringstream oss;
     oss.exceptions(std::ios::failbit);

src/Disks/ObjectStorages/IObjectStorage.h

@@ -9,6 +9,7 @@
 
 #include <Poco/Timestamp.h>
 #include <Poco/Util/AbstractConfiguration.h>
+#include <Poco/JSON/Object.h>
 #include <Core/Defines.h>
 #include <IO/ReadSettings.h>
 #include <IO/WriteSettings.h>
@@ -101,6 +102,10 @@ struct ObjectMetadata
     ObjectAttributes attributes;
 };
 
+struct DataFileInfo;
+class DataFileMetaInfo;
+using DataFileMetaInfoPtr = std::shared_ptr<DataFileMetaInfo>;
+
 struct RelativePathWithMetadata
 {
     class CommandInTaskResponse
@@ -109,31 +114,35 @@ struct RelativePathWithMetadata
         CommandInTaskResponse() = default;
         explicit CommandInTaskResponse(const std::string & task);
 
-        bool is_parsed() const { return successfully_parsed; }
-        void set_retry_after_us(Poco::Timestamp::TimeDiff time_us) { retry_after_us = time_us; }
+        bool isParsed() const { return successfully_parsed; }
+        void setFilePath(const std::string & file_path_ ) { file_path = file_path_; }
+        void setRetryAfterUs(Poco::Timestamp::TimeDiff time_us) { retry_after_us = time_us; }
+        void setFileMetaInfo(DataFileMetaInfoPtr file_meta_info_ ) { file_meta_info = file_meta_info_; }
+
+        std::string toString() const;
+
+        std::optional<std::string> getFilePath() const { return file_path; }
 
-        std::string to_string() const;
+        std::optional<Poco::Timestamp::TimeDiff> getRetryAfterUs() const { return retry_after_us; }
 
-        std::optional<Poco::Timestamp::TimeDiff> get_retry_after_us() const { return retry_after_us; }
+        std::optional<DataFileMetaInfoPtr> getFileMetaInfo() const { return file_meta_info; }
 
     private:
         bool successfully_parsed = false;
+        std::optional<std::string> file_path;
         std::optional<Poco::Timestamp::TimeDiff> retry_after_us;
+        std::optional<DataFileMetaInfoPtr> file_meta_info;
     };
 
     String relative_path;
     std::optional<ObjectMetadata> metadata;
     CommandInTaskResponse command;
+    std::optional<DataFileMetaInfoPtr> file_meta_info;
 
     RelativePathWithMetadata() = default;
 
-    explicit RelativePathWithMetadata(const String & task_string, std::optional<ObjectMetadata> metadata_ = std::nullopt)
-        : metadata(std::move(metadata_))
-        , command(task_string)
-    {
-        if (!command.is_parsed())
-            relative_path = task_string;
-    }
+    explicit RelativePathWithMetadata(const String & task_string, std::optional<ObjectMetadata> metadata_ = std::nullopt);
+    explicit RelativePathWithMetadata(const DataFileInfo & info, std::optional<ObjectMetadata> metadata_ = std::nullopt);
 
     virtual ~RelativePathWithMetadata() = default;
 
@@ -145,6 +154,10 @@ struct RelativePathWithMetadata
     virtual std::string getPathToArchive() const { throw Exception(ErrorCodes::LOGICAL_ERROR, "Not an archive"); }
     virtual size_t fileSizeInArchive() const { throw Exception(ErrorCodes::LOGICAL_ERROR, "Not an archive"); }
 
+    void setFileMetaInfo(DataFileMetaInfoPtr file_meta_info_ ) { file_meta_info = file_meta_info_; }
+    void setFileMetaInfo(std::optional<DataFileMetaInfoPtr> file_meta_info_ ) { file_meta_info = file_meta_info_; }
+    std::optional<DataFileMetaInfoPtr> getFileMetaInfo() const { return file_meta_info; }
+
     void loadMetadata(ObjectStoragePtr object_storage, bool ignore_non_existent_file);
     const CommandInTaskResponse & getCommand() const { return command; }
 };

src/Processors/Chunk.cpp

@@ -108,7 +108,12 @@ void Chunk::addColumn(ColumnPtr column)
 
 void Chunk::addColumn(size_t position, ColumnPtr column)
 {
-    if (position >= columns.size())
+    if (position == columns.size())
+    {
+        addColumn(column);
+        return;
+    }
+    if (position > columns.size())
         throw Exception(ErrorCodes::POSITION_OUT_OF_BOUND,
                         "Position {} out of bound in Chunk::addColumn(), max position = {}",
                         position, !columns.empty() ? columns.size() - 1 : 0);

src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h

@@ -104,7 +104,7 @@ class DataLakeConfiguration : public BaseStorageConfiguration, public std::enabl
             return std::nullopt;
         auto data_files = current_metadata->getDataFiles();
         if (!data_files.empty())
-            return data_files[0];
+            return data_files[0].file_path;
         return std::nullopt;
     }
 

src/Storages/ObjectStorage/DataLakes/DeltaLakeMetadata.cpp

@@ -158,7 +158,7 @@ struct DeltaLakeMetadataImpl
     struct DeltaLakeMetadata
     {
         NamesAndTypesList schema;
-        Strings data_files;
+        DataFileInfos data_files;
         DeltaLakePartitionColumns partition_columns;
     };
 
@@ -195,7 +195,7 @@ struct DeltaLakeMetadataImpl
                 processMetadataFile(key, current_schema, current_partition_columns, result_files);
         }
 
-        return DeltaLakeMetadata{current_schema, Strings(result_files.begin(), result_files.end()), current_partition_columns};
+        return DeltaLakeMetadata{current_schema, DataFileInfos(result_files.begin(), result_files.end()), current_partition_columns};
     }
 
     /**

src/Storages/ObjectStorage/DataLakes/DeltaLakeMetadata.h

@@ -35,7 +35,7 @@ class DeltaLakeMetadata final : public IDataLakeMetadata
 
     DeltaLakeMetadata(ObjectStoragePtr object_storage_, ConfigurationObserverPtr configuration_, ContextPtr context_);
 
-    Strings getDataFiles() const override { return data_files; }
+    DataFileInfos getDataFiles() const override { return data_files; }
 
     NamesAndTypesList getTableSchema() const override { return schema; }
 
@@ -67,12 +67,12 @@ class DeltaLakeMetadata final : public IDataLakeMetadata
         ContextPtr context) const override;
 
 private:
-    mutable Strings data_files;
+    mutable DataFileInfos data_files;
     NamesAndTypesList schema;
     DeltaLakePartitionColumns partition_columns;
     ObjectStoragePtr object_storage;
 
-    Strings getDataFiles(const ActionsDAG *) const { return data_files; }
+    DataFileInfos getDataFiles(const ActionsDAG *) const { return data_files; }
 };
 
 }