Merge branch 'antalya-25.8' into backports/antalya-25.8/87220

Author: Enmk

src/Databases/DataLake/DatabaseDataLake.cpp

@@ -434,17 +434,18 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con
         configuration,
         configuration->createObjectStorage(context_copy, /* is_readonly */ false),
         StorageID(getDatabaseName(), name),
-        /* columns */columns,
-        /* constraints */ConstraintsDescription{},
-        /* partition_by */nullptr,
+        /* columns */ columns,
+        /* constraints */ ConstraintsDescription{},
+        /* partition_by */ nullptr,
         context_copy,
-        /* comment */"",
+        /* comment */ "",
         getFormatSettings(context_copy),
         LoadingStrictnessLevel::CREATE,
         getCatalog(),
-        /* if_not_exists*/true,
-        /* is_datalake_query*/true,
-        /* lazy_init */true);
+        /* if_not_exists */ true,
+        /* is_datalake_query */ true,
+        /* is_table_function */ true,
+        /* lazy_init */ true);
 }
 
 void DatabaseDataLake::dropTable( /// NOLINT

src/Parsers/ASTSystemQuery.cpp

@@ -163,7 +163,13 @@ void ASTSystemQuery::formatImpl(WriteBuffer & ostr, const FormatSettings & setti
 
     print_keyword("SYSTEM") << " ";
     print_keyword(typeToString(type));
-    if (!cluster.empty())
+
+    std::unordered_set<Type> queries_with_on_cluster_at_end = {
+        Type::DROP_FILESYSTEM_CACHE,
+        Type::SYNC_FILESYSTEM_CACHE,
+    };
+
+    if (!queries_with_on_cluster_at_end.contains(type) && !cluster.empty())
         formatOnCluster(ostr, settings);
 
     switch (type)
@@ -519,6 +525,9 @@ void ASTSystemQuery::formatImpl(WriteBuffer & ostr, const FormatSettings & setti
         case Type::END:
             throw Exception(ErrorCodes::LOGICAL_ERROR, "Unknown SYSTEM command");
     }
+
+    if (queries_with_on_cluster_at_end.contains(type) && !cluster.empty())
+        formatOnCluster(ostr, settings);
 }
 
 

src/Storages/IStorage.h

@@ -412,6 +412,7 @@ class IStorage : public std::enable_shared_from_this<IStorage>, public TypePromo
         size_t /*max_block_size*/,
         size_t /*num_streams*/);
 
+public:
     /// Should we process blocks of data returned by the storage in parallel
     /// even when the storage returned only one stream of data for reading?
     /// It is beneficial, for example, when you read from a file quickly,
@@ -422,7 +423,6 @@ class IStorage : public std::enable_shared_from_this<IStorage>, public TypePromo
     /// useless).
     virtual bool parallelizeOutputAfterReading(ContextPtr) const { return !isSystemStorage(); }
 
-public:
     /// Other version of read which adds reading step to query plan.
     /// Default implementation creates ReadFromStorageStep and uses usual read.
     /// Can be called after `shutdown`, but not after `drop`.

src/Storages/IStorageCluster.cpp

@@ -288,6 +288,8 @@ void IStorageCluster::read(
         return;
     }
 
+    updateConfigurationIfNeeded(context);
+
     storage_snapshot->check(column_names);
 
     const auto & settings = context->getSettingsRef();

src/Storages/IStorageCluster.h

@@ -97,6 +97,8 @@ class IStorageCluster : public IStorage
         throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Method writeFallBackToPure is not supported by storage {}", getName());
     }
 
+    virtual void updateConfigurationIfNeeded(ContextPtr /* context */) {}
+
 private:
     static ClusterPtr getClusterImpl(ContextPtr context, const String & cluster_name_, size_t max_hosts = 0);
 

src/Storages/ObjectStorage/Local/Configuration.cpp

@@ -81,4 +81,20 @@ StorageObjectStorageQuerySettings StorageLocalConfiguration::getQuerySettings(co
         .ignore_non_existent_file = false};
 }
 
+ASTPtr StorageLocalConfiguration::createArgsWithAccessData() const
+{
+    auto arguments = std::make_shared<ASTExpressionList>();
+
+    arguments->children.push_back(std::make_shared<ASTLiteral>(path.path));
+    if (getFormat() != "auto")
+        arguments->children.push_back(std::make_shared<ASTLiteral>(getFormat()));
+    if (getStructure() != "auto")
+        arguments->children.push_back(std::make_shared<ASTLiteral>(getStructure()));
+    if (getCompressionMethod() != "auto")
+        arguments->children.push_back(std::make_shared<ASTLiteral>(getCompressionMethod()));
+
+    return arguments;
+}
+
+
 }

