Merge pull request #61537 from ClickHouse/s3queue-disallow-sharded-processing

Disallow s3queue sharded mode

src/Core/Settings.h

@@ -114,6 +114,7 @@ class IColumn;
     M(Bool, enable_s3_requests_logging, false, "Enable very explicit logging of S3 requests. Makes sense for debug only.", 0) \
     M(String, s3queue_default_zookeeper_path, "/clickhouse/s3queue/", "Default zookeeper path prefix for S3Queue engine", 0) \
     M(Bool, s3queue_enable_logging_to_s3queue_log, false, "Enable writing to system.s3queue_log. The value can be overwritten per table with table settings", 0) \
+    M(Bool, s3queue_allow_experimental_sharded_mode, false, "Enable experimental sharded mode of S3Queue table engine. It is experimental because it will be rewritten", 0) \
     M(UInt64, hdfs_replication, 0, "The actual number of replications can be specified when the hdfs file is created.", 0) \
     M(Bool, hdfs_truncate_on_insert, false, "Enables or disables truncate before insert in s3 engine tables", 0) \
     M(Bool, hdfs_create_new_file_on_insert, false, "Enables or disables creating a new file on each insert in hdfs engine tables", 0) \

src/Core/SettingsChangesHistory.h

@@ -100,6 +100,7 @@ static std::map<ClickHouseVersion, SettingsChangesHistory::SettingsChanges> sett
               {"keeper_max_retries", 10, 10, "Max retries for general keeper operations"},
               {"keeper_retry_initial_backoff_ms", 100, 100, "Initial backoff timeout for general keeper operations"},
               {"keeper_retry_max_backoff_ms", 5000, 5000, "Max backoff timeout for general keeper operations"},
+              {"s3queue_allow_experimental_sharded_mode", false, false, "Enable experimental sharded mode of S3Queue table engine. It is experimental because it will be rewritten"},
               }},
     {"24.2", {{"allow_suspicious_variant_types", true, false, "Don't allow creating Variant type with suspicious variants by default"},
               {"validate_experimental_and_suspicious_types_inside_nested_types", false, true, "Validate usage of experimental and suspicious types inside nested types"},

src/Storages/S3Queue/StorageS3Queue.cpp

@@ -105,7 +105,8 @@ StorageS3Queue::StorageS3Queue(
     const String & comment,
     ContextPtr context_,
     std::optional<FormatSettings> format_settings_,
-    ASTStorage * engine_args)
+    ASTStorage * engine_args,
+    LoadingStrictnessLevel mode)
     : IStorage(table_id_)
     , WithContext(context_)
     , s3queue_settings(std::move(s3queue_settings_))
@@ -129,6 +130,14 @@ StorageS3Queue::StorageS3Queue(
         throw Exception(ErrorCodes::QUERY_NOT_ALLOWED, "S3Queue url must either end with '/' or contain globs");
     }
 
+    if (mode == LoadingStrictnessLevel::CREATE
+        && !context_->getSettingsRef().s3queue_allow_experimental_sharded_mode
+        && s3queue_settings->mode == S3QueueMode::ORDERED
+        && (s3queue_settings->s3queue_total_shards_num > 1 || s3queue_settings->s3queue_processing_threads_num > 1))
+    {
+        throw Exception(ErrorCodes::QUERY_NOT_ALLOWED, "S3Queue sharded mode is not allowed. To enable use `s3queue_allow_experimental_sharded_mode`");
+    }
+
     checkAndAdjustSettings(*s3queue_settings, context_->getSettingsRef());
 
     configuration.update(context_);
@@ -663,7 +672,8 @@ void registerStorageS3QueueImpl(const String & name, StorageFactory & factory)
                 args.comment,
                 args.getContext(),
                 format_settings,
-                args.storage_def);
+                args.storage_def,
+                args.mode);
         },
         {
             .supports_settings = true,

src/Storages/S3Queue/StorageS3Queue.h

@@ -32,7 +32,8 @@ public:
         const String & comment,
         ContextPtr context_,
         std::optional<FormatSettings> format_settings_,
-        ASTStorage * engine_args);
+        ASTStorage * engine_args,
+        LoadingStrictnessLevel mode);
 
     String getName() const override { return "S3Queue"; }
 

tests/integration/test_storage_s3_queue/configs/users.xml

@@ -3,6 +3,7 @@
         <default>
             <stream_like_engine_allow_direct_select>1</stream_like_engine_allow_direct_select>
             <s3queue_enable_logging_to_s3queue_log>1</s3queue_enable_logging_to_s3queue_log>
+            <s3queue_allow_experimental_sharded_mode>1</s3queue_allow_experimental_sharded_mode>
         </default>
     </profiles>
 </clickhouse>