Read optimization using Iceberg metadata

Author: ianton-ru

src/Core/ProtocolDefines.h

@@ -35,7 +35,8 @@ static constexpr auto DBMS_MIN_REVISION_WITH_AGGREGATE_FUNCTIONS_VERSIONING = 54
 
 static constexpr auto DBMS_CLUSTER_INITIAL_PROCESSING_PROTOCOL_VERSION = 1;
 static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_DATA_LAKE_METADATA = 2;
-static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION = 2;
+static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_DATA_LAKE_COLUMNS_METADATA = 3;
+static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION = 3;
 
 static constexpr auto DBMS_MIN_SUPPORTED_PARALLEL_REPLICAS_PROTOCOL_VERSION = 3;
 static constexpr auto DBMS_PARALLEL_REPLICAS_MIN_VERSION_WITH_MARK_SEGMENT_SIZE_FIELD = 4;

src/Core/Range.cpp

@@ -2,13 +2,15 @@
 #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
 {
 
+
 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 +153,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.

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;

src/Core/Settings.cpp

@@ -7024,6 +7024,9 @@ DECLARE(Bool, allow_experimental_ytsaurus_dictionary_source, false, R"(
     )", EXPERIMENTAL) \
     DECLARE(Bool, distributed_plan_force_shuffle_aggregation, false, R"(
 Use Shuffle aggregation strategy instead of PartialAggregation + Merge in distributed query plan.
+)", 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

@@ -132,6 +132,18 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"distributed_plan_force_shuffle_aggregation", 0, 0, "New experimental setting"},
             {"allow_experimental_insert_into_iceberg", false, false, "New setting."},
             /// RELEASE CLOSED
+            {"allow_experimental_database_iceberg", false, true, "Turned ON by default for Antalya"},
+            {"allow_experimental_database_unity_catalog", false, true, "Turned ON by default for Antalya"},
+            {"allow_experimental_database_glue_catalog", false, true, "Turned ON by default for Antalya"},
+            {"output_format_parquet_enum_as_byte_array", true, true, "Enable writing Enum as byte array in Parquet by default"},
+            {"lock_object_storage_task_distribution_ms", 0, 0, "New setting."},
+            {"object_storage_cluster", "", "", "New setting"},
+            {"object_storage_max_nodes", 0, 0, "New setting"},
+            {"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

@@ -7,7 +7,7 @@
 #include <Interpreters/Context.h>
 #include <Common/Exception.h>
 #include <Common/ObjectStorageKeyGenerator.h>
-
+#include <Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h>
 
 namespace DB
 {
@@ -97,6 +97,13 @@ WriteSettings IObjectStorage::patchSettings(const WriteSettings & write_settings
     return write_settings;
 }
 
+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;
+}
+
 std::string RelativePathWithMetadata::getPathOrPathToArchiveIfArchive() const
 {
     if (isArchive())

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>
@@ -107,6 +108,11 @@ struct ObjectMetadata
     ObjectAttributes attributes;
 };
 
+
+struct DataFileInfo;
+class DataFileMetaInfo;
+using DataFileMetaInfoPtr = std::shared_ptr<DataFileMetaInfo>;
+
 struct DataLakeObjectMetadata;
 
 struct RelativePathWithMetadata
@@ -116,13 +122,16 @@ struct RelativePathWithMetadata
     std::optional<ObjectMetadata> metadata;
     /// Delta lake related object metadata.
     std::optional<DataLakeObjectMetadata> data_lake_metadata;
+    /// Information about columns
+    std::optional<DataFileMetaInfoPtr> file_meta_info;
 
     RelativePathWithMetadata() = default;
 
     explicit RelativePathWithMetadata(String relative_path_, std::optional<ObjectMetadata> metadata_ = std::nullopt)
         : relative_path(std::move(relative_path_))
         , metadata(std::move(metadata_))
     {}
+    explicit RelativePathWithMetadata(const DataFileInfo & info, std::optional<ObjectMetadata> metadata_ = std::nullopt);
 
     RelativePathWithMetadata(const RelativePathWithMetadata & other) = default;
 
@@ -134,6 +143,9 @@ 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"); }
     virtual std::string getPathOrPathToArchiveIfArchive() const;
+
+    void setFileMetaInfo(std::optional<DataFileMetaInfoPtr> file_meta_info_ ) { file_meta_info = file_meta_info_; }
+    std::optional<DataFileMetaInfoPtr> getFileMetaInfo() const { return file_meta_info; }
 };
 
 struct ObjectKeyWithMetadata

src/Interpreters/ClusterFunctionReadTask.cpp

@@ -29,6 +29,8 @@ ClusterFunctionReadTaskResponse::ClusterFunctionReadTaskResponse(ObjectInfoPtr o
     if (object->data_lake_metadata.has_value())
         data_lake_metadata = object->data_lake_metadata.value();
 
+    file_meta_info = object->file_meta_info;
+
     const bool send_over_whole_archive = !context->getSettingsRef()[Setting::cluster_function_process_archive_on_multiple_nodes];
     path = send_over_whole_archive ? object->getPathOrPathToArchiveIfArchive() : object->getPath();
 }
@@ -45,6 +47,7 @@ ObjectInfoPtr ClusterFunctionReadTaskResponse::getObjectInfo() const
 
     auto object = std::make_shared<ObjectInfo>(path);
     object->data_lake_metadata = data_lake_metadata;
+    object->file_meta_info = file_meta_info;
     return object;
 }
 
@@ -61,6 +64,14 @@ void ClusterFunctionReadTaskResponse::serialize(WriteBuffer & out, size_t protoc
         else
             ActionsDAG().serialize(out, registry);
     }
+
+    if (protocol_version >= DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_DATA_LAKE_COLUMNS_METADATA)
+    {
+        if (file_meta_info.has_value())
+            file_meta_info.value()->serialize(out);
+        else
+            DataFileMetaInfo().serialize(out);
+    }
 }
 
 void ClusterFunctionReadTaskResponse::deserialize(ReadBuffer & in)
@@ -87,6 +98,14 @@ void ClusterFunctionReadTaskResponse::deserialize(ReadBuffer & in)
             data_lake_metadata.transform = std::move(transform);
         }
     }
+
+    if (protocol_version >= DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_DATA_LAKE_COLUMNS_METADATA)
+    {
+        auto info = std::make_shared<DataFileMetaInfo>(DataFileMetaInfo::deserialize(in));
+
+        if (!path.empty() && !info->empty())
+            file_meta_info = info;
+    }
 }
 
 }

src/Interpreters/ClusterFunctionReadTask.h

@@ -20,6 +20,8 @@ struct ClusterFunctionReadTaskResponse
     String path;
     /// Object metadata path, in case of data lake object.
     DataLakeObjectMetadata data_lake_metadata;
+    /// File's columns info
+    std::optional<DataFileMetaInfoPtr> file_meta_info;
 
     /// Convert received response into ObjectInfo.
     ObjectInfoPtr getObjectInfo() const;

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);