Merge pull request #39039 from CurtizJ/randomize-more-settings

Add more settings for randomization

src/Compression/CachedCompressedReadBuffer.cpp

@@ -89,7 +89,7 @@ void CachedCompressedReadBuffer::seek(size_t offset_in_compressed_file, size_t o
 {
     /// Nothing to do if we already at required position
     if (!owned_cell && file_pos == offset_in_compressed_file
-        && (offset() == offset_in_decompressed_block ||
+        && ((!buffer().empty() && offset() == offset_in_decompressed_block) ||
             nextimpl_working_buffer_offset == offset_in_decompressed_block))
         return;
 

src/DataTypes/Serializations/SerializationSparse.cpp

@@ -263,6 +263,12 @@ void SerializationSparse::deserializeBinaryBulkWithMultipleStreams(
 {
     auto * state_sparse = checkAndGetState<DeserializeStateSparse>(state);
 
+    if (auto cached_column = getFromSubstreamsCache(cache, settings.path))
+    {
+        column = cached_column;
+        return;
+    }
+
     if (!settings.continuous_reading)
         state_sparse->reset();
 
@@ -281,7 +287,8 @@ void SerializationSparse::deserializeBinaryBulkWithMultipleStreams(
     size_t values_limit = offsets_data.size() - old_size;
 
     settings.path.back() = Substream::SparseElements;
-    nested->deserializeBinaryBulkWithMultipleStreams(values_column, values_limit, settings, state_sparse->nested, cache);
+    /// Do not use substream cache while reading values column, because ColumnSparse can be cached only in a whole.
+    nested->deserializeBinaryBulkWithMultipleStreams(values_column, values_limit, settings, state_sparse->nested, nullptr);
     settings.path.pop_back();
 
     if (offsets_data.size() + 1 != values_column->size())
@@ -291,6 +298,7 @@ void SerializationSparse::deserializeBinaryBulkWithMultipleStreams(
     /// 'insertManyDefaults' just increases size of column.
     column_sparse.insertManyDefaults(read_rows);
     column = std::move(mutable_column);
+    addToSubstreamsCache(cache, settings.path, column);
 }
 
 /// All methods below just wrap nested serialization.

src/Disks/IO/createReadBufferFromFileBase.cpp

@@ -52,7 +52,7 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(
     {
         try
         {
-            auto res = std::make_unique<MMapReadBufferFromFileWithCache>(*settings.mmap_cache, filename, 0);
+            auto res = std::make_unique<MMapReadBufferFromFileWithCache>(*settings.mmap_cache, filename, 0, file_size.value_or(-1));
             ProfileEvents::increment(ProfileEvents::CreatedReadBufferMMap);
             return res;
         }

src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp

@@ -1470,6 +1470,9 @@ MarkRanges MergeTreeDataSelectExecutor::markRangesFromPKRange(
     {
         // Do exclusion search, where we drop ranges that do not match
 
+        if (settings.merge_tree_coarse_index_granularity <= 1)
+            throw Exception(ErrorCodes::ARGUMENT_OUT_OF_BOUND, "Setting merge_tree_coarse_index_granularity should be greater than 1");
+
         size_t min_marks_for_seek = roundRowsOrBytesToMarks(
             settings.merge_tree_min_rows_for_seek,
             settings.merge_tree_min_bytes_for_seek,

src/Storages/MergeTree/MergeTreeWriteAheadLog.cpp

@@ -126,7 +126,7 @@ MergeTreeData::MutableDataPartsVector MergeTreeWriteAheadLog::restore(const Stor
     std::unique_lock lock(write_mutex);
 
     MergeTreeData::MutableDataPartsVector parts;
-    auto in = disk->readFile(path, {});
+    auto in = disk->readFile(path);
     NativeReader block_in(*in, 0);
     NameSet dropped_parts;
 

src/Storages/StorageLog.cpp

@@ -535,15 +535,16 @@ StorageLog::StorageLog(
     const ConstraintsDescription & constraints_,
     const String & comment,
     bool attach,
-    size_t max_compress_block_size_)
+    ContextMutablePtr context_)
     : IStorage(table_id_)
+    , WithMutableContext(context_)
     , engine_name(engine_name_)
     , disk(std::move(disk_))
     , table_path(relative_path_)
     , use_marks_file(engine_name == "Log")
     , marks_file_path(table_path + DBMS_STORAGE_LOG_MARKS_FILE_NAME)
     , file_checker(disk, table_path + "sizes.json")
-    , max_compress_block_size(max_compress_block_size_)
+    , max_compress_block_size(context_->getSettingsRef().max_compress_block_size)
 {
     StorageInMemoryMetadata storage_metadata;
     storage_metadata.setColumns(columns_);
@@ -750,9 +751,9 @@ static std::chrono::seconds getLockTimeout(ContextPtr context)
     return std::chrono::seconds{lock_timeout};
 }
 
-void StorageLog::truncate(const ASTPtr &, const StorageMetadataPtr &, ContextPtr context, TableExclusiveLockHolder &)
+void StorageLog::truncate(const ASTPtr &, const StorageMetadataPtr &, ContextPtr local_context, TableExclusiveLockHolder &)
 {
-    WriteLock lock{rwlock, getLockTimeout(context)};
+    WriteLock lock{rwlock, getLockTimeout(local_context)};
     if (!lock)
         throw Exception("Lock timeout exceeded", ErrorCodes::TIMEOUT_EXCEEDED);
 
@@ -769,6 +770,7 @@ void StorageLog::truncate(const ASTPtr &, const StorageMetadataPtr &, ContextPtr
 
     marks_loaded = true;
     num_marks_saved = 0;
+    getContext()->dropMMappedFileCache();
 }
 
 
@@ -776,14 +778,14 @@ Pipe StorageLog::read(
     const Names & column_names,
     const StorageSnapshotPtr & storage_snapshot,
     SelectQueryInfo & /*query_info*/,
-    ContextPtr context,
+    ContextPtr local_context,
     QueryProcessingStage::Enum /*processed_stage*/,
     size_t max_block_size,
     unsigned num_streams)
 {
     storage_snapshot->check(column_names);
 
-    auto lock_timeout = getLockTimeout(context);
+    auto lock_timeout = getLockTimeout(local_context);
     loadMarks(lock_timeout);
 
     ReadLock lock{rwlock, lock_timeout};
@@ -817,7 +819,7 @@ Pipe StorageLog::read(
     bool limited_by_file_sizes = !use_marks_file;
     size_t row_limit = std::numeric_limits<size_t>::max();
 
-    ReadSettings read_settings = context->getReadSettings();
+    ReadSettings read_settings = local_context->getReadSettings();
     Pipes pipes;
 
     for (size_t stream = 0; stream < num_streams; ++stream)
@@ -848,18 +850,18 @@ Pipe StorageLog::read(
     return Pipe::unitePipes(std::move(pipes));
 }
 
-SinkToStoragePtr StorageLog::write(const ASTPtr & /*query*/, const StorageMetadataPtr & metadata_snapshot, ContextPtr context)
+SinkToStoragePtr StorageLog::write(const ASTPtr & /*query*/, const StorageMetadataPtr & metadata_snapshot, ContextPtr local_context)
 {
-    WriteLock lock{rwlock, getLockTimeout(context)};
+    WriteLock lock{rwlock, getLockTimeout(local_context)};
     if (!lock)
         throw Exception("Lock timeout exceeded", ErrorCodes::TIMEOUT_EXCEEDED);
 
     return std::make_shared<LogSink>(*this, metadata_snapshot, std::move(lock));
 }
 
-CheckResults StorageLog::checkData(const ASTPtr & /* query */, ContextPtr context)
+CheckResults StorageLog::checkData(const ASTPtr & /* query */, ContextPtr local_context)
 {
-    ReadLock lock{rwlock, getLockTimeout(context)};
+    ReadLock lock{rwlock, getLockTimeout(local_context)};
     if (!lock)
         throw Exception("Lock timeout exceeded", ErrorCodes::TIMEOUT_EXCEEDED);
 
@@ -1114,7 +1116,7 @@ void registerStorageLog(StorageFactory & factory)
             args.constraints,
             args.comment,
             args.attach,
-            args.getContext()->getSettings().max_compress_block_size);
+            args.getContext());
     };
 
     factory.registerStorage("Log", create_fn, features);

src/Storages/StorageLog.h

@@ -12,6 +12,7 @@
 
 namespace DB
 {
+
 class IBackup;
 using BackupPtr = std::shared_ptr<const IBackup>;
 
@@ -21,7 +22,7 @@ using BackupPtr = std::shared_ptr<const IBackup>;
   * Also implements TinyLog - a table engine that is suitable for small chunks of the log.
   * It differs from Log in the absence of mark files.
   */
-class StorageLog final : public IStorage
+class StorageLog final : public IStorage, public WithMutableContext
 {
     friend class LogSource;
     friend class LogSink;
@@ -40,7 +41,7 @@ public:
         const ConstraintsDescription & constraints_,
         const String & comment,
         bool attach,
-        size_t max_compress_block_size_);
+        ContextMutablePtr context_);
 
     ~StorageLog() override;
     String getName() const override { return engine_name; }
@@ -49,16 +50,16 @@ public:
         const Names & column_names,
         const StorageSnapshotPtr & storage_snapshot,
         SelectQueryInfo & query_info,
-        ContextPtr context,
+        ContextPtr local_context,
         QueryProcessingStage::Enum processed_stage,
         size_t max_block_size,
         unsigned num_streams) override;
 
-    SinkToStoragePtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, ContextPtr context) override;
+    SinkToStoragePtr write(const ASTPtr & query, const StorageMetadataPtr & metadata_snapshot, ContextPtr local_context) override;
 
     void rename(const String & new_path_to_table_data, const StorageID & new_table_id) override;
 
-    CheckResults checkData(const ASTPtr & /* query */, ContextPtr /* context */) override;
+    CheckResults checkData(const ASTPtr & query, ContextPtr local_context) override;
 
     void truncate(const ASTPtr &, const StorageMetadataPtr &, ContextPtr, TableExclusiveLockHolder &) override;
 

src/Storages/StorageStripeLog.cpp

@@ -265,14 +265,15 @@ StorageStripeLog::StorageStripeLog(
     const ConstraintsDescription & constraints_,
     const String & comment,
     bool attach,
-    size_t max_compress_block_size_)
+    ContextMutablePtr context_)
     : IStorage(table_id_)
+    , WithMutableContext(context_)
     , disk(std::move(disk_))
     , table_path(relative_path_)
     , data_file_path(table_path + "data.bin")
     , index_file_path(table_path + "index.mrk")
     , file_checker(disk, table_path + "sizes.json")
-    , max_compress_block_size(max_compress_block_size_)
+    , max_compress_block_size(context_->getSettings().max_compress_block_size)
     , log(&Poco::Logger::get("StorageStripeLog"))
 {
     StorageInMemoryMetadata storage_metadata;
@@ -330,9 +331,9 @@ void StorageStripeLog::rename(const String & new_path_to_table_data, const Stora
 }
 
 
-static std::chrono::seconds getLockTimeout(ContextPtr context)
+static std::chrono::seconds getLockTimeout(ContextPtr local_context)
 {
-    const Settings & settings = context->getSettingsRef();
+    const Settings & settings = local_context->getSettingsRef();
     Int64 lock_timeout = settings.lock_acquire_timeout.totalSeconds();
     if (settings.max_execution_time.totalSeconds() != 0 && settings.max_execution_time.totalSeconds() < lock_timeout)
         lock_timeout = settings.max_execution_time.totalSeconds();
@@ -344,14 +345,14 @@ Pipe StorageStripeLog::read(
     const Names & column_names,
     const StorageSnapshotPtr & storage_snapshot,
     SelectQueryInfo & /*query_info*/,
-    ContextPtr context,
+    ContextPtr local_context,
     QueryProcessingStage::Enum /*processed_stage*/,
     const size_t /*max_block_size*/,
     unsigned num_streams)
 {
     storage_snapshot->check(column_names);
 
-    auto lock_timeout = getLockTimeout(context);
+    auto lock_timeout = getLockTimeout(local_context);
     loadIndices(lock_timeout);
 
     ReadLock lock{rwlock, lock_timeout};
@@ -369,7 +370,7 @@ Pipe StorageStripeLog::read(
     if (num_streams > size)
         num_streams = size;
 
-    ReadSettings read_settings = context->getReadSettings();
+    ReadSettings read_settings = local_context->getReadSettings();
     Pipes pipes;
 
     for (size_t stream = 0; stream < num_streams; ++stream)
@@ -390,9 +391,9 @@ Pipe StorageStripeLog::read(
 }
 
 
-SinkToStoragePtr StorageStripeLog::write(const ASTPtr & /*query*/, const StorageMetadataPtr & metadata_snapshot, ContextPtr context)
+SinkToStoragePtr StorageStripeLog::write(const ASTPtr & /*query*/, const StorageMetadataPtr & metadata_snapshot, ContextPtr local_context)
 {
-    WriteLock lock{rwlock, getLockTimeout(context)};
+    WriteLock lock{rwlock, getLockTimeout(local_context)};
     if (!lock)
         throw Exception("Lock timeout exceeded", ErrorCodes::TIMEOUT_EXCEEDED);
 
@@ -400,9 +401,9 @@ SinkToStoragePtr StorageStripeLog::write(const ASTPtr & /*query*/, const Storage
 }
 
 
-CheckResults StorageStripeLog::checkData(const ASTPtr & /* query */, ContextPtr context)
+CheckResults StorageStripeLog::checkData(const ASTPtr & /* query */, ContextPtr local_context)
 {
-    ReadLock lock{rwlock, getLockTimeout(context)};
+    ReadLock lock{rwlock, getLockTimeout(local_context)};
     if (!lock)
         throw Exception("Lock timeout exceeded", ErrorCodes::TIMEOUT_EXCEEDED);
 
@@ -420,6 +421,7 @@ void StorageStripeLog::truncate(const ASTPtr &, const StorageMetadataPtr &, Cont
 
     indices_loaded = true;
     num_indices_saved = 0;
+    getContext()->dropMMappedFileCache();
 }
 
 
@@ -686,7 +688,7 @@ void registerStorageStripeLog(StorageFactory & factory)
             args.constraints,
             args.comment,
             args.attach,
-            args.getContext()->getSettings().max_compress_block_size);
+            args.getContext());
     }, features);
 }
 

src/Storages/StorageStripeLog.h

@@ -20,7 +20,7 @@ using BackupPtr = std::shared_ptr<const IBackup>;
 /** Implements a table engine that is suitable for small chunks of the log.
   * In doing so, stores all the columns in a single Native file, with a nearby index.
   */
-class StorageStripeLog final : public IStorage
+class StorageStripeLog final : public IStorage, public WithMutableContext
 {
 friend class StripeLogSource;
 friend class StripeLogSink;
@@ -34,7 +34,7 @@ public:
         const ConstraintsDescription & constraints_,
         const String & comment,
         bool attach,
-        size_t max_compress_block_size_);
+        ContextMutablePtr context_);
 
     ~StorageStripeLog() override;
 
@@ -44,16 +44,16 @@ public:
         const Names & column_names,
         const StorageSnapshotPtr & storage_snapshot,
         SelectQueryInfo & query_info,
-        ContextPtr context,
+        ContextPtr local_context,
         QueryProcessingStage::Enum processed_stage,
         size_t max_block_size,
         unsigned num_streams) override;
 
-    SinkToStoragePtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, ContextPtr context) override;
+    SinkToStoragePtr write(const ASTPtr & query, const StorageMetadataPtr & metadata_snapshot, ContextPtr local_context) override;
 
     void rename(const String & new_path_to_table_data, const StorageID & new_table_id) override;
 
-    CheckResults checkData(const ASTPtr & /* query */, ContextPtr /* context */) override;
+    CheckResults checkData(const ASTPtr & query, ContextPtr ocal_context) override;
 
     bool storesDataOnDisk() const override { return true; }
     Strings getDataPaths() const override { return {DB::fullPath(disk, table_path)}; }

src/Storages/tests/gtest_storage_log.cpp

@@ -40,7 +40,7 @@ DB::StoragePtr createStorage(DB::DiskPtr & disk)
 
     StoragePtr table = std::make_shared<StorageLog>(
         "Log", disk, "table/", StorageID("test", "test"), ColumnsDescription{names_and_types},
-        ConstraintsDescription{}, String{}, false, static_cast<size_t>(1048576));
+        ConstraintsDescription{}, String{}, false, getContext().context);
 
     table->startup();
 

tests/clickhouse-test

@@ -426,6 +426,27 @@ class SettingsRandomizer:
         "read_in_order_two_level_merge_threshold": lambda: random.randint(0, 100),
         "optimize_aggregation_in_order": lambda: random.randint(0, 1),
         "aggregation_in_order_max_block_bytes": lambda: random.randint(0, 50000000),
+        "use_uncompressed_cache": lambda: random.randint(0, 1),
+        "min_bytes_to_use_direct_io": lambda: 0
+        if random.random() < 0.5
+        else 1
+        if random.random() < 0.2
+        else random.randint(1, 1024 * 1024 * 1024),
+        "min_bytes_to_use_mmap_io": lambda: 0
+        if random.random() < 0.5
+        else 1
+        if random.random() < 0.2
+        else random.randint(1, 1024 * 1024 * 1024),
+        "local_filesystem_read_method": lambda: random.choice(
+            ["read", "pread", "mmap", "pread_threadpool"]
+        ),
+        "remote_filesystem_read_method": lambda: random.choice(["read", "threadpool"]),
+        "local_filesystem_read_prefetch": lambda: random.randint(0, 1),
+        "remote_filesystem_read_prefetch": lambda: random.randint(0, 1),
+        "compile_expressions": lambda: random.randint(0, 1),
+        "compile_aggregate_expressions": lambda: random.randint(0, 1),
+        "compile_sort_description": lambda: random.randint(0, 1),
+        "merge_tree_coarse_index_granularity": lambda: random.randint(2, 32),
     }
 
     @staticmethod
@@ -710,7 +731,9 @@ class TestCase:
 
         return None
 
-    def process_result_impl(self, proc, stdout: str, stderr: str, debug_log: str, total_time: float):
+    def process_result_impl(
+        self, proc, stdout: str, stderr: str, debug_log: str, total_time: float
+    ):
         description = ""
 
         if proc:
@@ -1038,7 +1061,9 @@ class TestCase:
                 server_logs_level, client_options
             )
 
-            result = self.process_result_impl(proc, stdout, stderr, debug_log, total_time)
+            result = self.process_result_impl(
+                proc, stdout, stderr, debug_log, total_time
+            )
             result.check_if_need_retry(args, stdout, stderr, self.runs_count)
             if result.status == TestStatus.FAIL:
                 result.description = self.add_info_about_settings(
@@ -1530,7 +1555,8 @@ def collect_build_flags(args):
         result.append(BuildFlags.RELEASE)
 
     value = clickhouse_execute(
-        args, "SELECT value FROM system.settings WHERE name = 'allow_deprecated_database_ordinary'"
+        args,
+        "SELECT value FROM system.settings WHERE name = 'allow_deprecated_database_ordinary'",
     )
     if value == b"1":
         result.append(BuildFlags.ORDINARY_DATABASE)
@@ -1634,7 +1660,9 @@ def do_run_tests(jobs, test_suite: TestSuite, parallel):
 
                 queue.close()
         except Full:
-            print("Couldn't put test to the queue within timeout. Server probably hung.")
+            print(
+                "Couldn't put test to the queue within timeout. Server probably hung."
+            )
             print_stacktraces()
             queue.close()
 

tests/queries/0_stateless/01475_read_subcolumns.sql

@@ -1,4 +1,4 @@
--- Tags: no-s3-storage
+-- Tags: no-s3-storage, no-random-settings
 
 SET use_uncompressed_cache = 0;
 

tests/queries/0_stateless/01533_multiple_nested.sql

@@ -4,6 +4,7 @@ DROP TABLE IF EXISTS nested;
 
 SET flatten_nested = 0;
 SET use_uncompressed_cache = 0;
+SET local_filesystem_read_method='pread';
 
 CREATE TABLE nested
 (

tests/queries/0_stateless/01780_column_sparse_tuple.sql

@@ -1,4 +1,3 @@
--- Tags: no-s3-storage
 DROP TABLE IF EXISTS sparse_tuple;
 
 CREATE TABLE sparse_tuple (id UInt64, t Tuple(a UInt64, s String))

tests/queries/0_stateless/02226_s3_with_cache.sql

@@ -1,4 +1,4 @@
--- Tags: no-parallel, no-fasttest, long
+-- Tags: no-parallel, no-fasttest, long, no-random-settings
 
 SET max_memory_usage='20G';
 SET enable_filesystem_cache_on_write_operations = 0;

tests/queries/0_stateless/02240_system_remote_filesystem_cache.sql

@@ -1,4 +1,4 @@
--- Tags: no-parallel, no-fasttest, no-s3-storage
+-- Tags: no-parallel, no-fasttest, no-s3-storage, no-random-settings
 
 -- { echo }
 

tests/queries/0_stateless/02240_system_remote_filesystem_cache_log.sql

@@ -1,4 +1,4 @@
--- Tags: no-parallel, no-fasttest, no-s3-storage
+-- Tags: no-parallel, no-fasttest, no-s3-storage, no-random-settings
 
 -- { echo }
 

tests/queries/0_stateless/02240_system_remote_filesystem_query_cache.sql

@@ -1,4 +1,4 @@
--- Tags: no-parallel, no-fasttest, no-s3-storage
+-- Tags: no-parallel, no-fasttest, no-s3-storage, no-random-settings
 
 -- { echo }
 

tests/queries/0_stateless/02286_drop_filesystem_cache.sql

@@ -1,4 +1,4 @@
--- Tags: no-parallel, no-fasttest, no-s3-storage
+-- Tags: no-parallel, no-fasttest, no-s3-storage, no-random-settings
 
 -- { echo }