src/Storages/ObjectStorage/Local/Configuration.h

@@ -60,6 +60,8 @@ class StorageLocalConfiguration : public StorageObjectStorageConfiguration
 
     void addStructureAndFormatToArgsIfNeeded(ASTs &, const String &, const String &, ContextPtr, bool) override { }
 
+    ASTPtr createArgsWithAccessData() const override;
+
 private:
     void fromNamedCollection(const NamedCollection & collection, ContextPtr context) override;
     void fromAST(ASTs & args, ContextPtr context, bool with_structure) override;

src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp

@@ -91,6 +91,7 @@ StorageObjectStorageCluster::StorageObjectStorageCluster(
     std::shared_ptr<DataLake::ICatalog> catalog,
     bool if_not_exists,
     bool is_datalake_query,
+    bool is_table_function,
     bool lazy_init)
     : IStorageCluster(
         cluster_name_, table_id_, getLogger(fmt::format("{}({})", configuration_->getEngineName(), table_id_.table_name)))
@@ -145,6 +146,10 @@ StorageObjectStorageCluster::StorageObjectStorageCluster(
         tryLogCurrentException(log_);
     }
 
+    // For tables need to update configuration on each read
+    // because data can be changed after previous update
+    update_configuration_on_read_write = !is_table_function;
+
     ColumnsDescription columns{columns_in_table_or_function_definition};
     std::string sample_path;
     if (need_resolve_columns_or_format)
@@ -295,6 +300,7 @@ void StorageObjectStorageCluster::updateQueryForDistributedEngineIfNeeded(ASTPtr
         {"IcebergS3", "icebergS3"},
         {"IcebergAzure", "icebergAzure"},
         {"IcebergHDFS", "icebergHDFS"},
+        {"IcebergLocal", "icebergLocal"},
         {"DeltaLake", "deltaLake"},
         {"DeltaLakeS3", "deltaLakeS3"},
         {"DeltaLakeAzure", "deltaLakeAzure"},
@@ -416,6 +422,7 @@ void StorageObjectStorageCluster::updateQueryToSendIfNeeded(
             {"icebergS3", "icebergS3Cluster"},
             {"icebergAzure", "icebergAzureCluster"},
             {"icebergHDFS", "icebergHDFSCluster"},
+            {"icebergLocal", "icebergLocalCluster"},
             {"deltaLake", "deltaLakeCluster"},
             {"deltaLakeS3", "deltaLakeS3Cluster"},
             {"deltaLakeAzure", "deltaLakeAzureCluster"},
@@ -741,6 +748,18 @@ IDataLakeMetadata * StorageObjectStorageCluster::getExternalMetadata(ContextPtr
     return configuration->getExternalMetadata();
 }
 
+void StorageObjectStorageCluster::updateConfigurationIfNeeded(ContextPtr context)
+{
+    if (update_configuration_on_read_write)
+    {
+        configuration->update(
+            object_storage,
+            context,
+            /* if_not_updated_before */false,
+            /* check_consistent_with_previous_metadata */false);
+    }
+}
+
 void StorageObjectStorageCluster::checkAlterIsPossible(const AlterCommands & commands, ContextPtr context) const
 {
     if (getClusterName(context).empty())
@@ -905,4 +924,74 @@ bool StorageObjectStorageCluster::prefersLargeBlocks() const
     return IStorageCluster::prefersLargeBlocks();
 }
 
+bool StorageObjectStorageCluster::supportsPartitionBy() const
+{
+    if (pure_storage)
+        return pure_storage->supportsPartitionBy();
+    return IStorageCluster::supportsPartitionBy();
+}
+
+bool StorageObjectStorageCluster::supportsSubcolumns() const
+{
+    if (pure_storage)
+        return pure_storage->supportsSubcolumns();
+    return IStorageCluster::supportsSubcolumns();
+}
+
+bool StorageObjectStorageCluster::supportsDynamicSubcolumns() const
+{
+    if (pure_storage)
+        return pure_storage->supportsDynamicSubcolumns();
+    return IStorageCluster::supportsDynamicSubcolumns();
+}
+
+bool StorageObjectStorageCluster::supportsTrivialCountOptimization(const StorageSnapshotPtr & snapshot, ContextPtr context) const
+{
+    if (pure_storage)
+        return pure_storage->supportsTrivialCountOptimization(snapshot, context);
+    return IStorageCluster::supportsTrivialCountOptimization(snapshot, context);
+}
+
+bool StorageObjectStorageCluster::supportsPrewhere() const
+{
+    if (pure_storage)
+        return pure_storage->supportsPrewhere();
+    return IStorageCluster::supportsPrewhere();
+}
+
+bool StorageObjectStorageCluster::canMoveConditionsToPrewhere() const
+{
+    if (pure_storage)
+        return pure_storage->canMoveConditionsToPrewhere();
+    return IStorageCluster::canMoveConditionsToPrewhere();
+}
+
+std::optional<NameSet> StorageObjectStorageCluster::supportedPrewhereColumns() const
+{
+    if (pure_storage)
+        return pure_storage->supportedPrewhereColumns();
+    return IStorageCluster::supportedPrewhereColumns();
+}
+
+IStorageCluster::ColumnSizeByName StorageObjectStorageCluster::getColumnSizes() const
+{
+    if (pure_storage)
+        return pure_storage->getColumnSizes();
+    return IStorageCluster::getColumnSizes();
+}
+
+bool StorageObjectStorageCluster::parallelizeOutputAfterReading(ContextPtr context) const
+{
+    if (pure_storage)
+        return pure_storage->parallelizeOutputAfterReading(context);
+    return IStorageCluster::parallelizeOutputAfterReading(context);
+}
+
+bool StorageObjectStorageCluster::supportsDelete() const
+{
+    if (pure_storage)
+        return pure_storage->supportsDelete();
+    return IStorageCluster::supportsDelete();
+}
+
 }

src/Storages/ObjectStorage/StorageObjectStorageCluster.h

@@ -25,7 +25,8 @@ class StorageObjectStorageCluster : public IStorageCluster
         std::shared_ptr<DataLake::ICatalog> catalog,
         bool if_not_exists,
         bool is_datalake_query,
-        bool lazy_init = false);
+        bool is_table_function,
+        bool lazy_init);
 
     std::string getName() const override;
 
