Merge branch 'antalya-25.6.5' into bugfix/antalya-25.6.5/lock_object_storage_task_distribution_ms_lost_host

Author: ianton-ru

programs/server/config.xml

@@ -1701,9 +1701,9 @@
         <!-- the ClickHouse core developers team. -->
         <!-- Doing so at least in pre-production environments is highly appreciated -->
         <!-- The reports are anonymized -->
-        <enabled>true</enabled>
-        <send_logical_errors>true</send_logical_errors>
-        <endpoint>https://crash.clickhouse.com/</endpoint>
+        <enabled>false</enabled>
+        <send_logical_errors>false</send_logical_errors>
+        <endpoint></endpoint>
     </send_crash_reports>
 
     <!-- Uncomment to disable ClickHouse internal DNS caching. -->

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

@@ -6918,6 +6918,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

@@ -83,6 +83,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/Parsers/FunctionSecretArgumentsFinder.h

@@ -241,7 +241,7 @@ class FunctionSecretArgumentsFinder
                 break;
             if (f->name() == "headers")
                 result.nested_maps.push_back(f->name());
-            else if (f->name() != "extra_credentials")
+            else if (f->name() != "extra_credentials" && f->name() != "equals")
                 break;
             count -= 1;
         }
@@ -266,6 +266,8 @@ class FunctionSecretArgumentsFinder
             return;
         }
 
+        findSecretNamedArgument("secret_access_key", url_arg_idx);
+
         /// We should check other arguments first because we don't need to do any replacement in case of
         /// s3('url', NOSIGN, 'format' [, 'compression'] [, extra_credentials(..)] [, headers(..)])
         /// s3('url', 'format', 'structure' [, 'compression'] [, extra_credentials(..)] [, headers(..)])
@@ -625,6 +627,8 @@ class FunctionSecretArgumentsFinder
             return;
         }
 
+        findSecretNamedArgument("secret_access_key", 0);
+
         /// We should check other arguments first because we don't need to do any replacement in case of
         /// S3('url', NOSIGN, 'format' [, 'compression'] [, extra_credentials(..)] [, headers(..)])
         /// S3('url', 'format', 'compression' [, extra_credentials(..)] [, headers(..)])

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/HivePartitioningUtils.cpp

@@ -9,6 +9,7 @@
 #include <Formats/EscapingRuleUtils.h>
 #include <Formats/FormatFactory.h>
 #include <Processors/Chunk.h>
+#include <DataTypes/IDataType.h>
 
 namespace DB
 {
@@ -85,7 +86,7 @@ NamesAndTypesList extractHivePartitionColumnsFromPath(
         {
             if (const auto type = tryInferDataTypeByEscapingRule(value, format_settings ? *format_settings : getFormatSettings(context), FormatSettings::EscapingRule::Raw))
             {
-                if (type->canBeInsideLowCardinality())
+                if (type->canBeInsideLowCardinality() && isStringOrFixedString(type))
                 {
                     hive_partition_columns_to_read_from_file_path.emplace_back(key, std::make_shared<DataTypeLowCardinality>(type));
                 }