@@ -132,6 +133,24 @@ class StorageObjectStorageCluster : public IStorageCluster
         ContextPtr /* context */) override;
     bool prefersLargeBlocks() const override;
 
+    bool supportsPartitionBy() const override;
+
+    bool supportsSubcolumns() const override;
+
+    bool supportsDynamicSubcolumns() const override;
+
+    bool supportsTrivialCountOptimization(const StorageSnapshotPtr &, ContextPtr) const override;
+
+    /// Things required for PREWHERE.
+    bool supportsPrewhere() const override;
+    bool canMoveConditionsToPrewhere() const override;
+    std::optional<NameSet> supportedPrewhereColumns() const override;
+    ColumnSizeByName getColumnSizes() const override;
+
+    bool parallelizeOutputAfterReading(ContextPtr context) const override;
+
+    bool supportsDelete() const override;
+
 private:
     void updateQueryToSendIfNeeded(
         ASTPtr & query,
@@ -154,6 +173,8 @@ class StorageObjectStorageCluster : public IStorageCluster
         ContextPtr context,
         bool async_insert) override;
 
+    void updateConfigurationIfNeeded(ContextPtr context) override;
+
     /*
     In case the table was created with `object_storage_cluster` setting,
     modify the AST query object so that it uses the table function implementation
@@ -176,6 +197,7 @@ class StorageObjectStorageCluster : public IStorageCluster
 
     /// non-clustered storage to fall back on pure realisation if needed
     std::shared_ptr<StorageObjectStorage> pure_storage;
+    bool update_configuration_on_read_write;
 };
 
 }

src/Storages/ObjectStorage/StorageObjectStorageDefinitions.h

@@ -155,6 +155,13 @@ struct IcebergHDFSClusterDefinition
     static constexpr auto non_clustered_storage_engine_name = IcebergHDFSDefinition::storage_engine_name;
 };
 
+struct IcebergLocalClusterDefinition
+{
+    static constexpr auto name = "icebergLocalCluster";
+    static constexpr auto storage_engine_name = "IcebergLocalCluster";
+    static constexpr auto non_clustered_storage_engine_name = IcebergLocalDefinition::storage_engine_name;
+};
+
 struct DeltaLakeClusterDefinition
 {
     static constexpr auto name = "deltaLakeCluster";