diff --git a/cmake/autogenerated_versions.txt b/cmake/autogenerated_versions.txt index 1f338de77e60..c65fb52d1e2e 100644 --- a/cmake/autogenerated_versions.txt +++ b/cmake/autogenerated_versions.txt @@ -2,13 +2,13 @@ # NOTE: VERSION_REVISION has nothing common with DBMS_TCP_PROTOCOL_VERSION, # only DBMS_TCP_PROTOCOL_VERSION should be incremented on protocol changes. -SET(VERSION_REVISION 54530) +SET(VERSION_REVISION 54533) SET(VERSION_MAJOR 25) SET(VERSION_MINOR 8) -SET(VERSION_PATCH 30) -SET(VERSION_GITHASH 54df9137dcfc1ef0b307f264199e004a05a9a5dd) -SET(VERSION_DESCRIBE v25.8.30.20001.altinityantalya) -SET(VERSION_STRING 25.8.30.20001.altinityantalya) +SET(VERSION_PATCH 33) +SET(VERSION_GITHASH 46297556f3ac83e1a3ee4a05cc31f4287f5e17ac) +SET(VERSION_DESCRIBE v25.8.33.20001.altinityantalya) +SET(VERSION_STRING 25.8.33.20001.altinityantalya) # end of autochange SET(VERSION_TWEAK 20001) diff --git a/contrib/libssh b/contrib/libssh index 47305a2f7257..50313883f3a0 160000 --- a/contrib/libssh +++ b/contrib/libssh @@ -1 +1 @@ -Subproject commit 47305a2f7257b56ca407260a72af85db058d551f +Subproject commit 50313883f3a077458cde4ea95bf46bfeb0771b34 diff --git a/contrib/libssh-cmake/CMakeLists.txt b/contrib/libssh-cmake/CMakeLists.txt index 0e1fae5880af..e6d58878f840 100644 --- a/contrib/libssh-cmake/CMakeLists.txt +++ b/contrib/libssh-cmake/CMakeLists.txt @@ -7,8 +7,8 @@ endif() # CMake variables needed by libssh_version.h.cmake, update them when you update libssh set(libssh_VERSION_MAJOR 0) -set(libssh_VERSION_MINOR 9) -set(libssh_VERSION_PATCH 8) +set(libssh_VERSION_MINOR 12) +set(libssh_VERSION_PATCH 0) set(LIB_SOURCE_DIR "${ClickHouse_SOURCE_DIR}/contrib/libssh") set(LIB_BINARY_DIR "${ClickHouse_BINARY_DIR}/contrib/libssh") @@ -37,6 +37,7 @@ set(libssh_SRCS ${LIB_SOURCE_DIR}/src/external/poly1305.c ${LIB_SOURCE_DIR}/src/external/sntrup761.c ${LIB_SOURCE_DIR}/src/getpass.c + ${LIB_SOURCE_DIR}/src/hybrid_mlkem.c ${LIB_SOURCE_DIR}/src/init.c ${LIB_SOURCE_DIR}/src/kdf.c ${LIB_SOURCE_DIR}/src/kex.c @@ -47,6 +48,7 @@ set(libssh_SRCS ${LIB_SOURCE_DIR}/src/match.c ${LIB_SOURCE_DIR}/src/messages.c ${LIB_SOURCE_DIR}/src/misc.c + ${LIB_SOURCE_DIR}/src/mlkem.c ${LIB_SOURCE_DIR}/src/options.c ${LIB_SOURCE_DIR}/src/packet.c ${LIB_SOURCE_DIR}/src/packet_cb.c @@ -79,6 +81,7 @@ set(libssh_SRCS ${LIB_SOURCE_DIR}/src/gzip.c ${LIB_SOURCE_DIR}/src/libcrypto.c ${LIB_SOURCE_DIR}/src/md_crypto.c + ${LIB_SOURCE_DIR}/src/mlkem_crypto.c ${LIB_SOURCE_DIR}/src/pki_crypto.c ${LIB_SOURCE_DIR}/src/pki_context.c ${LIB_SOURCE_DIR}/src/sntrup761.c diff --git a/contrib/libssh-cmake/darwin/config.h b/contrib/libssh-cmake/darwin/config.h index 12378a64ceaa..f748858a9055 100644 --- a/contrib/libssh-cmake/darwin/config.h +++ b/contrib/libssh-cmake/darwin/config.h @@ -14,6 +14,18 @@ /* Global client configuration file path */ #define GLOBAL_CLIENT_CONFIG "/etc/ssh/ssh_config" +/* Global configuration directory (libssh >= 0.12 uses it unconditionally) */ +#define GLOBAL_CONF_DIR "/etc/ssh" + +/* Define to 1 if we have support for ML-KEM in libgcrypt */ +/* #undef HAVE_GCRYPT_MLKEM */ + +/* Define to 1 if we have support for ML-KEM in OpenSSL */ +#define HAVE_OPENSSL_MLKEM 1 + +/* Define to 1 if we have support for ML-KEM1024 in either backend */ +#define HAVE_MLKEM1024 1 + /************************** HEADER FILES *************************/ /* Define to 1 if you have the header file. */ diff --git a/contrib/libssh-cmake/freebsd/config.h b/contrib/libssh-cmake/freebsd/config.h index 8a70acb473c0..857bdef90e6f 100644 --- a/contrib/libssh-cmake/freebsd/config.h +++ b/contrib/libssh-cmake/freebsd/config.h @@ -14,6 +14,18 @@ /* Global client configuration file path */ #define GLOBAL_CLIENT_CONFIG "/etc/ssh/ssh_config" +/* Global configuration directory (libssh >= 0.12 uses it unconditionally) */ +#define GLOBAL_CONF_DIR "/etc/ssh" + +/* Define to 1 if we have support for ML-KEM in libgcrypt */ +/* #undef HAVE_GCRYPT_MLKEM */ + +/* Define to 1 if we have support for ML-KEM in OpenSSL */ +#define HAVE_OPENSSL_MLKEM 1 + +/* Define to 1 if we have support for ML-KEM1024 in either backend */ +#define HAVE_MLKEM1024 1 + /************************** HEADER FILES *************************/ /* Define to 1 if you have the header file. */ diff --git a/contrib/libssh-cmake/linux/aarch64-musl/config.h b/contrib/libssh-cmake/linux/aarch64-musl/config.h index 9cc21c1df4ad..30620274c4fb 100644 --- a/contrib/libssh-cmake/linux/aarch64-musl/config.h +++ b/contrib/libssh-cmake/linux/aarch64-musl/config.h @@ -14,6 +14,18 @@ /* Global client configuration file path */ #define GLOBAL_CLIENT_CONFIG "/etc/ssh/ssh_config" +/* Global configuration directory (libssh >= 0.12 uses it unconditionally) */ +#define GLOBAL_CONF_DIR "/etc/ssh" + +/* Define to 1 if we have support for ML-KEM in libgcrypt */ +/* #undef HAVE_GCRYPT_MLKEM */ + +/* Define to 1 if we have support for ML-KEM in OpenSSL */ +#define HAVE_OPENSSL_MLKEM 1 + +/* Define to 1 if we have support for ML-KEM1024 in either backend */ +#define HAVE_MLKEM1024 1 + /************************** HEADER FILES *************************/ /* Define to 1 if you have the header file. */ diff --git a/contrib/libssh-cmake/linux/aarch64/config.h b/contrib/libssh-cmake/linux/aarch64/config.h index 7e21b1b4f683..bfab9f7f49c4 100644 --- a/contrib/libssh-cmake/linux/aarch64/config.h +++ b/contrib/libssh-cmake/linux/aarch64/config.h @@ -14,6 +14,18 @@ /* Global client configuration file path */ #define GLOBAL_CLIENT_CONFIG "/etc/ssh/ssh_config" +/* Global configuration directory (libssh >= 0.12 uses it unconditionally) */ +#define GLOBAL_CONF_DIR "/etc/ssh" + +/* Define to 1 if we have support for ML-KEM in libgcrypt */ +/* #undef HAVE_GCRYPT_MLKEM */ + +/* Define to 1 if we have support for ML-KEM in OpenSSL */ +#define HAVE_OPENSSL_MLKEM 1 + +/* Define to 1 if we have support for ML-KEM1024 in either backend */ +#define HAVE_MLKEM1024 1 + /************************** HEADER FILES *************************/ /* Define to 1 if you have the header file. */ diff --git a/contrib/libssh-cmake/linux/loongarch64/config.h b/contrib/libssh-cmake/linux/loongarch64/config.h index 3e19e6ab945d..1856298aa39d 100644 --- a/contrib/libssh-cmake/linux/loongarch64/config.h +++ b/contrib/libssh-cmake/linux/loongarch64/config.h @@ -14,6 +14,18 @@ /* Global client configuration file path */ #define GLOBAL_CLIENT_CONFIG "/etc/ssh/ssh_config" +/* Global configuration directory (libssh >= 0.12 uses it unconditionally) */ +#define GLOBAL_CONF_DIR "/etc/ssh" + +/* Define to 1 if we have support for ML-KEM in libgcrypt */ +/* #undef HAVE_GCRYPT_MLKEM */ + +/* Define to 1 if we have support for ML-KEM in OpenSSL */ +#define HAVE_OPENSSL_MLKEM 1 + +/* Define to 1 if we have support for ML-KEM1024 in either backend */ +#define HAVE_MLKEM1024 1 + /************************** HEADER FILES *************************/ /* Define to 1 if you have the header file. */ diff --git a/contrib/libssh-cmake/linux/ppc64le/config.h b/contrib/libssh-cmake/linux/ppc64le/config.h index 701ee00416bb..ae317dff7d1f 100644 --- a/contrib/libssh-cmake/linux/ppc64le/config.h +++ b/contrib/libssh-cmake/linux/ppc64le/config.h @@ -14,6 +14,18 @@ /* Global client configuration file path */ #define GLOBAL_CLIENT_CONFIG "/etc/ssh/ssh_config" +/* Global configuration directory (libssh >= 0.12 uses it unconditionally) */ +#define GLOBAL_CONF_DIR "/etc/ssh" + +/* Define to 1 if we have support for ML-KEM in libgcrypt */ +/* #undef HAVE_GCRYPT_MLKEM */ + +/* Define to 1 if we have support for ML-KEM in OpenSSL */ +#define HAVE_OPENSSL_MLKEM 1 + +/* Define to 1 if we have support for ML-KEM1024 in either backend */ +#define HAVE_MLKEM1024 1 + /************************** HEADER FILES *************************/ /* Define to 1 if you have the header file. */ diff --git a/contrib/libssh-cmake/linux/riscv64/config.h b/contrib/libssh-cmake/linux/riscv64/config.h index ca0868072bdc..5e2ae515edfd 100644 --- a/contrib/libssh-cmake/linux/riscv64/config.h +++ b/contrib/libssh-cmake/linux/riscv64/config.h @@ -14,6 +14,18 @@ /* Global client configuration file path */ #define GLOBAL_CLIENT_CONFIG "/etc/ssh/ssh_config" +/* Global configuration directory (libssh >= 0.12 uses it unconditionally) */ +#define GLOBAL_CONF_DIR "/etc/ssh" + +/* Define to 1 if we have support for ML-KEM in libgcrypt */ +/* #undef HAVE_GCRYPT_MLKEM */ + +/* Define to 1 if we have support for ML-KEM in OpenSSL */ +#define HAVE_OPENSSL_MLKEM 1 + +/* Define to 1 if we have support for ML-KEM1024 in either backend */ +#define HAVE_MLKEM1024 1 + /************************** HEADER FILES *************************/ /* Define to 1 if you have the header file. */ diff --git a/contrib/libssh-cmake/linux/s390x/config.h b/contrib/libssh-cmake/linux/s390x/config.h index 6c284c9d6d4f..ebc39273e5b3 100644 --- a/contrib/libssh-cmake/linux/s390x/config.h +++ b/contrib/libssh-cmake/linux/s390x/config.h @@ -14,6 +14,18 @@ /* Global client configuration file path */ #define GLOBAL_CLIENT_CONFIG "/etc/ssh/ssh_config" +/* Global configuration directory (libssh >= 0.12 uses it unconditionally) */ +#define GLOBAL_CONF_DIR "/etc/ssh" + +/* Define to 1 if we have support for ML-KEM in libgcrypt */ +/* #undef HAVE_GCRYPT_MLKEM */ + +/* Define to 1 if we have support for ML-KEM in OpenSSL */ +#define HAVE_OPENSSL_MLKEM 1 + +/* Define to 1 if we have support for ML-KEM1024 in either backend */ +#define HAVE_MLKEM1024 1 + /************************** HEADER FILES *************************/ /* Define to 1 if you have the header file. */ diff --git a/contrib/libssh-cmake/linux/x86-64-musl/config.h b/contrib/libssh-cmake/linux/x86-64-musl/config.h index 9b325957c358..7a50d2df123f 100644 --- a/contrib/libssh-cmake/linux/x86-64-musl/config.h +++ b/contrib/libssh-cmake/linux/x86-64-musl/config.h @@ -14,6 +14,18 @@ /* Global client configuration file path */ #define GLOBAL_CLIENT_CONFIG "/etc/ssh/ssh_config" +/* Global configuration directory (libssh >= 0.12 uses it unconditionally) */ +#define GLOBAL_CONF_DIR "/etc/ssh" + +/* Define to 1 if we have support for ML-KEM in libgcrypt */ +/* #undef HAVE_GCRYPT_MLKEM */ + +/* Define to 1 if we have support for ML-KEM in OpenSSL */ +#define HAVE_OPENSSL_MLKEM 1 + +/* Define to 1 if we have support for ML-KEM1024 in either backend */ +#define HAVE_MLKEM1024 1 + /************************** HEADER FILES *************************/ /* Define to 1 if you have the header file. */ diff --git a/contrib/libssh-cmake/linux/x86-64/config.h b/contrib/libssh-cmake/linux/x86-64/config.h index f6316af195cd..181ece488eb5 100644 --- a/contrib/libssh-cmake/linux/x86-64/config.h +++ b/contrib/libssh-cmake/linux/x86-64/config.h @@ -14,6 +14,18 @@ /* Global client configuration file path */ #define GLOBAL_CLIENT_CONFIG "/etc/ssh/ssh_config" +/* Global configuration directory (libssh >= 0.12 uses it unconditionally) */ +#define GLOBAL_CONF_DIR "/etc/ssh" + +/* Define to 1 if we have support for ML-KEM in libgcrypt */ +/* #undef HAVE_GCRYPT_MLKEM */ + +/* Define to 1 if we have support for ML-KEM in OpenSSL */ +#define HAVE_OPENSSL_MLKEM 1 + +/* Define to 1 if we have support for ML-KEM1024 in either backend */ +#define HAVE_MLKEM1024 1 + /************************** HEADER FILES *************************/ /* Define to 1 if you have the header file. */ @@ -188,8 +200,6 @@ /* Define to 1 if we have support for blowfish */ /* #undef HAVE_BLOWFISH */ -/* Define to 1 if we have support for ML-KEM */ -/* #undef HAVE_MLKEM */ /*************************** LIBRARIES ***************************/ diff --git a/contrib/postgres-cmake/CMakeLists.txt b/contrib/postgres-cmake/CMakeLists.txt index 46d7af59c870..6c35ee13f7a4 100644 --- a/contrib/postgres-cmake/CMakeLists.txt +++ b/contrib/postgres-cmake/CMakeLists.txt @@ -66,6 +66,12 @@ if(NOT OS_DARWIN) ) endif() +if(OS_DARWIN OR ARCH_PPC64LE) + set(SRCS ${SRCS} + "${POSTGRES_SOURCE_DIR}/src/port/explicit_bzero.c" + ) +endif() + add_library(_libpq ${SRCS}) add_definitions(-DFRONTEND) diff --git a/src/Analyzer/Passes/UniqInjectiveFunctionsEliminationPass.cpp b/src/Analyzer/Passes/UniqInjectiveFunctionsEliminationPass.cpp index 39d88491553d..6717f98c0b7e 100644 --- a/src/Analyzer/Passes/UniqInjectiveFunctionsEliminationPass.cpp +++ b/src/Analyzer/Passes/UniqInjectiveFunctionsEliminationPass.cpp @@ -11,6 +11,8 @@ #include +#include + namespace DB { @@ -66,6 +68,12 @@ class UniqInjectiveFunctionsEliminationVisitor : public InDepthQueryTreeVisitorW if (!arg_function->isInjective({})) return false; + /// The `Null` combinator makes `uniq*` skip rows where a Nullable argument is NULL: `uniq(tuple(x))` + /// counts the (NULL) row while `uniq(x)` skips it. + if (isNullableOrLowCardinalityNullable(arg->getResultType()) + != isNullableOrLowCardinalityNullable(arg_arguments_nodes[0]->getResultType())) + return false; + arg = arg_arguments_nodes[0]; return replaced_argument = true; }; diff --git a/src/Backups/BackupImpl.cpp b/src/Backups/BackupImpl.cpp index 724d20f6221e..edc93ffc8b19 100644 --- a/src/Backups/BackupImpl.cpp +++ b/src/Backups/BackupImpl.cpp @@ -954,18 +954,71 @@ String BackupImpl::getObjectKey(const String & file_name) const } size_t BackupImpl::copyFileToDisk(const String & file_name, - DiskPtr destination_disk, const String & destination_path, WriteMode write_mode) const + DiskPtr destination_disk, const String & destination_path, WriteMode write_mode, bool sync) const { -#if CLICKHOUSE_CLOUD String object_key = getObjectKey(file_name); if (!object_key.empty()) + { + /// The optimized object-key copy exposes no buffer to fsync, so the sync case needs a buffered path. + if (sync) + return copyObjectKeyEntryToDiskSynced(object_key, destination_disk, destination_path, write_mode); +#if CLICKHOUSE_CLOUD return copyFileToDiskByObjectKey(object_key, destination_disk, destination_path, write_mode); #endif - return copyFileToDisk(getFileSizeAndChecksum(file_name), destination_disk, destination_path, write_mode); + } + return copyFileToDisk(getFileSizeAndChecksum(file_name), destination_disk, destination_path, write_mode, sync); +} + +size_t BackupImpl::copyObjectKeyEntryToDiskSynced( + const String & object_key, DiskPtr destination_disk, const String & destination_path, WriteMode write_mode) const +{ + if (open_mode == OpenMode::WRITE) + throw Exception(ErrorCodes::LOGICAL_ERROR, "The backup file should not be opened for writing. Something is wrong internally"); + + BackupFileInfo info; + { + std::lock_guard lock{mutex}; + auto it = lightweight_snapshot_file_infos.find(object_key); + if (it == lightweight_snapshot_file_infos.end()) + throw Exception( + ErrorCodes::BACKUP_ENTRY_NOT_FOUND, + "Backup {}: Entry with object key {} not found in the backup", + backup_name_for_logging, object_key); + info = it->second; + } + + if (info.encrypted_by_disk && !destination_disk->getDataSourceDescription().is_encrypted) + { + throw Exception( + ErrorCodes::CANNOT_RESTORE_TO_NONENCRYPTED_DISK, + "File {} is encrypted in the backup, it can be restored only to an encrypted disk", + info.data_file_name); + } + + auto read_buffer = readFileByObjectKey(info); + size_t buf_size = std::min(info.size ? info.size : DBMS_DEFAULT_BUFFER_SIZE, reader->getWriteBufferSize()); + std::unique_ptr write_buffer; + /// readFileByObjectKey returns the bytes as stored (still encrypted for encrypted-by-disk entries), + /// so write them through writeEncryptedFile to avoid re-encrypting, mirroring the generic copy path. + if (info.encrypted_by_disk) + write_buffer = destination_disk->writeEncryptedFile(destination_path, buf_size, write_mode, reader->getWriteSettings()); + else + write_buffer = destination_disk->writeFile(destination_path, buf_size, write_mode, reader->getWriteSettings()); + copyData(*read_buffer, *write_buffer, info.size); + write_buffer->finalize(); + /// fdatasync the contents so a restored part survives power loss (see copyFileToDisk above). + write_buffer->sync(); + + { + std::lock_guard lock{mutex}; + ++num_read_files; + num_read_bytes += info.size; + } + return info.size; } size_t BackupImpl::copyFileToDisk(const SizeAndChecksum & size_and_checksum, - DiskPtr destination_disk, const String & destination_path, WriteMode write_mode) const + DiskPtr destination_disk, const String & destination_path, WriteMode write_mode, bool sync) const { if (open_mode == OpenMode::WRITE) throw Exception(ErrorCodes::LOGICAL_ERROR, "The backup file should not be opened for writing. Something is wrong internally"); @@ -975,8 +1028,18 @@ size_t BackupImpl::copyFileToDisk(const SizeAndChecksum & size_and_checksum, /// Entry's data is empty. if (write_mode == WriteMode::Rewrite) { - /// Just create an empty file. - destination_disk->createFile(destination_path); + if (sync) + { + /// createFile() leaves the empty contents unsynced; a live buffer lets us fsync it. + auto write_buffer = destination_disk->writeFile(destination_path, DBMS_DEFAULT_BUFFER_SIZE, write_mode, reader->getWriteSettings()); + write_buffer->finalize(); + write_buffer->sync(); + } + else + { + /// Just create an empty file. + destination_disk->createFile(destination_path); + } } std::lock_guard lock{mutex}; ++num_read_files; @@ -1008,16 +1071,25 @@ size_t BackupImpl::copyFileToDisk(const SizeAndChecksum & size_and_checksum, bool file_copied = false; - if (info.size && !info.base_size && !use_archive) + /// When `sync` is requested we must copy through a live destination buffer so we can fsync its + /// contents below. The optimized delegate paths (reader->copyFileToDisk / base backup) may use + /// fs::copy or an object-storage copy and expose no buffer, so skip them and take the buffered + /// branch, which is already correct for every source (this backup, base backup, archive). + if (!sync && info.size && !info.base_size && !use_archive) { - /// Data comes completely from this backup. + /// Data comes completely from this backup. The reader copies without exposing a write + /// buffer we could fsync, so this fast path is used only when `sync` isn't requested. reader->copyFileToDisk(info.data_file_name, info.size, info.encrypted_by_disk, destination_disk, destination_path, write_mode); file_copied = true; } else if (info.size && (info.size == info.base_size)) { - /// Data comes completely from the base backup (nothing comes from this backup). - getBaseBackup()->copyFileToDisk(std::pair{info.base_size, info.base_checksum}, destination_disk, destination_path, write_mode); + /// Data comes completely from the base backup (nothing comes from this backup). The base + /// backup is itself a BackupImpl that honours `sync` and can read its own encrypted-by-disk + /// entries, so forward the copy (and the `sync` request) there. Going through the generic + /// branch below instead would read the base via the public readFile(), which always requests + /// unencrypted data and would fail on an encrypted entry (CANNOT_RESTORE_TO_NONENCRYPTED_DISK). + getBaseBackup()->copyFileToDisk(std::pair{info.base_size, info.base_checksum}, destination_disk, destination_path, write_mode, sync); file_copied = true; } @@ -1032,7 +1104,7 @@ size_t BackupImpl::copyFileToDisk(const SizeAndChecksum & size_and_checksum, { /// Use the generic way to copy data. `readFile()` will update `num_read_files`. auto read_buffer = readFileImpl(info.file_name, size_and_checksum, /* read_encrypted= */ info.encrypted_by_disk); - std::unique_ptr write_buffer; + std::unique_ptr write_buffer; size_t buf_size = std::min(info.size, reader->getWriteBufferSize()); if (info.encrypted_by_disk) write_buffer = destination_disk->writeEncryptedFile(destination_path, buf_size, write_mode, reader->getWriteSettings()); @@ -1040,6 +1112,10 @@ size_t BackupImpl::copyFileToDisk(const SizeAndChecksum & size_and_checksum, write_buffer = destination_disk->writeFile(destination_path, buf_size, write_mode, reader->getWriteSettings()); copyData(*read_buffer, *write_buffer, info.size); write_buffer->finalize(); + /// fdatasync the contents so a restored part survives power loss, matching the durability + /// an inserted part gets from fsync_after_insert (the caller passes `sync` accordingly). + if (sync) + write_buffer->sync(); } return info.size; diff --git a/src/Backups/BackupImpl.h b/src/Backups/BackupImpl.h index 8c04776f26fd..74fa3573701d 100644 --- a/src/Backups/BackupImpl.h +++ b/src/Backups/BackupImpl.h @@ -81,8 +81,8 @@ class BackupImpl : public IBackup SizeAndChecksum getFileSizeAndChecksum(const String & file_name) const override; std::unique_ptr readFile(const String & file_name) const override; std::unique_ptr readFile(const String & file_name, const SizeAndChecksum & size_and_checksum) const override; - size_t copyFileToDisk(const String & file_name, DiskPtr destination_disk, const String & destination_path, WriteMode write_mode) const override; - size_t copyFileToDisk(const SizeAndChecksum & size_and_checksum, DiskPtr destination_disk, const String & destination_path, WriteMode write_mode) const override; + size_t copyFileToDisk(const String & file_name, DiskPtr destination_disk, const String & destination_path, WriteMode write_mode, bool sync) const override; + size_t copyFileToDisk(const SizeAndChecksum & size_and_checksum, DiskPtr destination_disk, const String & destination_path, WriteMode write_mode, bool sync) const override; void writeFile(const BackupFileInfo & info, BackupEntryPtr entry) override; bool supportsWritingInMultipleThreads() const override { return !use_archive; } void finalizeWriting() override; @@ -108,6 +108,11 @@ class BackupImpl : public IBackup String getObjectKey(const String & file_name) const; std::unique_ptr readFileByObjectKey(const BackupFileInfo & info) const; + /// Copies a lightweight-snapshot (object-key) entry to the destination through a live write buffer + /// and fsyncs it (the optimized object-key copy exposes no buffer to fsync). Reached only in the + /// cloud build, where object keys are present; defined unconditionally so it is type-checked everywhere. + size_t copyObjectKeyEntryToDiskSynced(const String & object_key, DiskPtr destination_disk, const String & destination_path, WriteMode write_mode) const; + /// Returns the base backup or null if there is no base backup. std::shared_ptr getBaseBackupUnlocked() const TSA_REQUIRES(mutex); diff --git a/src/Backups/IBackup.h b/src/Backups/IBackup.h index a9b68ef88341..a602c0ec7ce7 100644 --- a/src/Backups/IBackup.h +++ b/src/Backups/IBackup.h @@ -115,9 +115,11 @@ class IBackup : public std::enable_shared_from_this virtual std::unique_ptr readFile(const String & file_name, const SizeAndChecksum & size_and_checksum) const = 0; /// Copies a file from the backup to a specified destination disk. Returns the number of bytes written. - virtual size_t copyFileToDisk(const String & file_name, DiskPtr destination_disk, const String & destination_path, WriteMode write_mode) const = 0; + /// When `sync` is true the destination file's contents are fsynced before this call returns, so a + /// restored part can be made as durable as an inserted one (see MergeTreeData::restorePartFromBackup). + virtual size_t copyFileToDisk(const String & file_name, DiskPtr destination_disk, const String & destination_path, WriteMode write_mode, bool sync) const = 0; - virtual size_t copyFileToDisk(const SizeAndChecksum & size_and_checksum, DiskPtr destination_disk, const String & destination_path, WriteMode write_mode) const = 0; + virtual size_t copyFileToDisk(const SizeAndChecksum & size_and_checksum, DiskPtr destination_disk, const String & destination_path, WriteMode write_mode, bool sync) const = 0; /// Puts a new entry to the backup. virtual void writeFile(const BackupFileInfo & file_info, BackupEntryPtr entry) = 0; diff --git a/src/Columns/ColumnUnique.h b/src/Columns/ColumnUnique.h index 85cceba778f3..e56899a24083 100644 --- a/src/Columns/ColumnUnique.h +++ b/src/Columns/ColumnUnique.h @@ -176,6 +176,10 @@ class ColumnUnique final : public COWHelper getOrFindValueIndex(StringRef value) const override { + /// The reserved prefix slots are not in the reverse index, so match the default value here. + if (auto index = getNestedTypeDefaultValueIndex(); getRawColumnPtr()->getDataAt(index) == value) + return index; + if (std::optional res = reverse_index.getIndex(value); res) return res; diff --git a/src/Compression/LZ4_decompress_faster.cpp b/src/Compression/LZ4_decompress_faster.cpp index 224a8a14cd1d..cd4f4719e00f 100644 --- a/src/Compression/LZ4_decompress_faster.cpp +++ b/src/Compression/LZ4_decompress_faster.cpp @@ -643,9 +643,14 @@ bool decompress( size_t dest_size, PerformanceStatistics & statistics [[maybe_unused]]) { - if (source_size == 0 || dest_size == 0) + if (dest_size == 0) return true; + /// There is nothing to decompress from, but the caller expects `dest_size` bytes to be written, + /// and would otherwise hand out the previous contents of the destination buffer. + if (source_size == 0) + return false; + /// Don't run timer if the block is too small. if (dest_size >= 32768) { diff --git a/src/Dictionaries/DictionaryHelpers.h b/src/Dictionaries/DictionaryHelpers.h index d48ab9b47b49..6f266148004d 100644 --- a/src/Dictionaries/DictionaryHelpers.h +++ b/src/Dictionaries/DictionaryHelpers.h @@ -655,7 +655,8 @@ Block mergeBlockWithPipe( /** * Returns ColumnVector data as PaddedPodArray. - * If column is constant parameter backup_storage is used to store values. + * If the column has to be converted to a full one, parameter backup_storage is used to store values, + * because the converted column may not be owned by anything that outlives this call. */ /// TODO: Remove template @@ -664,7 +665,6 @@ static const PaddedPODArray & getColumnVectorData( const ColumnPtr column, PaddedPODArray & backup_storage) { - bool is_const_column = isColumnConst(*column); auto full_column = recursiveRemoveSparse(column->convertToFullColumnIfConst()); auto vector_col = checkAndGetColumn>(full_column.get()); @@ -676,12 +676,13 @@ static const PaddedPODArray & getColumnVectorData( TypeName); } - if (is_const_column) + /// A different pointer means a conversion happened (Const or Sparse; a Tuple never reaches here + /// because the check above requires a ColumnVector), so the data may live only in a column owned + /// by `full_column` and die at return: copy it. An unconverted column is kept alive by `column` + /// itself. + if (full_column.get() != column.get()) { - // With type conversion and const columns we need to use backup storage here - auto & data = vector_col->getData(); - backup_storage.assign(data); - + backup_storage.assign(vector_col->getData()); return backup_storage; } diff --git a/src/Functions/array/arrayIndex.h b/src/Functions/array/arrayIndex.h index 4459f82fd5e7..17d3d96aaa7d 100644 --- a/src/Functions/array/arrayIndex.h +++ b/src/Functions/array/arrayIndex.h @@ -29,9 +29,80 @@ namespace DB namespace ErrorCodes { + extern const int CANNOT_CONVERT_TYPE; + extern const int CANNOT_PARSE_BOOL; + extern const int CANNOT_PARSE_DATE; + extern const int CANNOT_PARSE_DATETIME; + extern const int CANNOT_PARSE_IPV4; + extern const int CANNOT_PARSE_IPV6; + extern const int CANNOT_PARSE_NUMBER; + extern const int CANNOT_PARSE_TEXT; + extern const int CANNOT_PARSE_UUID; + extern const int DECIMAL_OVERFLOW; extern const int ILLEGAL_COLUMN; extern const int ILLEGAL_TYPE_OF_ARGUMENT; extern const int LOGICAL_ERROR; + extern const int NOT_IMPLEMENTED; + extern const int TOO_LARGE_STRING_SIZE; + extern const int UNKNOWN_ELEMENT_OF_ENUM; + extern const int VALUE_IS_OUT_OF_RANGE_OF_DATA_TYPE; +} + +namespace ArrayIndexLowCardinalityHelpers +{ + +/// Is [code] a cast declining its input, rather than a fault of the caller? Anything else (a memory +/// limit, a logical error, a cancellation) is not an answer about the value and must propagate. +inline bool isConstantCastDecline(int code) +{ + return code == ErrorCodes::CANNOT_CONVERT_TYPE + || code == ErrorCodes::CANNOT_PARSE_BOOL + || code == ErrorCodes::CANNOT_PARSE_DATE + || code == ErrorCodes::CANNOT_PARSE_DATETIME + || code == ErrorCodes::CANNOT_PARSE_IPV4 + || code == ErrorCodes::CANNOT_PARSE_IPV6 + || code == ErrorCodes::CANNOT_PARSE_NUMBER + || code == ErrorCodes::CANNOT_PARSE_TEXT + || code == ErrorCodes::CANNOT_PARSE_UUID + || code == ErrorCodes::DECIMAL_OVERFLOW + || code == ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT + || code == ErrorCodes::NOT_IMPLEMENTED + || code == ErrorCodes::TOO_LARGE_STRING_SIZE + || code == ErrorCodes::UNKNOWN_ELEMENT_OF_ENUM + || code == ErrorCodes::VALUE_IS_OUT_OF_RANGE_OF_DATA_TYPE; +} + +/// Did [value] survive the cast that produced [image]? The cast alone cannot report loss, since it +/// truncates UInt64(256) to UInt8(0) and succeeds, so compare the two in the type they meet in, where +/// neither side's padding is a difference. +inline bool targetTypeRepresentsValue( + const ColumnPtr & value, const DataTypePtr & value_type, const ColumnPtr & image, const DataTypePtr & image_type) +{ + try + { + /// Without a common type the pair only compares as numbers, so [value_type] is where they meet. + const auto common_type = tryGetLeastSupertype(DataTypes{value_type, image_type}); + const auto compare_type = common_type ? makeNullable(common_type) : makeNullable(value_type); + + const auto restored = castColumnAccurateOrNull({image, image_type, ""}, compare_type); + if (restored->empty() || restored->isNullAt(0)) + return false; + + const auto original = castColumnAccurateOrNull({value, value_type, ""}, compare_type); + if (original->empty() || original->isNullAt(0)) + return false; + + return accurateEquals((*restored)[0], (*original)[0]); + } + catch (const Exception & e) + { + if (!isConstantCastDecline(e.code())) + throw; + + return false; + } +} + } using NullMap = PaddedPODArray; @@ -797,6 +868,14 @@ class FunctionArrayIndex : public IFunction const auto target_type = recursiveRemoveLowCardinality(array_type.getNestedType()); auto right = recursiveRemoveLowCardinality(right_const->getDataColumnPtr()); + /// A float zero equals two byte-distinct dictionary entries, -0.0 and 0.0, and a single index + /// cannot denote both, so leave a zero needle to the path that compares values. The needle + /// type is narrowed only so that reading it as a float is total. + const auto needle_type = removeNullable(recursiveRemoveLowCardinality(arguments[1].type)); + if (isFloat(removeNullable(target_type)) && (isNumber(needle_type) || isEnum(needle_type)) + && !right_const->isNullAt(0) && right_const->getDataColumnPtr()->getFloat64(0) == 0.0) + return nullptr; + UInt64 index = 0; UInt64 left_size = arguments[0].column->size(); ResultColumnPtr col_result = ResultColumnType::create(); @@ -804,15 +883,33 @@ class FunctionArrayIndex : public IFunction if (!right->isNullAt(0)) { auto right_type = recursiveRemoveLowCardinality(arguments[1].type); + auto original_right = right; + auto cast_type = target_type; right = castColumn({right, right_type, ""}, target_type); if (right->isNullable()) + { right = checkAndGetColumn(*right).getNestedColumnPtr(); + cast_type = removeNullable(cast_type); + } StringRef elem = right->getDataAt(0); const auto & left_dict = left_lc->getDictionary(); - if (std::optional maybe_index = left_dict.getOrFindValueIndex(elem); maybe_index) + auto find_in_dictionary = [&](StringRef value) -> std::optional + { + /// The default slot holds its value whether or not any row references it, and the cast above + /// narrows without reporting loss, so UInt64(256) reaches it as UInt8(0). Answering from that + /// slot requires the constant to have survived the cast; one that did not equals no element. + if (value == left_dict.getNestedNotNullableColumn()->getDataAt(left_dict.getNestedTypeDefaultValueIndex()) + && !target_type->equals(*right_type) + && !ArrayIndexLowCardinalityHelpers::targetTypeRepresentsValue(original_right, right_type, right, cast_type)) + return {}; + + return left_dict.getOrFindValueIndex(value); + }; + + if (std::optional maybe_index = find_in_dictionary(elem); maybe_index) { index = *maybe_index; } diff --git a/src/Functions/formatRow.cpp b/src/Functions/formatRow.cpp index 366d79043f96..0551e07c225e 100644 --- a/src/Functions/formatRow.cpp +++ b/src/Functions/formatRow.cpp @@ -95,7 +95,16 @@ class FunctionFormatRow : public IFunction row_output_format->finalize(); if constexpr (no_newline) { - if (buffer.position() != buffer.buffer().begin() && buffer.position()[-1] == '\n') + /// Strip a single trailing newline, but only when this row actually emitted at least one byte. + /// `buffer.count()` is the absolute number of bytes written; the current row starts at the + /// previous row's end offset (0 for the first row). Comparing against it prevents rewinding into + /// the previous row when this row is empty, which would make `offsets` non-monotonic and cause a + /// `size_t` underflow in `ColumnString::sizeAt`. The check against `buffer.buffer().begin()` + /// additionally keeps the position within the current working buffer so `--buffer.position()` + /// never moves the cursor before it. + const size_t row_start = i == 0 ? 0 : offsets[i - 1]; + if (buffer.count() > row_start && buffer.position() > buffer.buffer().begin() + && buffer.position()[-1] == '\n') --buffer.position(); } diff --git a/src/Interpreters/Cache/QueryLimit.cpp b/src/Interpreters/Cache/QueryLimit.cpp index a7c964022a54..34b46be23d0c 100644 --- a/src/Interpreters/Cache/QueryLimit.cpp +++ b/src/Interpreters/Cache/QueryLimit.cpp @@ -23,21 +23,46 @@ FileCacheQueryLimit::QueryContextPtr FileCacheQueryLimit::tryGetQueryContext(con if (!isQueryInitialized()) return nullptr; + std::lock_guard lock(query_map_mutex); auto query_iter = query_map.find(std::string(CurrentThread::getQueryId())); return (query_iter == query_map.end()) ? nullptr : query_iter->second; } -void FileCacheQueryLimit::removeQueryContext(const std::string & query_id, const CachePriorityGuard::Lock &) +FileCacheQueryLimit::QueryContextPtr +FileCacheQueryLimit::removeQueryContext(const std::string & query_id, QueryContextPtr & context, const CachePriorityGuard::Lock &) { - auto query_iter = query_map.find(query_id); - if (query_iter == query_map.end()) + QueryContextPtr doomed; { - throw Exception( - ErrorCodes::LOGICAL_ERROR, - "Attempt to release query context that does not exist (query_id: {})", - query_id); + std::lock_guard lock(query_map_mutex); + + auto query_iter = query_map.find(query_id); + const bool owns_map_entry = query_iter != query_map.end() && query_iter->second == context; + + /// Drop this holder's own reference to the context under the lock, then decide. use_count() + /// is not a synchronization primitive, so the decision must be made after every reference + /// change to the context is serialized by this mutex (which also guards getOrSetQueryContext). + /// Deciding before dropping the reference (or dropping it outside the lock) is a TOCTOU: + /// two holders releasing at once can both observe the shared count and both skip the erase, + /// orphaning the map entry, or one can erase while the other is being revived (see #109508). + context.reset(); + + if (owns_map_entry && query_iter->second.use_count() == 1) + { + /// The reference this holder held is gone and the map entry is now the sole owner, so + /// this was the last holder. Extract the pointer instead of erasing in place so the + /// QueryContext (its records map and per-query priority queue) is destroyed by the + /// caller after the cache write lock is released, not under it. Otherwise a query that + /// touched many segments frees all of that state while holding cache->lockCache(), + /// blocking unrelated reserve/eviction work for the duration of teardown. + doomed = std::move(query_iter->second); + query_map.erase(query_iter); + } + /// If owns_map_entry is false, the entry was already removed or re-created for a newer holder + /// via getOrSetQueryContext; another live holder now owns it, so leave it in place. If the + /// map entry is not the sole owner, another holder for the same query_id is still alive and + /// the context must stay so the per-query limit keeps being enforced. } - query_map.erase(query_iter); + return doomed; } FileCacheQueryLimit::QueryContextPtr FileCacheQueryLimit::getOrSetQueryContext( @@ -48,6 +73,7 @@ FileCacheQueryLimit::QueryContextPtr FileCacheQueryLimit::getOrSetQueryContext( if (query_id.empty()) return nullptr; + std::lock_guard lock(query_map_mutex); auto [it, inserted] = query_map.emplace(query_id, nullptr); if (inserted) { @@ -125,12 +151,19 @@ FileCacheQueryLimit::QueryContextHolder::QueryContextHolder( FileCacheQueryLimit::QueryContextHolder::~QueryContextHolder() { - /// If only the query_map and the current holder hold the context_query, - /// the query has been completed and the query_context is released. - if (context && context.use_count() == 2) + /// The last-holder decision (and the drop of this holder's reference) must happen inside + /// removeQueryContext under the cache write lock, not here: dropping the reference or deciding + /// outside the lock races with revival via getOrSetQueryContext and can leak or orphan the entry. + /// context is only set when the per-query download limit is enabled, so this is a no-op otherwise. + if (context) { - auto lock = cache->lockCache(); - query_limit->removeQueryContext(query_id, lock); + /// When this is the last holder, removeQueryContext hands the context back so it is destroyed + /// here, after the cache lock scope has ended, rather than under cache->lockCache(). + QueryContextPtr doomed; + { + auto lock = cache->lockCache(); + doomed = query_limit->removeQueryContext(query_id, context, lock); + } } } diff --git a/src/Interpreters/Cache/QueryLimit.h b/src/Interpreters/Cache/QueryLimit.h index 7553eff82baa..c12f4948a742 100644 --- a/src/Interpreters/Cache/QueryLimit.h +++ b/src/Interpreters/Cache/QueryLimit.h @@ -2,6 +2,8 @@ #include #include +#include + namespace DB { struct ReadSettings; @@ -20,7 +22,11 @@ class FileCacheQueryLimit const ReadSettings & settings, const CachePriorityGuard::Lock &); - void removeQueryContext(const std::string & query_id, const CachePriorityGuard::Lock &); + /// Releases this holder's reference to the query context and, when it was the last holder, + /// removes the map entry and returns the now-orphaned context so the caller can destroy it + /// after releasing the cache write lock (see ~QueryContextHolder). Returns nullptr when the + /// context is still owned by another live holder. + QueryContextPtr removeQueryContext(const std::string & query_id, QueryContextPtr & context, const CachePriorityGuard::Lock &); class QueryContext { @@ -77,6 +83,9 @@ class FileCacheQueryLimit private: using QueryContextMap = std::unordered_map; QueryContextMap query_map; + /// query_map is protected by this dedicated leaf mutex. Its callers can hold the cache + /// priority lock, but the mutex is the single lock that serializes map access. + mutable std::mutex query_map_mutex; }; using FileCacheQueryLimitPtr = std::unique_ptr; diff --git a/src/Interpreters/tests/gtest_filecache.cpp b/src/Interpreters/tests/gtest_filecache.cpp index 13a403583f14..f5a6cb543ddd 100644 --- a/src/Interpreters/tests/gtest_filecache.cpp +++ b/src/Interpreters/tests/gtest_filecache.cpp @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -1603,3 +1604,131 @@ TEST_F(FileCacheTest, ContinueEvictionPos) priority.resetEvictionPos(cache_lock); ASSERT_EQ(priority.getEvictionPos(), 3); /// queue.end() } + +TEST_F(FileCacheTest, QueryLimitContextRevivedDuringRelease) +{ + /// Regression for STID 4192-71db: a holder for some query_id decides it is the last one and + /// releases its query context, but a concurrent holder for the same query_id revives the + /// context first. The release must then be a no-op: the revived context must survive (so the + /// per-query download limit keeps being enforced for the rest of the query) and a later release + /// of the revived context must not fail with "Attempt to release query context that does not exist". + + CachePriorityGuard cache_guard; + FileCacheQueryLimit query_limit; + + const std::string query_id = "query_id_revive"; + ReadSettings read_settings; + read_settings.filesystem_cache_max_download_size = 1024; + + /// holder1 takes the context; query_map and holder1 both reference it (use_count == 2). + auto context1 = query_limit.getOrSetQueryContext(query_id, read_settings, cache_guard.lock()); + ASSERT_TRUE(context1 != nullptr); + ASSERT_EQ(context1.use_count(), 2); + + /// holder2 revives the same context before holder1 releases (getOrSetQueryContext returns the + /// existing entry). Now query_map, holder1 and holder2 all reference it (use_count == 3). + auto context2 = query_limit.getOrSetQueryContext(query_id, read_settings, cache_guard.lock()); + ASSERT_EQ(context1.get(), context2.get()); + ASSERT_EQ(context1.use_count(), 3); + + /// holder1 releases. The map still maps query_id to the live context and another holder is + /// alive, so the entry must be kept (no erase, no throw) and nothing is handed back for + /// destruction. + FileCacheQueryLimit::QueryContextPtr doomed1; + ASSERT_NO_THROW(doomed1 = query_limit.removeQueryContext(query_id, context1, cache_guard.lock())); + ASSERT_EQ(doomed1, nullptr); + context1.reset(); + + /// Enforcement is preserved: the revived context is still discoverable. + { + DB::ThreadStatus thread_status; + auto query_context = DB::Context::createCopy(getContext().context); + query_context->makeQueryContext(); + query_context->setCurrentQueryId(query_id); + DB::CurrentThread::QueryScope query_scope_holder(query_context); + + auto found = query_limit.tryGetQueryContext(cache_guard.lock()); + ASSERT_EQ(found.get(), context2.get()); + } + + /// holder2 is now the last holder; releasing it actually removes the entry, once, and hands the + /// orphaned context back so it is destroyed by the caller outside the cache lock. + const auto * context2_raw = context2.get(); + FileCacheQueryLimit::QueryContextPtr doomed2; + ASSERT_NO_THROW(doomed2 = query_limit.removeQueryContext(query_id, context2, cache_guard.lock())); + ASSERT_EQ(doomed2.get(), context2_raw); + ASSERT_EQ(doomed2.use_count(), 1); + context2.reset(); + + /// After full release the context is gone. + { + DB::ThreadStatus thread_status; + auto query_context = DB::Context::createCopy(getContext().context); + query_context->makeQueryContext(); + query_context->setCurrentQueryId(query_id); + DB::CurrentThread::QueryScope query_scope_holder(query_context); + + auto found = query_limit.tryGetQueryContext(cache_guard.lock()); + ASSERT_EQ(found.get(), nullptr); + } +} + +TEST_F(FileCacheTest, QueryLimitConcurrentReleaseNoLeak) +{ + /// Regression for #109508: two holders for the same query_id release "at the same time". + /// A query with parallel read streams has several holders (each CachedOnDiskReadBufferFromFile + /// creates its own), so use_count is > 2. If the last-holder decision reads use_count before this + /// holder drops its own reference (or drops it outside the lock), both releasers observe the shared + /// count, both skip the erase, and after both drop their reference only the map entry remains and is + /// never removed. That orphans query_map[query_id] for the lifetime of the cache and lets a later + /// query reusing the same query_id pick up stale per-query limit state. The fix drops each holder's + /// reference under the lock and erases once the map entry is the sole owner. + + CachePriorityGuard cache_guard; + FileCacheQueryLimit query_limit; + + const std::string query_id = "query_id_concurrent_release"; + ReadSettings read_settings; + read_settings.filesystem_cache_max_download_size = 1024; + + /// Two holders take the same context; query_map + both holders reference it (use_count == 3). + auto context1 = query_limit.getOrSetQueryContext(query_id, read_settings, cache_guard.lock()); + auto context2 = query_limit.getOrSetQueryContext(query_id, read_settings, cache_guard.lock()); + ASSERT_EQ(context1.get(), context2.get()); + ASSERT_EQ(context1.use_count(), 3); + + /// Keep a raw pointer to assert which release actually surrenders the context for destruction. + const auto * context_raw = context1.get(); + + /// Both holders decide to release while both are still alive (the interleaving that leaks): each + /// removeQueryContext drops that holder's reference under the lock. The first keeps the entry (one + /// holder still alive) and returns nullptr; the second erases it and returns the now-orphaned + /// context so the caller destroys it after the cache lock is released. Neither throws. + FileCacheQueryLimit::QueryContextPtr doomed1; + FileCacheQueryLimit::QueryContextPtr doomed2; + ASSERT_NO_THROW(doomed1 = query_limit.removeQueryContext(query_id, context1, cache_guard.lock())); + ASSERT_NO_THROW(doomed2 = query_limit.removeQueryContext(query_id, context2, cache_guard.lock())); + + /// removeQueryContext resets each passed reference, so both are already null here. + ASSERT_EQ(context1, nullptr); + ASSERT_EQ(context2, nullptr); + + /// Only the last release hands the context back for out-of-lock destruction; the earlier one + /// returns nullptr because another holder was still alive. + ASSERT_EQ(doomed1, nullptr); + ASSERT_EQ(doomed2.get(), context_raw); + ASSERT_EQ(doomed2.use_count(), 1); + + /// The entry must be gone: with the pre-fix logic both releases skipped the erase and the entry + /// leaked, so tryGetQueryContext would still find it. + { + DB::ThreadStatus thread_status; + auto query_context = DB::Context::createCopy(getContext().context); + query_context->makeQueryContext(); + query_context->setCurrentQueryId(query_id); + DB::CurrentThread::QueryScope query_scope_holder(query_context); + + auto found = query_limit.tryGetQueryContext(cache_guard.lock()); + ASSERT_EQ(found.get(), nullptr); + } +} diff --git a/src/Parsers/ExpressionElementParsers.cpp b/src/Parsers/ExpressionElementParsers.cpp index 84219a8ce02e..01fe4da1b433 100644 --- a/src/Parsers/ExpressionElementParsers.cpp +++ b/src/Parsers/ExpressionElementParsers.cpp @@ -1241,7 +1241,9 @@ inline static bool makeHexOrBinStringLiteral(IParser::Pos & pos, ASTPtr & node, binStringDecode(str_begin, str_end, res_pos, word_size); } - return makeStringLiteral(pos, node, String(reinterpret_cast(res.data()), res.size())); + /// The buffer is sized for the worst case; a binary literal whose length is not a multiple of + /// eight can write fewer bytes than that, and the unwritten tail is uninitialized memory. + return makeStringLiteral(pos, node, String(res_begin, res_pos - res_begin)); } bool ParserStringLiteral::parseImpl(Pos & pos, ASTPtr & node, Expected & expected) diff --git a/src/Planner/PlannerJoinTree.cpp b/src/Planner/PlannerJoinTree.cpp index 0f94600d0a49..1aec043a6ebf 100644 --- a/src/Planner/PlannerJoinTree.cpp +++ b/src/Planner/PlannerJoinTree.cpp @@ -20,6 +20,7 @@ #include #include +#include #include #include #include @@ -41,6 +42,8 @@ #include #include +#include +#include #include #include @@ -359,6 +362,12 @@ bool applyTrivialCountIfPossible( if (!count_func) return false; + /// `arrayJoin` in the argument multiplies rows above the source read, so the aggregate does not + /// observe `totalRows()` rows. Must precede `optimize_trivial_count`: storages that count in + /// read() act on that flag even when this function later declines. + if (hasFunctionNode(aggregates.front(), "arrayJoin")) + return false; + /// Some storages can optimize trivial count in read() method instead of totalRows() because it still can /// require reading some data (but much faster than reading columns). /// Set a special flag in query info so the storage will see it and optimize count in read() method. @@ -1169,7 +1178,17 @@ JoinTreeQueryPlan buildQueryPlanForTableExpression(QueryTreeNodePtr table_expres return false; }; - if (query_context->canUseParallelReplicasCustomKey() && query_context->getClientInfo().distributed_depth == 0) + /// The custom-key read below replaces the plan with a remote read at the fixed stage + /// `WithMergeableStateAfterAggregationAndLimit`, so it is only allowed when the requested + /// stage is not below that: a plan built up to a partial stage - e.g. a `Merge` table plans + /// its children up to `WithMergeableState` when one of the underlying tables is read through + /// an interpreter - must not receive finalized (post-aggregation, post-LIMIT) data instead + /// of the partial aggregation states its consumer expects. + const bool to_stage_supports_custom_key = select_query_options.to_stage == QueryProcessingStage::Complete + || select_query_options.to_stage == QueryProcessingStage::WithMergeableStateAfterAggregationAndLimit; + + if (query_context->canUseParallelReplicasCustomKey() && to_stage_supports_custom_key + && query_context->getClientInfo().distributed_depth == 0) { if (auto cluster = query_context->getClusterForParallelReplicas(); query_context->canUseParallelReplicasCustomKeyForCluster(*cluster)) @@ -2536,6 +2555,35 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, * Examples: Distributed, LiveView, Merge storages. */ auto left_table_expression = table_expressions_stack.front(); + + /** If the leftmost table uses IStorageCluster (e.g., s3Cluster, hdfsCluster) + * and there are multiple tables (indicating a JOIN), we must wrap it in a subquery. + * This prevents IStorageCluster from receiving the full JOIN query, which it cannot handle. + * + * IStorageCluster is a simple storage that just forwards queries to remote nodes. + * Unlike StorageDistributed, it cannot decompose and handle JOINs across multiple tables, + * because remote nodes don't have access to other tables in the JOIN. + * + * StorageDistributed has sophisticated query planning logic to handle JOINs and should + * NOT be wrapped (wrapping breaks tests like 03577_server_constant_folding). + */ + bool should_wrap_left_table = false; + bool has_multiple_tables = table_expressions_stack.size() > 1; + + if (has_multiple_tables) + { + // Get the actual storage to check its type + auto * table_node = left_table_expression->as(); + auto * table_function_node = left_table_expression->as(); + + if (table_node || table_function_node) + { + const auto & storage = table_node ? table_node->getStorage() : table_function_node->getStorage(); + // Only wrap if it's specifically IStorageCluster, not StorageDistributed or other remote storages + should_wrap_left_table = (dynamic_cast(storage.get()) != nullptr); + } + } + auto left_table_expression_query_plan = buildQueryPlanForTableExpression( left_table_expression, parent_join_tree, @@ -2543,7 +2591,7 @@ JoinTreeQueryPlan buildJoinTreeQueryPlan(const QueryTreeNodePtr & query_node, select_query_options, planner_context, is_single_table_expression, - false /*wrap_read_columns_in_subquery*/); + should_wrap_left_table /*wrap_read_columns_in_subquery*/); if (left_table_expression_query_plan.stage != QueryProcessingStage::FetchColumns) return left_table_expression_query_plan; diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp index 9cf2077b0c19..09d8523dcd46 100644 --- a/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp +++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.cpp @@ -32,9 +32,14 @@ void Prefetcher::init(ReadBuffer * reader_, const ReadOptions & options, FormatP range_sets.resize(1); } -Prefetcher::~Prefetcher() +void Prefetcher::shutdownTasks() { shutdown->shutdown(); +} + +Prefetcher::~Prefetcher() +{ + shutdownTasks(); /// Assert that all PrefetchHandle-s were destroyed. chassert(std::all_of(requests.begin(), requests.end(), [](const RequestState & req) diff --git a/src/Processors/Formats/Impl/Parquet/Prefetcher.h b/src/Processors/Formats/Impl/Parquet/Prefetcher.h index 954007941a47..4d0dc92868f3 100644 --- a/src/Processors/Formats/Impl/Parquet/Prefetcher.h +++ b/src/Processors/Formats/Impl/Parquet/Prefetcher.h @@ -28,6 +28,10 @@ class Prefetcher /// Waits for in-progress reads to complete, cancels queued reads that haven't started yet. ~Prefetcher(); + /// Same handshake as the destructor. After this returns, no background task reads through the + /// ReadBuffer passed to init() anymore, so that buffer may be destroyed. Idempotent. + void shutdownTasks(); + /// Not thread safe. /// All ranges must be registered before any reading happens (except direct readSync). /// Ranges are allowed to overlap a little, but this decreases the effectiveness of range diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp index 65448a8ec2b7..e99539f9ff2b 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.cpp +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.cpp @@ -94,9 +94,15 @@ void ReadManager::init(FormatParserSharedResourcesPtr parser_shared_resources_, flushMemoryUsageDiff(std::move(diff)); } -ReadManager::~ReadManager() +void ReadManager::shutdownTasks() { shutdown->shutdown(); + reader.prefetcher.shutdownTasks(); +} + +ReadManager::~ReadManager() +{ + shutdownTasks(); } void ReadManager::cancel() noexcept diff --git a/src/Processors/Formats/Impl/Parquet/ReadManager.h b/src/Processors/Formats/Impl/Parquet/ReadManager.h index 113d4430ee2a..724901b432be 100644 --- a/src/Processors/Formats/Impl/Parquet/ReadManager.h +++ b/src/Processors/Formats/Impl/Parquet/ReadManager.h @@ -44,6 +44,11 @@ class ReadManager ~ReadManager(); + /// Same handshake as the destructor, but keeps `reader` and its metadata intact. After this + /// returns, no decode task runs anymore, so nothing can re-enter the prefetcher. Idempotent. + /// Drain this before the prefetcher: decode tasks read ranges through it. + void shutdownTasks(); + struct ReadResult { Chunk chunk; diff --git a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp index 2d2bc2f318f0..0698ea695efe 100644 --- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp +++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp @@ -89,6 +89,7 @@ void ParquetV3BlockInputFormat::initializeIfNeeded() parser_shared_resources->opaque = ext; }); + std::lock_guard lock(reader_mutex); reader.emplace(); reader->reader.prefetcher.init(in, read_options, parser_shared_resources); reader->reader.init(read_options, getPort().getHeader(), format_filter_info); @@ -136,13 +137,30 @@ const BlockMissingValues * ParquetV3BlockInputFormat::getMissingValues() const void ParquetV3BlockInputFormat::onCancel() noexcept { + std::lock_guard lock(reader_mutex); if (reader) reader->cancel(); } +void ParquetV3BlockInputFormat::resetReadBuffer() +{ + { + /// Background tasks read through a non-owning pointer to the buffers the base class is + /// about to release, so they have to be stopped first. `reader` stays alive: + /// getMatchedBuckets() reads row group metadata after the source is exhausted. + std::lock_guard lock(reader_mutex); + if (reader) + reader->shutdownTasks(); + } + IInputFormat::resetReadBuffer(); +} + void ParquetV3BlockInputFormat::resetParser() { - reader.reset(); + { + std::lock_guard lock(reader_mutex); + reader.reset(); + } previous_block_missing_values.clear(); IInputFormat::resetParser(); } diff --git a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.h b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.h index ff7c80a6b230..055b72eef49c 100644 --- a/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.h +++ b/src/Processors/Formats/Impl/ParquetV3BlockInputFormat.h @@ -2,6 +2,8 @@ #include "config.h" #if USE_PARQUET +#include + #include #include #include @@ -24,6 +26,8 @@ class ParquetV3BlockInputFormat : public IInputFormat void resetParser() override; + void resetReadBuffer() override; + String getName() const override { return "ParquetV3BlockInputFormat"; } const BlockMissingValues * getMissingValues() const override; @@ -45,6 +49,8 @@ class ParquetV3BlockInputFormat : public IInputFormat FormatParserSharedResourcesPtr parser_shared_resources; FormatFilterInfoPtr format_filter_info; + /// Protects the optional against concurrent initialization and cancellation. + std::mutex reader_mutex; std::optional reader; bool reported_count = false; // if need_only_count diff --git a/src/Storages/MergeTree/BackgroundJobsAssignee.cpp b/src/Storages/MergeTree/BackgroundJobsAssignee.cpp index 90244a3d6a35..0199816e6368 100644 --- a/src/Storages/MergeTree/BackgroundJobsAssignee.cpp +++ b/src/Storages/MergeTree/BackgroundJobsAssignee.cpp @@ -118,10 +118,19 @@ void BackgroundJobsAssignee::start() void BackgroundJobsAssignee::finish() { - /// No lock here, because scheduled tasks could call trigger method - if (holder) + /// Move the holder to a local variable under the lock, then release the lock + /// before calling deactivate(). We cannot hold holder_mutex during deactivate() + /// because it waits for the background task (threadFunc) to finish, and threadFunc + /// calls trigger()/postpone() which also lock holder_mutex — that would deadlock. + BackgroundSchedulePoolTaskHolder local_holder; { - holder->deactivate(); + std::lock_guard lock(holder_mutex); + local_holder = std::move(holder); + } + + if (local_holder) + { + local_holder->deactivate(); auto storage_id = data.getStorageID(); diff --git a/src/Storages/MergeTree/IMergeTreeDataPart.cpp b/src/Storages/MergeTree/IMergeTreeDataPart.cpp index c816cb18cf9d..8afdd764a37d 100644 --- a/src/Storages/MergeTree/IMergeTreeDataPart.cpp +++ b/src/Storages/MergeTree/IMergeTreeDataPart.cpp @@ -1327,6 +1327,9 @@ void IMergeTreeDataPart::writeMetadataVersion(ContextPtr context, int32_t metada { getDataPartStorage().beginTransaction(); { + // We need to remove the old file first to overwrite it only, not all its hard links. + getDataPartStorage().removeFileIfExists(METADATA_VERSION_FILE_NAME); + auto out_metadata = getDataPartStorage().writeFile(METADATA_VERSION_FILE_NAME, 4096, context->getWriteSettings()); writeText(metadata_version_, *out_metadata); out_metadata->finalize(); diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp index 5e58d7c85a72..4acba33f3bb5 100644 --- a/src/Storages/MergeTree/MergeTreeData.cpp +++ b/src/Storages/MergeTree/MergeTreeData.cpp @@ -6876,6 +6876,12 @@ void MergeTreeData::restorePartFromBackup(std::shared_ptr r /// Subdirectories in the part's directory. It's used to restore projections. std::unordered_set subdirs; + /// A restored part is committed data the moment RESTORE is acknowledged, so it must get the same + /// durability an inserted part gets: fsync the file contents when the table enables fsync_after_insert. + /// Only meaningful on a local disk - on object storage the object is durable once finalized. The part + /// directory itself is fsynced later by IMergeTreeDataPart::renameTo (gated on fsync_part_directory). + const bool fsync_files = (*getSettings())[MergeTreeSetting::fsync_after_insert] && !disk->isRemote(); + /// Copy files from the backup to the directory `tmp_part_dir`. disk->createDirectories(temp_part_dir); @@ -6899,7 +6905,7 @@ void MergeTreeData::restorePartFromBackup(std::shared_ptr r continue; } - size_t file_size = backup->copyFileToDisk(part_path_in_backup_fs / filename, disk, temp_part_dir / filename, WriteMode::Rewrite); + size_t file_size = backup->copyFileToDisk(part_path_in_backup_fs / filename, disk, temp_part_dir / filename, WriteMode::Rewrite, fsync_files); reservation->update(reservation->getSize() - file_size); } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index 2e7160c9f64b..0dff37a962fb 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -688,6 +688,8 @@ MetadataFileWithInfo getLatestOrExplicitMetadataFileAndVersion( if (data_lake_settings[DataLakeStorageSetting::iceberg_metadata_file_path].changed) { auto explicit_metadata_path = data_lake_settings[DataLakeStorageSetting::iceberg_metadata_file_path].value; + if (explicit_metadata_path.find('\0') != String::npos) + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg metadata file path contains a null byte"); try { LOG_TEST(log, "Explicit metadata file path is specified {}, will read from this metadata file", explicit_metadata_path); diff --git a/src/Storages/StorageLog.cpp b/src/Storages/StorageLog.cpp index 6fa4c07a44d4..255ec33ae707 100644 --- a/src/Storages/StorageLog.cpp +++ b/src/Storages/StorageLog.cpp @@ -1171,7 +1171,7 @@ void StorageLog::restoreDataImpl(const BackupPtr & backup, const String & data_p if (!backup->fileExists(file_path_in_backup)) throw Exception(ErrorCodes::CANNOT_RESTORE_TABLE, "File {} in backup is required to restore table", file_path_in_backup); - backup->copyFileToDisk(file_path_in_backup, disk, data_file.path, WriteMode::Append); + backup->copyFileToDisk(file_path_in_backup, disk, data_file.path, WriteMode::Append, /* sync= */ false); } if (use_marks_file) diff --git a/src/Storages/StorageMerge.cpp b/src/Storages/StorageMerge.cpp index d3d2034f8c5b..60880ee96101 100644 --- a/src/Storages/StorageMerge.cpp +++ b/src/Storages/StorageMerge.cpp @@ -1284,7 +1284,16 @@ ReadFromMerge::RowPolicyData::RowPolicyData(RowPolicyFilterPtr row_policy_filter auto storage_columns = storage_metadata_snapshot->getColumns(); auto needed_columns = storage_columns.getAll(); - ASTPtr expr = row_policy_filter_ptr->expression; + /// `RowPolicyFilter::expression` is the parsed policy condition owned by `RowPolicyCache`. That AST is + /// shared: every query of every user reading this table gets the same nodes, and a policy defined on a + /// whole database is shared by all its tables. `TreeRewriter` and `ExpressionAnalyzer` rewrite the AST + /// they are given in place - they normalize identifiers, substitute the results of scalar subqueries for + /// the subqueries themselves, and record `ASTLiteral::unique_column_name` - so they must be handed a + /// private copy. Analyzing the shared AST is both a data race against concurrent readers of the same + /// policy and a correctness bug: a scalar subquery such as `USING x <= (SELECT max(v) FROM limits)` gets + /// replaced by its value in the cache and is then frozen for the rest of the server's lifetime. + /// `generateFilterActions` in `InterpreterSelectQuery` clones for the same reason. + ASTPtr expr = row_policy_filter_ptr->expression->clone(); auto syntax_result = TreeRewriter(local_context).analyze(expr, needed_columns); auto expression_analyzer = ExpressionAnalyzer{expr, syntax_result, local_context}; diff --git a/src/Storages/StorageStripeLog.cpp b/src/Storages/StorageStripeLog.cpp index b7c300e88c8f..867247fdb815 100644 --- a/src/Storages/StorageStripeLog.cpp +++ b/src/Storages/StorageStripeLog.cpp @@ -656,7 +656,7 @@ void StorageStripeLog::restoreDataImpl(const BackupPtr & backup, const String & if (!backup->fileExists(file_path_in_backup)) throw Exception(ErrorCodes::CANNOT_RESTORE_TABLE, "File {} in backup is required to restore table", file_path_in_backup); - backup->copyFileToDisk(file_path_in_backup, disk, data_file_path, WriteMode::Append); + backup->copyFileToDisk(file_path_in_backup, disk, data_file_path, WriteMode::Append, /* sync= */ false); } /// Append the index. diff --git a/src/Storages/System/StorageSystemContributors.generated.cpp b/src/Storages/System/StorageSystemContributors.generated.cpp index 462267d2cb58..a8bc53d9cfe4 100644 --- a/src/Storages/System/StorageSystemContributors.generated.cpp +++ b/src/Storages/System/StorageSystemContributors.generated.cpp @@ -1617,6 +1617,7 @@ const char * auto_contributors[] { "chenxing-xc", "chenxing.xc", "chertus", + "chethan-64", "chhetripradeep", "chloro", "chou.fan", diff --git a/tests/integration/test_storage_s3_queue/test_parallel_inserts.py b/tests/integration/test_storage_s3_queue/test_parallel_inserts.py index 2a68102d1cac..da6b35981273 100644 --- a/tests/integration/test_storage_s3_queue/test_parallel_inserts.py +++ b/tests/integration/test_storage_s3_queue/test_parallel_inserts.py @@ -302,6 +302,7 @@ def test_batch_set_processing_failure_does_not_crash(started_cluster): "enable_hash_ring_filtering": 1, "s3queue_processing_threads_num": 1, "s3queue_loading_retries": 100, + "use_persistent_processing_nodes": 1, # Both conditions must land in the SAME batch, so pin the listing batch size instead # of relying on the engine default (1000) happening to exceed files_to_generate. "list_objects_batch_size": files_to_generate, @@ -315,8 +316,10 @@ def test_batch_set_processing_failure_does_not_crash(started_cluster): conflict_file = f"{files_path}/test_1.csv" conflict_node = node.query(f"SELECT sipHash64('{conflict_file}')").strip() zk = started_cluster.get_kazoo_client("zoo1") - zk.ensure_path(f"{keeper_path}/processing") - zk.create(f"{keeper_path}/processing/{conflict_node}", b"conflict") + # Use the persistent-processing directory configured above. + processing_path = f"{keeper_path}/persistent_processing" + zk.ensure_path(processing_path) + zk.create(f"{processing_path}/{conflict_node}", b"conflict") def batch_set_processing_failures(): node.query("SELECT 1") # fails loudly if the server aborted @@ -353,7 +356,7 @@ def batch_set_processing_failures(): # Remove the artificial conflict and confirm the queue keeps making progress after the # failed batch (the iterator recovered rather than getting stuck or having crashed). - zk.delete(f"{keeper_path}/processing/{conflict_node}") + zk.delete(f"{processing_path}/{conflict_node}") def get_count(): return int(node.query(f"SELECT count() FROM {dst_table_name}")) diff --git a/tests/queries/0_stateless/03800_s3_cluster_join.reference b/tests/queries/0_stateless/03800_s3_cluster_join.reference new file mode 100644 index 000000000000..f3c39ef07117 --- /dev/null +++ b/tests/queries/0_stateless/03800_s3_cluster_join.reference @@ -0,0 +1,18 @@ +LEFT JOIN +false 2000 false 2000 +true 1000 true 1000 +INNER JOIN +false 2000 false 2000 +true 1000 true 1000 +RIGHT JOIN +false 2000 false 2000 +true 1000 true 1000 +LEFT OUTER JOIN +false 2000 false 2000 +true 1000 true 1000 +FULL OUTER JOIN +false 2000 false 2000 +true 1000 true 1000 +Simple SELECT +false 2000 +true 1000 diff --git a/tests/queries/0_stateless/03800_s3_cluster_join.sql b/tests/queries/0_stateless/03800_s3_cluster_join.sql new file mode 100644 index 000000000000..928550b7c300 --- /dev/null +++ b/tests/queries/0_stateless/03800_s3_cluster_join.sql @@ -0,0 +1,49 @@ +-- Tags: no-fasttest, no-parallel + +SET enable_analyzer = 1; + +-- Test LEFT JOIN +SELECT 'LEFT JOIN'; +SELECT * +FROM s3Cluster(test_shard_localhost, 'http://localhost:11111/test/03800_a.parquet', 'NOSIGN', 'Parquet', 'boolean_col Boolean, long_col Int64') AS t1 +LEFT JOIN s3Cluster(test_shard_localhost, 'http://localhost:11111/test/03800_b.parquet', 'NOSIGN', 'Parquet', 'boolean_col Boolean, long_col Int64') AS t2 +ON t1.boolean_col = t2.boolean_col +ORDER BY t1.boolean_col; + +-- Test INNER JOIN +SELECT 'INNER JOIN'; +SELECT * +FROM s3Cluster(test_shard_localhost, 'http://localhost:11111/test/03800_a.parquet', 'NOSIGN', 'Parquet', 'boolean_col Boolean, long_col Int64') AS t1 +INNER JOIN s3Cluster(test_shard_localhost, 'http://localhost:11111/test/03800_b.parquet', 'NOSIGN', 'Parquet', 'boolean_col Boolean, long_col Int64') AS t2 +ON t1.boolean_col = t2.boolean_col +ORDER BY t1.boolean_col; + +-- Test RIGHT JOIN +SELECT 'RIGHT JOIN'; +SELECT * +FROM s3Cluster(test_shard_localhost, 'http://localhost:11111/test/03800_a.parquet', 'NOSIGN', 'Parquet', 'boolean_col Boolean, long_col Int64') AS t1 +RIGHT JOIN s3Cluster(test_shard_localhost, 'http://localhost:11111/test/03800_b.parquet', 'NOSIGN', 'Parquet', 'boolean_col Boolean, long_col Int64') AS t2 +ON t1.boolean_col = t2.boolean_col +ORDER BY t2.boolean_col; + +-- Test LEFT OUTER JOIN +SELECT 'LEFT OUTER JOIN'; +SELECT * +FROM s3Cluster(test_shard_localhost, 'http://localhost:11111/test/03800_a.parquet', 'NOSIGN', 'Parquet', 'boolean_col Boolean, long_col Int64') AS t1 +LEFT OUTER JOIN s3Cluster(test_shard_localhost, 'http://localhost:11111/test/03800_b.parquet', 'NOSIGN', 'Parquet', 'boolean_col Boolean, long_col Int64') AS t2 +ON t1.boolean_col = t2.boolean_col +ORDER BY t1.boolean_col; + +-- Test FULL OUTER JOIN +SELECT 'FULL OUTER JOIN'; +SELECT * +FROM s3Cluster(test_shard_localhost, 'http://localhost:11111/test/03800_a.parquet', 'NOSIGN', 'Parquet', 'boolean_col Boolean, long_col Int64') AS t1 +FULL OUTER JOIN s3Cluster(test_shard_localhost, 'http://localhost:11111/test/03800_b.parquet', 'NOSIGN', 'Parquet', 'boolean_col Boolean, long_col Int64') AS t2 +ON t1.boolean_col = t2.boolean_col +ORDER BY t1.boolean_col, t2.boolean_col; + +-- Test that simple SELECT still works (no JOIN) +SELECT 'Simple SELECT'; +SELECT * +FROM s3Cluster(test_shard_localhost, 'http://localhost:11111/test/03800_a.parquet', 'NOSIGN', 'Parquet', 'boolean_col Boolean, long_col Int64') +ORDER BY boolean_col; diff --git a/tests/queries/0_stateless/04512_format_row_no_newline_empty_row_underflow.reference b/tests/queries/0_stateless/04512_format_row_no_newline_empty_row_underflow.reference new file mode 100644 index 000000000000..1b34d3678143 --- /dev/null +++ b/tests/queries/0_stateless/04512_format_row_no_newline_empty_row_underflow.reference @@ -0,0 +1,16 @@ +0 +2 + +610A +62 +0 +0 +0 +1 +1 +50 +44 +1\t2\tgood +0,"good" +1,"good" +2,"good" diff --git a/tests/queries/0_stateless/04512_format_row_no_newline_empty_row_underflow.sql b/tests/queries/0_stateless/04512_format_row_no_newline_empty_row_underflow.sql new file mode 100644 index 000000000000..af4df55206ce --- /dev/null +++ b/tests/queries/0_stateless/04512_format_row_no_newline_empty_row_underflow.sql @@ -0,0 +1,31 @@ +-- Tags: no-fasttest +-- no-fasttest: the RawBLOB format is not available in fast test builds. + +-- `formatRowNoNewline` strips a trailing newline from each row. It must never rewind past the start of the +-- current row into the previous row's bytes. Otherwise a row that emits no bytes (e.g. an empty string with the +-- `RawBLOB` format) produces non-monotonic `ColumnString` offsets and a `size_t` underflow in the string size. + +-- An empty row right after a non-empty one that ended with a newline must stay empty (length 0), not underflow. +SELECT length(formatRowNoNewline('RawBLOB', s)) AS len +FROM (SELECT arrayJoin(['a\n\n', '']) AS s) +ORDER BY ALL; + +-- The bytes of a following row must be exactly those the row emitted (no cross-row bleed). +SELECT hex(formatRowNoNewline('RawBLOB', s)) AS bytes +FROM (SELECT arrayJoin(['a\n\n', '', 'b']) AS s) +ORDER BY ALL; + +-- Several consecutive empty rows around non-empty ones stay empty and keep offsets monotonic. +SELECT length(formatRowNoNewline('RawBLOB', s)) AS len +FROM (SELECT arrayJoin(['x\n', '', '', 'y\n', '']) AS s) +ORDER BY ALL; + +-- Regression guard for the newline-stripping itself: it must keep working for rows after the internal write +-- buffer has grown past its initial size and been flushed at least once (which happens from the second row on +-- for these sizes). Every row must have its trailing newline stripped, so all lengths are equal. +SELECT DISTINCT length(formatRowNoNewline('TSV', repeat('z', 50))) AS len FROM numbers(5); +SELECT DISTINCT length(formatRowNoNewline('CSV', number, repeat('w', 40))) AS len FROM numbers(5); + +-- Normal row formats keep their usual behavior. +SELECT formatRowNoNewline('TSV', 1, 2, 'good') AS f; +SELECT formatRowNoNewline('CSV', number, 'good') FROM numbers(3); diff --git a/tests/queries/0_stateless/04550_restore_fsync_after_insert.reference b/tests/queries/0_stateless/04550_restore_fsync_after_insert.reference new file mode 100644 index 000000000000..a1f56bd020f6 --- /dev/null +++ b/tests/queries/0_stateless/04550_restore_fsync_after_insert.reference @@ -0,0 +1,6 @@ +has zero-byte part file: 1 +count on: 1000 +count off: 1000 +restore fsync delta covers all part files: 1 +encrypted incremental count: 1000 +encrypted incremental restore with fsync_after_insert=1, all part files fsynced: 1 diff --git a/tests/queries/0_stateless/04550_restore_fsync_after_insert.sh b/tests/queries/0_stateless/04550_restore_fsync_after_insert.sh new file mode 100755 index 000000000000..2bfaaa91e840 --- /dev/null +++ b/tests/queries/0_stateless/04550_restore_fsync_after_insert.sh @@ -0,0 +1,110 @@ +#!/usr/bin/env bash +# Tags: no-fasttest, no-object-storage, no-random-merge-tree-settings, no-replicated-database, no-shared-merge-tree +# no-fasttest: the encrypted case below needs the encrypted disk type, which is built only with SSL. +# no-object-storage: object storage does not fsync file contents (the fix is gated on !isRemote()). +# no-random-merge-tree-settings: the test asserts on FileSync counts, which depend on the part layout. +# no-replicated-database, no-shared-merge-tree: the encrypted case below pins a custom local disk. + +# Regression test for https://github.com/ClickHouse/ClickHouse/issues/111321 +# RESTORE must fsync the restored part file contents when the table enables fsync_after_insert, +# otherwise a power loss right after RESTORE returns leaves the parts torn and the table empty. +# We assert on the RESTORE query's FileSync ProfileEvent (parallel-safe: filtered by query_id + +# current_database). RESTORE also performs a few backup-side FileSync events unrelated to the part +# files, so the discriminating signal is the on-vs-off FileSync DELTA: that constant backup-side +# noise cancels out, and the remaining delta must cover every physical file of the restored part +# (an empty Array column contributes a zero-byte .bin, which INSERT fsyncs too). + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +# `arr` is Array(UInt32) left empty for every row, so its `.bin` is a required zero-byte part file - +# INSERT fsyncs it, so RESTORE must too. Two tables with identical data, differing only in fsync_after_insert. +$CLICKHOUSE_CLIENT -m -q " + DROP TABLE IF EXISTS t_restore_fsync_on; + DROP TABLE IF EXISTS t_restore_fsync_off; + + CREATE TABLE t_restore_fsync_on (id UInt64, s String, arr Array(UInt32)) ENGINE = MergeTree ORDER BY id + SETTINGS fsync_after_insert = 1, fsync_part_directory = 1, min_bytes_for_wide_part = 0; + INSERT INTO t_restore_fsync_on SELECT number, toString(number), [] FROM numbers(1000); + + CREATE TABLE t_restore_fsync_off (id UInt64, s String, arr Array(UInt32)) ENGINE = MergeTree ORDER BY id + SETTINGS fsync_after_insert = 0, fsync_part_directory = 0, min_bytes_for_wide_part = 0; + INSERT INTO t_restore_fsync_off SELECT number, toString(number), [] FROM numbers(1000); +" + +# Count the physical files RESTORE actually copies (and therefore must fsync): every file in the part +# except the version-metadata files RESTORE deliberately skips (see restorePartFromBackup: +# txn_version.txt[.tmp] and metadata_version.txt are not copied). The restored part has the same +# on-disk file set. +part_path=$($CLICKHOUSE_CLIENT -q "SELECT path FROM system.parts WHERE database = currentDatabase() AND table = 't_restore_fsync_on' AND active") +copied_files=$(find "$part_path" -type f \ + ! -name 'txn_version.txt' ! -name 'txn_version.txt.tmp' ! -name 'metadata_version.txt' | wc -l) +# Sanity: there is a required zero-byte file in the part (the empty Array's .bin), which INSERT fsyncs too. +zero_byte_files=$(find "$part_path" -type f -size 0 | wc -l) +echo "has zero-byte part file: $([ "$zero_byte_files" -ge 1 ] && echo 1 || echo 0)" + +$CLICKHOUSE_CLIENT -q "BACKUP TABLE t_restore_fsync_on TO Disk('backups', '${CLICKHOUSE_TEST_UNIQUE_NAME}_on')" > /dev/null +$CLICKHOUSE_CLIENT -q "BACKUP TABLE t_restore_fsync_off TO Disk('backups', '${CLICKHOUSE_TEST_UNIQUE_NAME}_off')" > /dev/null +$CLICKHOUSE_CLIENT -m -q "DROP TABLE t_restore_fsync_on SYNC; DROP TABLE t_restore_fsync_off SYNC;" + +qid_on="restore-on-$CLICKHOUSE_DATABASE" +qid_off="restore-off-$CLICKHOUSE_DATABASE" +$CLICKHOUSE_CLIENT --query_id "$qid_on" -q "RESTORE TABLE t_restore_fsync_on FROM Disk('backups', '${CLICKHOUSE_TEST_UNIQUE_NAME}_on')" > /dev/null +$CLICKHOUSE_CLIENT --query_id "$qid_off" -q "RESTORE TABLE t_restore_fsync_off FROM Disk('backups', '${CLICKHOUSE_TEST_UNIQUE_NAME}_off')" > /dev/null + +# Data must be intact after restore. +echo "count on: $($CLICKHOUSE_CLIENT -q "SELECT count() FROM t_restore_fsync_on")" +echo "count off: $($CLICKHOUSE_CLIENT -q "SELECT count() FROM t_restore_fsync_off")" + +$CLICKHOUSE_CLIENT -q "SYSTEM FLUSH LOGS query_log" + +# The on/off FileSync delta cancels the constant backup-side syncs that both restores perform and +# isolates the restored part-file syncs. With fsync_after_insert=1 every restore-copied file (incl. the +# zero-byte arr.bin) is fsynced, so the delta must be >= the number of files restore copied. Before the +# fix no part file was synced and the delta was ~0. +$CLICKHOUSE_CLIENT --param_qid_on "$qid_on" --param_qid_off "$qid_off" --param_copied "$copied_files" -q " + WITH + (SELECT ProfileEvents['FileSync'] FROM system.query_log + WHERE query_id = {qid_on:String} AND type = 'QueryFinish' AND current_database = currentDatabase() + ORDER BY event_time_microseconds DESC LIMIT 1) AS sync_on, + (SELECT ProfileEvents['FileSync'] FROM system.query_log + WHERE query_id = {qid_off:String} AND type = 'QueryFinish' AND current_database = currentDatabase() + ORDER BY event_time_microseconds DESC LIMIT 1) AS sync_off + SELECT 'restore fsync delta covers all part files: ', (toInt64(sync_on) - toInt64(sync_off)) >= {copied:UInt64}" + +$CLICKHOUSE_CLIENT -m -q "DROP TABLE t_restore_fsync_on SYNC; DROP TABLE t_restore_fsync_off SYNC;" + +# Encrypted incremental restore: files that come entirely from the base backup are copied via +# getBaseBackup()->copyFileToDisk(..., sync). That branch must forward the encrypted read (else the +# restore fails with CANNOT_RESTORE_TO_NONENCRYPTED_DISK) and still fsync the files when requested. +# The table below has the same schema, data and part-layout settings as t_restore_fsync_on, so its +# part holds the same file set and $copied_files is its restore-copied file count too. +enc_disk="disk(type = encrypted, disk = disk(type = local, path = '${CLICKHOUSE_DISKS_FILES}/${CLICKHOUSE_TEST_UNIQUE_NAME}_enc/'), key = '1234567812345678')" +$CLICKHOUSE_CLIENT -q " + DROP TABLE IF EXISTS t_restore_fsync_enc; + CREATE TABLE t_restore_fsync_enc (id UInt64, s String, arr Array(UInt32)) ENGINE = MergeTree ORDER BY id + SETTINGS fsync_after_insert = 1, fsync_part_directory = 1, min_bytes_for_wide_part = 0, disk = $enc_disk; + INSERT INTO t_restore_fsync_enc SELECT number, toString(number), [] FROM numbers(1000); +" + +# Full backup, then an unchanged incremental backup so every file is served by the base backup. +$CLICKHOUSE_CLIENT -q "BACKUP TABLE t_restore_fsync_enc TO Disk('backups', '${CLICKHOUSE_TEST_UNIQUE_NAME}_enc_base')" > /dev/null +$CLICKHOUSE_CLIENT -q "BACKUP TABLE t_restore_fsync_enc TO Disk('backups', '${CLICKHOUSE_TEST_UNIQUE_NAME}_enc_incr') SETTINGS base_backup = Disk('backups', '${CLICKHOUSE_TEST_UNIQUE_NAME}_enc_base')" > /dev/null +$CLICKHOUSE_CLIENT -q "DROP TABLE t_restore_fsync_enc SYNC" + +qid_enc="restore-enc-$CLICKHOUSE_DATABASE" +$CLICKHOUSE_CLIENT --query_id "$qid_enc" -q "RESTORE TABLE t_restore_fsync_enc FROM Disk('backups', '${CLICKHOUSE_TEST_UNIQUE_NAME}_enc_incr') SETTINGS base_backup = Disk('backups', '${CLICKHOUSE_TEST_UNIQUE_NAME}_enc_base')" > /dev/null + +# Before the fix the restore threw and the table stayed empty; now it restores every row. +echo "encrypted incremental count: $($CLICKHOUSE_CLIENT -q "SELECT count() FROM t_restore_fsync_enc")" + +$CLICKHOUSE_CLIENT -q "SYSTEM FLUSH LOGS query_log" +$CLICKHOUSE_CLIENT --param_query_id "$qid_enc" --param_files "$copied_files" -q " + SELECT 'encrypted incremental restore with fsync_after_insert=1, all part files fsynced: ', + ProfileEvents['FileSync'] >= {files:UInt64} + FROM system.query_log + WHERE query_id = {query_id:String} AND type = 'QueryFinish' AND current_database = currentDatabase() + ORDER BY event_time_microseconds DESC LIMIT 1" + +$CLICKHOUSE_CLIENT -q "DROP TABLE t_restore_fsync_enc SYNC" diff --git a/tests/queries/0_stateless/04652_direct_join_dictionary_sparse_key.reference b/tests/queries/0_stateless/04652_direct_join_dictionary_sparse_key.reference new file mode 100644 index 000000000000..88fe6660ee1f --- /dev/null +++ b/tests/queries/0_stateless/04652_direct_join_dictionary_sparse_key.reference @@ -0,0 +1,19 @@ +The join keys are really serialized Sparse +probe_sparse 1 +probe_sparse_arr 1 +Sparse key, key presence only +4000 3200 800 160 3200 800 +Sparse key, dictionary attribute read +336000 4000 3200 800 +Sparse default key 0 is really found +800 3200 160 +Sparse key, aggregation in order +1 1 50 +Sparse key replicated by ARRAY JOIN, attribute read +378000 4500 900 3600 +Sparse mapping equals dense mapping +1 +Sparse mapping equals hash join mapping +1 +Direct join is still chosen for the sparse key +1 diff --git a/tests/queries/0_stateless/04652_direct_join_dictionary_sparse_key.sql b/tests/queries/0_stateless/04652_direct_join_dictionary_sparse_key.sql new file mode 100644 index 000000000000..7c46f5f6b907 --- /dev/null +++ b/tests/queries/0_stateless/04652_direct_join_dictionary_sparse_key.sql @@ -0,0 +1,105 @@ +-- Tags: no-parallel-replicas +-- The direct join algorithm is not available with parallel replicas. + +DROP DICTIONARY IF EXISTS dict_sparse_key; +DROP TABLE IF EXISTS probe_sparse; +DROP TABLE IF EXISTS probe_dense; +DROP TABLE IF EXISTS probe_sparse_arr; +DROP TABLE IF EXISTS dict_source; + +-- The attribute is never 0, so with join_use_nulls = 0 the value `r.v = 0` means "key not found". +CREATE TABLE dict_source (j UInt64, v UInt64) ENGINE = MergeTree ORDER BY j; +INSERT INTO dict_source SELECT number, (number + 1) * 10 FROM numbers(20); + +CREATE DICTIONARY dict_sparse_key (j UInt64, v UInt64) PRIMARY KEY j +SOURCE(CLICKHOUSE(TABLE 'dict_source' DB currentDatabase())) LAYOUT(FLAT()) LIFETIME(MIN 0 MAX 0); + +-- `j` is default-heavy, so it is serialized Sparse. Keys 0..19 are in the dictionary, 20..24 are not. +CREATE TABLE probe_sparse (k UInt32, j UInt64) ENGINE = MergeTree ORDER BY k +SETTINGS ratio_of_defaults_for_sparse_serialization = 0.0; +INSERT INTO probe_sparse SELECT number % 50, number % 25 FROM numbers(4000); + +-- The same data written densely, used as the expected result below. +CREATE TABLE probe_dense (k UInt32, j UInt64) ENGINE = MergeTree ORDER BY k +SETTINGS ratio_of_defaults_for_sparse_serialization = 1.0; +INSERT INTO probe_dense SELECT number % 50, number % 25 FROM numbers(4000); + +-- A sparse `j` carried through ARRAY JOIN, so the join key arrives replicated rather than plain. +CREATE TABLE probe_sparse_arr (j UInt64, arr Array(UInt8)) ENGINE = MergeTree ORDER BY tuple() +SETTINGS ratio_of_defaults_for_sparse_serialization = 0.0; +INSERT INTO probe_sparse_arr SELECT number % 25, [1, 2, 3] FROM numbers(1500); + +SELECT 'The join keys are really serialized Sparse'; +SELECT table, countIf(serialization_kind = 'Sparse') > 0 FROM system.parts_columns +WHERE database = currentDatabase() AND table IN ('probe_sparse', 'probe_sparse_arr') + AND column = 'j' AND active +GROUP BY table ORDER BY table; + +SET join_algorithm = 'direct'; +-- Pinned, not left to randomization: a miss must read as 0 rather than NULL below. +SET join_use_nulls = 0; + +-- `r.j` observes which keys were found; the two `l.j` counts catch a key classified the wrong way. +SELECT 'Sparse key, key presence only'; +SELECT count(), countIf(r.v != 0), countIf(r.v = 0), countIf(l.j = 0 AND r.v = 10), + countIf(l.j < 20 AND r.j = l.j), countIf(l.j >= 20 AND r.j = 0) +FROM probe_sparse AS l LEFT JOIN dict_sparse_key AS r ON l.j = r.j; + +-- `r.v = (l.j + 1) * 10` restates the dictionary contents, so values swapped between keys are +-- caught even though the sum stays the same. +SELECT 'Sparse key, dictionary attribute read'; +SELECT sum(r.v), count(), countIf(r.v = (l.j + 1) * 10), countIf(r.v = 0 AND l.j >= 20) +FROM probe_sparse AS l LEFT JOIN dict_sparse_key AS r ON l.j = r.j; + +-- Key 0 is the sparse default and an unmatched right key is 0 too, so join_use_nulls = 1 is what +-- separates a found key 0 from a lost one here. +SELECT 'Sparse default key 0 is really found'; +SELECT countIf(r.j IS NULL), countIf(l.j < 20 AND r.j IS NOT NULL), countIf(l.j = 0 AND r.j = 0) +FROM probe_sparse AS l LEFT JOIN dict_sparse_key AS r ON l.j = r.j +SETTINGS join_use_nulls = 1; + +SELECT 'Sparse key, aggregation in order'; +SELECT max(u), min(u), count() FROM +( + SELECT l.k, uniqExact(l.k) AS u FROM probe_sparse AS l LEFT JOIN dict_sparse_key AS r ON l.j = r.j GROUP BY l.k +) +SETTINGS optimize_aggregation_in_order = 1, max_threads = 1; + +SELECT 'Sparse key replicated by ARRAY JOIN, attribute read'; +SELECT sum(r.v), count(), countIf(r.v = 0), countIf(r.v = (l.j + 1) * 10) +FROM (SELECT j FROM probe_sparse_arr ARRAY JOIN arr) AS l +LEFT JOIN dict_sparse_key AS r ON l.j = r.j; + +-- Comparing the whole per-key mapping, not a total: a total survives values swapped between keys. +SELECT 'Sparse mapping equals dense mapping'; +SELECT + (SELECT arraySort(groupArray((j, v, c))) FROM + (SELECT l.j AS j, r.v AS v, count() AS c FROM probe_sparse AS l + LEFT JOIN dict_sparse_key AS r ON l.j = r.j GROUP BY l.j, r.v)) + = (SELECT arraySort(groupArray((j, v, c))) FROM + (SELECT l.j AS j, r.v AS v, count() AS c FROM probe_dense AS l + LEFT JOIN dict_sparse_key AS r ON l.j = r.j GROUP BY l.j, r.v)); + +SELECT 'Sparse mapping equals hash join mapping'; +SELECT + (SELECT arraySort(groupArray((j, v, c))) FROM + (SELECT l.j AS j, r.v AS v, count() AS c FROM probe_sparse AS l + LEFT JOIN dict_sparse_key AS r ON l.j = r.j GROUP BY l.j, r.v)) + = (SELECT arraySort(groupArray((j, v, c))) FROM + (SELECT l.j AS j, r.v AS v, count() AS c FROM probe_sparse AS l + LEFT JOIN dict_sparse_key AS r ON l.j = r.j GROUP BY l.j, r.v + SETTINGS join_algorithm = 'hash')); + +SELECT 'Direct join is still chosen for the sparse key'; +SELECT count() > 0 FROM +( + EXPLAIN actions = 1 + SELECT count() FROM probe_sparse AS l LEFT JOIN dict_sparse_key AS r ON l.j = r.j +) +WHERE explain ILIKE '%Algorithm: DirectKeyValueJoin%'; + +DROP DICTIONARY dict_sparse_key; +DROP TABLE probe_sparse; +DROP TABLE probe_dense; +DROP TABLE probe_sparse_arr; +DROP TABLE dict_source; diff --git a/tests/queries/0_stateless/04743_trivial_count_array_join_argument.reference b/tests/queries/0_stateless/04743_trivial_count_array_join_argument.reference new file mode 100644 index 000000000000..afa98c5b90cd --- /dev/null +++ b/tests/queries/0_stateless/04743_trivial_count_array_join_argument.reference @@ -0,0 +1,20 @@ +6 +6 +5 +11 +6 +6 +5 +11 +3 +3 +3 +3 +6 +0 +1 +1 +1 +6 +6 +3 diff --git a/tests/queries/0_stateless/04743_trivial_count_array_join_argument.sh b/tests/queries/0_stateless/04743_trivial_count_array_join_argument.sh new file mode 100755 index 000000000000..d67c25798e69 --- /dev/null +++ b/tests/queries/0_stateless/04743_trivial_count_array_join_argument.sh @@ -0,0 +1,71 @@ +#!/usr/bin/env bash +# Tags: no-old-analyzer +# no-old-analyzer: The plan assertions describe applyTrivialCountIfPossible; the old analyzer decides trivial count in TreeRewriter + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +${CLICKHOUSE_CLIENT} -q " +DROP TABLE IF EXISTS t_04743; +CREATE TABLE t_04743 (A Array(UInt32), B Array(UInt32), n UInt32) ENGINE = MergeTree ORDER BY tuple(); +INSERT INTO t_04743 VALUES ([1,2,3],[1,2],1), ([4,5],[],2), ([6],[7,8,9],3); + +-- arrayJoin in the aggregate argument multiplies rows, so the stored row count (3) is not the answer +SELECT count(arrayJoin(A)) FROM t_04743 SETTINGS optimize_trivial_count_query = 1; +SELECT count(arrayJoin(A) + 1) FROM t_04743 SETTINGS optimize_trivial_count_query = 1; +SELECT count(arrayJoin(B)) FROM t_04743 SETTINGS optimize_trivial_count_query = 1; +SELECT count(arrayJoin(arrayJoin([A, B]))) FROM t_04743 SETTINGS optimize_trivial_count_query = 1; + +-- the same values with the optimization off +SELECT count(arrayJoin(A)) FROM t_04743 SETTINGS optimize_trivial_count_query = 0; +SELECT count(arrayJoin(A) + 1) FROM t_04743 SETTINGS optimize_trivial_count_query = 0; +SELECT count(arrayJoin(B)) FROM t_04743 SETTINGS optimize_trivial_count_query = 0; +SELECT count(arrayJoin(arrayJoin([A, B]))) FROM t_04743 SETTINGS optimize_trivial_count_query = 0; + +-- aggregates without arrayJoin keep the optimization +SELECT count() FROM t_04743 SETTINGS optimize_trivial_count_query = 1; +SELECT count(*) FROM t_04743 SETTINGS optimize_trivial_count_query = 1; +SELECT count(1) FROM t_04743 SETTINGS optimize_trivial_count_query = 1; +SELECT count(n) FROM t_04743 SETTINGS optimize_trivial_count_query = 1; +SELECT count() FROM t_04743 ARRAY JOIN A SETTINGS optimize_trivial_count_query = 1; + +-- plans: the optimization is refused for the arrayJoin argument and kept otherwise +SELECT count() > 0 FROM (EXPLAIN SELECT count(arrayJoin(A)) FROM t_04743 SETTINGS optimize_trivial_count_query = 1) +WHERE explain ILIKE '%Optimized trivial count%'; +SELECT count() > 0 FROM (EXPLAIN SELECT count(arrayJoin(A)) FROM t_04743 SETTINGS optimize_trivial_count_query = 1) +WHERE explain ILIKE '%ReadFromMergeTree%'; +SELECT count() > 0 FROM (EXPLAIN SELECT count() FROM t_04743 SETTINGS optimize_trivial_count_query = 1) +WHERE explain ILIKE '%Optimized trivial count%'; +SELECT count() > 0 FROM (EXPLAIN SELECT count(n) FROM t_04743 SETTINGS optimize_trivial_count_query = 1) +WHERE explain ILIKE '%Optimized trivial count%'; + +DROP TABLE t_04743; +" + +# file() counts inside read() when the flag is set, and it reaches that path even though +# totalRows() is unknown, so it distinguishes the guard's position from a later one. +unique_name=${CLICKHOUSE_TEST_UNIQUE_NAME} +tmp_dir=${USER_FILES_PATH}/${unique_name} +mkdir -p "${tmp_dir}" +rm -rf "${tmp_dir:?}"/* + +cat > "${tmp_dir}/arr.csv" <<'EOF' +"[1,2,3]" +"[4,5]" +"[6]" +EOF + +chmod 777 "${tmp_dir}" +chmod 777 "${tmp_dir}/arr.csv" + +${CLICKHOUSE_CLIENT} -q " +SELECT count(arrayJoin(A)) FROM file('${unique_name}/arr.csv', 'CSV', 'A Array(UInt32)') +SETTINGS optimize_trivial_count_query = 1, optimize_count_from_files = 1; +SELECT count(arrayJoin(A)) FROM file('${unique_name}/arr.csv', 'CSV', 'A Array(UInt32)') +SETTINGS optimize_trivial_count_query = 0, optimize_count_from_files = 1; +SELECT count() FROM file('${unique_name}/arr.csv', 'CSV', 'A Array(UInt32)') +SETTINGS optimize_trivial_count_query = 1, optimize_count_from_files = 1; +" + +rm -rf "${tmp_dir:?}" diff --git a/tests/queries/0_stateless/04812_merge_row_policy_shared_ast.reference b/tests/queries/0_stateless/04812_merge_row_policy_shared_ast.reference new file mode 100644 index 000000000000..2ec92d18c89b --- /dev/null +++ b/tests/queries/0_stateless/04812_merge_row_policy_shared_ast.reference @@ -0,0 +1,4 @@ +4 +4 +8 +8 diff --git a/tests/queries/0_stateless/04812_merge_row_policy_shared_ast.sh b/tests/queries/0_stateless/04812_merge_row_policy_shared_ast.sh new file mode 100755 index 000000000000..5303466321fb --- /dev/null +++ b/tests/queries/0_stateless/04812_merge_row_policy_shared_ast.sh @@ -0,0 +1,48 @@ +#!/usr/bin/env bash + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +# `RowPolicyFilter::expression` is the parsed policy condition owned by `RowPolicyCache` and is shared by +# every query that reads the table. `ReadFromMerge::RowPolicyData` used to hand it straight to +# `TreeRewriter`, which rewrites the AST it is given in place and substitutes the results of scalar +# subqueries for the subqueries themselves. Reading a `Merge` table therefore froze the value of a scalar +# subquery inside the policy in the cache, for the rest of the server's lifetime and for every user. +# The same in-place rewrite was reported by ThreadSanitizer as a data race on the shared AST. +# +# This is a shell test because the table inside the policy's subquery has to be qualified with the +# database name: the policy condition is analyzed anew for every read, including reads on the remote +# side of a parallel-replicas query, where the session default database is not the test database. + +$CLICKHOUSE_CLIENT -q " + DROP TABLE IF EXISTS t_04812_src; + DROP TABLE IF EXISTS t_04812_limit; + DROP TABLE IF EXISTS t_04812_merge; + DROP ROW POLICY IF EXISTS p_04812 ON t_04812_src; + + CREATE TABLE t_04812_src (x UInt64) ENGINE = MergeTree ORDER BY x; + INSERT INTO t_04812_src SELECT number FROM numbers(10); + + CREATE TABLE t_04812_limit (v UInt64) ENGINE = MergeTree ORDER BY v; + INSERT INTO t_04812_limit VALUES (3); + + CREATE ROW POLICY p_04812 ON t_04812_src USING x <= (SELECT max(v) FROM ${CLICKHOUSE_DATABASE}.t_04812_limit) TO ALL; + CREATE TABLE t_04812_merge (x UInt64) ENGINE = Merge(currentDatabase(), '^t_04812_src\$'); + + -- The policy admits 0, 1, 2, 3. + SELECT count() FROM t_04812_merge; + SELECT count() FROM t_04812_src; + + INSERT INTO t_04812_limit VALUES (7); + + -- The policy now admits 0 .. 7. Reading through Merge used to keep answering 4, and so did the direct + -- read, because the Merge read above had replaced the subquery with its value in the cached policy AST. + SELECT count() FROM t_04812_merge; + SELECT count() FROM t_04812_src; + + DROP ROW POLICY p_04812 ON t_04812_src; + DROP TABLE t_04812_merge; + DROP TABLE t_04812_limit; + DROP TABLE t_04812_src; +" diff --git a/tests/queries/0_stateless/04815_parallel_replicas_custom_key_merge_distributed_child.reference b/tests/queries/0_stateless/04815_parallel_replicas_custom_key_merge_distributed_child.reference new file mode 100644 index 000000000000..ba18b0fc3c08 --- /dev/null +++ b/tests/queries/0_stateless/04815_parallel_replicas_custom_key_merge_distributed_child.reference @@ -0,0 +1,4 @@ +300000 +14999850000 +47 9 +300000 diff --git a/tests/queries/0_stateless/04815_parallel_replicas_custom_key_merge_distributed_child.sql b/tests/queries/0_stateless/04815_parallel_replicas_custom_key_merge_distributed_child.sql new file mode 100644 index 000000000000..57f1540043c9 --- /dev/null +++ b/tests/queries/0_stateless/04815_parallel_replicas_custom_key_merge_distributed_child.sql @@ -0,0 +1,33 @@ +-- A `Merge` table over a `Distributed` child plans all of its children up to `WithMergeableState` +-- through an interpreter. The custom-key parallel replicas read replaces a child plan with a remote +-- read at the fixed stage `WithMergeableStateAfterAggregationAndLimit`, so the parent received +-- finalized (post-aggregation, post-LIMIT) data where it expected partial aggregation states: +-- `CANNOT_CONVERT_TYPE` for `count`, and an exception about a missing `AggregatedChunkInfo` in +-- `GroupingAggregatedTransform` when the finalized type coincides with the state type structurally. +-- https://github.com/ClickHouse/ClickHouse/issues/113741 + +DROP TABLE IF EXISTS t_mrg_ck_1; +DROP TABLE IF EXISTS t_mrg_ck_2; +DROP TABLE IF EXISTS t_mrg_ck_3; + +CREATE TABLE t_mrg_ck_1 (k UInt64) ENGINE = MergeTree ORDER BY k AS SELECT number FROM numbers(100000); +CREATE TABLE t_mrg_ck_2 (k UInt64) ENGINE = MergeTree ORDER BY k AS SELECT number FROM numbers(100000); +CREATE TABLE t_mrg_ck_3 (k UInt64) ENGINE = Distributed('test_shard_localhost', currentDatabase(), 't_mrg_ck_1'); + +SET enable_analyzer = 1; +SET enable_parallel_replicas = 1, max_parallel_replicas = 3, + cluster_for_parallel_replicas = 'test_cluster_one_shard_three_replicas_localhost', + parallel_replicas_for_non_replicated_merge_tree = 1, + parallel_replicas_mode = 'custom_key_sampling', parallel_replicas_custom_key = 'k'; + +SELECT count() FROM merge(currentDatabase(), '^t_mrg_ck_'); +SELECT sum(k) FROM merge(currentDatabase(), '^t_mrg_ck_') GROUP BY ALL; +SELECT 47, quantileExactInclusive(visibleWidth(['1', '2'])) IGNORE NULLS FROM merge(currentDatabase(), '^t_mrg_ck_') GROUP BY ALL LIMIT 973; + +SET parallel_replicas_mode = 'custom_key_range', parallel_replicas_custom_key_range_upper = 100000; + +SELECT count() FROM merge(currentDatabase(), '^t_mrg_ck_'); + +DROP TABLE t_mrg_ck_1; +DROP TABLE t_mrg_ck_2; +DROP TABLE t_mrg_ck_3; diff --git a/tests/queries/0_stateless/04881_low_cardinality_default_value_needle.reference b/tests/queries/0_stateless/04881_low_cardinality_default_value_needle.reference new file mode 100644 index 000000000000..cf7f820a545e --- /dev/null +++ b/tests/queries/0_stateless/04881_low_cardinality_default_value_needle.reference @@ -0,0 +1,90 @@ +-- { echo } +SET allow_suspicious_low_cardinality_types = 1; +-- A needle equal to the element type's default value must be found. Every row prints the +-- LowCardinality answer beside the same query over a plain array. +SELECT has(CAST(['', 'a'], 'Array(LowCardinality(String))'), '') AS lc, has(CAST(['', 'a'], 'Array(String)'), '') AS oracle; +1 1 +SELECT indexOf(materialize(CAST(['a', ''], 'Array(LowCardinality(String))')), '') AS lc, indexOf(materialize(CAST(['a', ''], 'Array(String)')), '') AS oracle; +2 2 +SELECT countEqual(materialize(CAST(['', 'a', ''], 'Array(LowCardinality(String))')), '') AS lc, countEqual(materialize(CAST(['', 'a', ''], 'Array(String)')), '') AS oracle; +2 2 +SELECT has(materialize(CAST([0, 5], 'Array(LowCardinality(UInt8))')), 0) AS lc, has(materialize(CAST([0, 5], 'Array(UInt8)')), 0) AS oracle; +1 1 +SELECT has(materialize(CAST([CAST('', 'FixedString(3)'), CAST('abc', 'FixedString(3)')], 'Array(LowCardinality(FixedString(3)))')), CAST('', 'FixedString(3)')) AS lc, has(materialize(CAST([CAST('', 'FixedString(3)'), CAST('abc', 'FixedString(3)')], 'Array(FixedString(3))')), CAST('', 'FixedString(3)')) AS oracle; +1 1 +-- An Enum needle compares to a FixedString element as a string, where the element type's own padding +-- is not a difference. +SELECT has(materialize(CAST([CAST('', 'FixedString(3)'), CAST('abc', 'FixedString(3)')], 'Array(LowCardinality(FixedString(3)))')), CAST('', 'Enum8('''' = 0, ''o'' = 1)')) AS lc, has(materialize(CAST([CAST('', 'FixedString(3)'), CAST('abc', 'FixedString(3)')], 'Array(FixedString(3))')), CAST('', 'Enum8('''' = 0, ''o'' = 1)')) AS oracle; +1 1 +-- The Map key and value dictionaries are the second call site. +SELECT mapContainsKey(materialize(CAST(map('', 'v_empty', 'k', 'v_k'), 'Map(LowCardinality(String), String)')), '') AS lc, mapContainsKey(materialize(CAST(map('', 'v_empty', 'k', 'v_k'), 'Map(String, String)')), '') AS oracle; +1 1 +SELECT materialize(CAST(map('', 'v_empty', 'k', 'v_k'), 'Map(LowCardinality(String), String)'))[''] AS lc, materialize(CAST(map('', 'v_empty', 'k', 'v_k'), 'Map(String, String)'))[''] AS oracle; +v_empty v_empty +SELECT mapContainsValue(materialize(CAST(map('k', '', 'j', 'v'), 'Map(String, LowCardinality(String))')), '') AS lc, mapContainsValue(materialize(CAST(map('k', '', 'j', 'v'), 'Map(String, String)')), '') AS oracle; +1 1 +-- A constant the element type cannot represent equals no element, while one that survives the cast +-- still finds the default element. The timezone is pinned because the cast to Date drops the +-- needle's time of day and which day that lands on is offset-dependent. +SELECT has(materialize(CAST([0, 5], 'Array(LowCardinality(UInt8))')), 256) AS lc, has(materialize(CAST([0, 5], 'Array(UInt8)')), 256) AS oracle; +0 0 +SELECT mapContainsKey(materialize(CAST(map(0, 'a', 5, 'b'), 'Map(LowCardinality(UInt8), String)')), 256) AS lc, mapContainsKey(materialize(CAST(map(0, 'a', 5, 'b'), 'Map(UInt8, String)')), 256) AS oracle; +0 0 +SELECT has(materialize(CAST([toDate('1970-01-01'), toDate('2020-01-01')], 'Array(LowCardinality(Date))')), toDateTime('1970-01-01 00:00:05')) AS lc, has(materialize(CAST([toDate('1970-01-01'), toDate('2020-01-01')], 'Array(Date)')), toDateTime('1970-01-01 00:00:05')) AS oracle SETTINGS session_timezone = 'UTC'; +0 0 +SELECT has(materialize(CAST([0, 5], 'Array(LowCardinality(UInt8))')), toUInt64(0)) AS widened_needle, has(materialize(CAST([0, 1.5], 'Array(LowCardinality(Float64))')), toUInt8(0)) AS integral_needle; +1 1 +-- An IPv4 element represents a UInt32 needle exactly, and the two have no accurate cast between them, +-- so representability is decided by comparing the needle against its own cast image. +SELECT has(materialize(CAST([toIPv4('0.0.0.0'), toIPv4('1.2.3.4')], 'Array(LowCardinality(IPv4))')), toUInt32(0)) AS lc, has(materialize(CAST([toIPv4('0.0.0.0'), toIPv4('1.2.3.4')], 'Array(IPv4)')), toUInt32(0)) AS oracle; +1 1 +-- A FixedString needle is padded to its own width and equality ignores that padding. The plain-array +-- arrayFilter oracle of the original test is dropped on 25.8: the vectorized `String = FixedString` +-- comparison on this branch does not ignore the padding yet, so it reads 0 even though the +-- constant-folded `'' = CAST('', 'FixedString(4)')` reads 1. +SELECT has(materialize(CAST(['', 'xy'], 'Array(LowCardinality(String))')), CAST('', 'FixedString(4)')) AS lc; +1 +-- -0.0 and 0.0 are equal but a text format stores them apart, so either zero as a needle must match +-- either spelling, and a count must see both. stored_bits is asserted in the same row: if it ever +-- reads 0 the array no longer holds -0.0 and the arm is void. +SELECT arrayMap(x -> reinterpretAsUInt64(x), a) AS stored_bits, has(a, toFloat64(0)) AS positive_zero_needle, has(a, reinterpretAsFloat64(toUInt64(9223372036854775808))) AS negative_zero_needle, length(arrayFilter(x -> x = toFloat64(0), a)) AS oracle FROM format(JSONEachRow, 'a Array(LowCardinality(Float64))', '{"a":[-0.0,1.5]}'); +[9223372036854775808,4609434218613702656] 1 1 1 +SELECT arrayMap(x -> reinterpretAsUInt64(x), a) AS stored_bits, indexOf(a, toFloat64(0)) AS positive_zero_needle, indexOf(a, reinterpretAsFloat64(toUInt64(9223372036854775808))) AS negative_zero_needle, indexOf(arrayMap(x -> toFloat64(x), a), toFloat64(0)) AS oracle FROM format(JSONEachRow, 'a Array(LowCardinality(Float64))', '{"a":[1.5,-0.0]}'); +[4609434218613702656,9223372036854775808] 2 2 2 +SELECT arrayMap(x -> reinterpretAsUInt64(x), a) AS stored_bits, countEqual(a, toFloat64(0)) AS positive_zero_needle, countEqual(a, reinterpretAsFloat64(toUInt64(9223372036854775808))) AS negative_zero_needle, length(arrayFilter(x -> x = toFloat64(0), a)) AS oracle FROM format(JSONEachRow, 'a Array(LowCardinality(Float64))', '{"a":[0.0,-0.0,1.5]}'); +[0,9223372036854775808,4609434218613702656] 2 2 2 +SELECT arrayMap(x -> reinterpretAsUInt32(x), a) AS stored_bits, has(a, toFloat32(0)) AS positive_zero_needle, has(a, reinterpretAsFloat32(toUInt32(2147483648))) AS negative_zero_needle, length(arrayFilter(x -> x = toFloat32(0), a)) AS oracle FROM format(JSONEachRow, 'a Array(LowCardinality(Float32))', '{"a":[-0.0,1.5]}'); +[2147483648,1069547520] 1 1 1 +-- A CAST array is folded onto the default slot, so its -0.0 element is stored as +0.0. +SELECT arrayMap(x -> reinterpretAsUInt64(x), materialize(CAST([reinterpretAsFloat64(toUInt64(9223372036854775808)), 1.5], 'Array(LowCardinality(Float64))'))) AS stored_bits, has(materialize(CAST([reinterpretAsFloat64(toUInt64(9223372036854775808)), 1.5], 'Array(LowCardinality(Float64))')), toFloat64(0)) AS positive_zero_needle, has(materialize(CAST([reinterpretAsFloat64(toUInt64(9223372036854775808)), 1.5], 'Array(LowCardinality(Float64))')), reinterpretAsFloat64(toUInt64(9223372036854775808))) AS negative_zero_needle; +[0,4609434218613702656] 1 1 +SELECT has(materialize(CAST([0, NULL, 1.5], 'Array(LowCardinality(Nullable(Float64)))')), reinterpretAsFloat64(toUInt64(9223372036854775808))) AS lc, length(arrayFilter(x -> x = reinterpretAsFloat64(toUInt64(9223372036854775808)), materialize(CAST([0, NULL, 1.5], 'Array(LowCardinality(Nullable(Float64)))')))) AS oracle; +1 1 +-- An Enum needle reaches a float element through its underlying number, so it must be declined too. +SELECT arrayMap(x -> reinterpretAsUInt64(x), a) AS stored_bits, has(a, CAST('z', 'Enum8(\'z\' = 0, \'o\' = 1)')) AS enum8_needle, indexOf(a, CAST('z', 'Enum8(\'z\' = 0, \'o\' = 1)')) AS enum8_index, countEqual(a, CAST('z', 'Enum16(\'z\' = 0, \'o\' = 1)')) AS enum16_count, length(arrayFilter(x -> x = toFloat64(0), a)) AS oracle FROM format(JSONEachRow, 'a Array(LowCardinality(Float64))', '{"a":[-0.0,1.5]}'); +[9223372036854775808,4609434218613702656] 1 1 1 1 +-- Controls that must not move. +SELECT has(materialize(CAST(['', 'a'], 'Array(LowCardinality(String))')), 'zzz') AS absent_needle, has(materialize(CAST(['a', 'b'], 'Array(LowCardinality(String))')), '') AS default_absent; +0 0 +SELECT has(materialize(CAST(['', 'a'], 'Array(LowCardinality(String))')), materialize('')) AS non_const_needle, indexOfAssumeSorted(materialize(CAST(['', 'a'], 'Array(LowCardinality(String))')), '') AS assume_sorted; +1 1 +-- Two answers that only the dictionary shortcut produces, so a build that stopped taking it would +-- move them. Both disagree with the plain-array oracle printed beside them, and both are known +-- defects of the value comparison tracked elsewhere, not of the lookup this test covers: a NaN is +-- one dictionary entry but never equal to itself, and a negative needle is compared to an unsigned +-- element as a raw number. +SELECT has(materialize(CAST([nan, 1.5], 'Array(LowCardinality(Float64))')), nan) AS nan_needle, has(materialize(CAST([nan, 1.5], 'Array(Float64)')), nan) AS oracle; +1 0 +SELECT has(materialize(CAST([0, 255], 'Array(LowCardinality(UInt8))')), toInt8(-1)) AS negative_needle, has(materialize(CAST([0, 255], 'Array(UInt8)')), toInt8(-1)) AS oracle; +1 0 +-- LowCardinality(Nullable(T)): a NULL needle finds the NULL element and a default needle finds the default one. +SELECT indexOf(materialize(CAST(['a', NULL, ''], 'Array(LowCardinality(Nullable(String)))')), NULL) AS null_needle, indexOf(materialize(CAST(['a', NULL, ''], 'Array(LowCardinality(Nullable(String)))')), '') AS default_needle; +2 3 +-- Reached through a real table read rather than a constant-folded literal. +DROP TABLE IF EXISTS t_04881; +CREATE TABLE t_04881 (a Array(LowCardinality(String)), m Map(LowCardinality(String), String)) ENGINE = MergeTree ORDER BY tuple(); +INSERT INTO t_04881 VALUES (['', 'a'], map('', 'v_empty', 'k', 'v_k')), (['x', 'y'], map('x', 'v_x', 'y', 'v_y')); +SELECT a, has(a, '') AS has_empty, indexOf(a, '') AS idx_empty, mapContainsKey(m, '') AS key_empty, m[''] AS subscript_empty FROM t_04881 ORDER BY a; +['','a'] 1 1 1 v_empty +['x','y'] 0 0 0 +DROP TABLE t_04881; diff --git a/tests/queries/0_stateless/04881_low_cardinality_default_value_needle.sql b/tests/queries/0_stateless/04881_low_cardinality_default_value_needle.sql new file mode 100644 index 000000000000..4c8e9ba3ee6e --- /dev/null +++ b/tests/queries/0_stateless/04881_low_cardinality_default_value_needle.sql @@ -0,0 +1,72 @@ +-- { echo } +SET allow_suspicious_low_cardinality_types = 1; + +-- A needle equal to the element type's default value must be found. Every row prints the +-- LowCardinality answer beside the same query over a plain array. +SELECT has(CAST(['', 'a'], 'Array(LowCardinality(String))'), '') AS lc, has(CAST(['', 'a'], 'Array(String)'), '') AS oracle; +SELECT indexOf(materialize(CAST(['a', ''], 'Array(LowCardinality(String))')), '') AS lc, indexOf(materialize(CAST(['a', ''], 'Array(String)')), '') AS oracle; +SELECT countEqual(materialize(CAST(['', 'a', ''], 'Array(LowCardinality(String))')), '') AS lc, countEqual(materialize(CAST(['', 'a', ''], 'Array(String)')), '') AS oracle; +SELECT has(materialize(CAST([0, 5], 'Array(LowCardinality(UInt8))')), 0) AS lc, has(materialize(CAST([0, 5], 'Array(UInt8)')), 0) AS oracle; +SELECT has(materialize(CAST([CAST('', 'FixedString(3)'), CAST('abc', 'FixedString(3)')], 'Array(LowCardinality(FixedString(3)))')), CAST('', 'FixedString(3)')) AS lc, has(materialize(CAST([CAST('', 'FixedString(3)'), CAST('abc', 'FixedString(3)')], 'Array(FixedString(3))')), CAST('', 'FixedString(3)')) AS oracle; +-- An Enum needle compares to a FixedString element as a string, where the element type's own padding +-- is not a difference. +SELECT has(materialize(CAST([CAST('', 'FixedString(3)'), CAST('abc', 'FixedString(3)')], 'Array(LowCardinality(FixedString(3)))')), CAST('', 'Enum8('''' = 0, ''o'' = 1)')) AS lc, has(materialize(CAST([CAST('', 'FixedString(3)'), CAST('abc', 'FixedString(3)')], 'Array(FixedString(3))')), CAST('', 'Enum8('''' = 0, ''o'' = 1)')) AS oracle; + +-- The Map key and value dictionaries are the second call site. +SELECT mapContainsKey(materialize(CAST(map('', 'v_empty', 'k', 'v_k'), 'Map(LowCardinality(String), String)')), '') AS lc, mapContainsKey(materialize(CAST(map('', 'v_empty', 'k', 'v_k'), 'Map(String, String)')), '') AS oracle; +SELECT materialize(CAST(map('', 'v_empty', 'k', 'v_k'), 'Map(LowCardinality(String), String)'))[''] AS lc, materialize(CAST(map('', 'v_empty', 'k', 'v_k'), 'Map(String, String)'))[''] AS oracle; +SELECT mapContainsValue(materialize(CAST(map('k', '', 'j', 'v'), 'Map(String, LowCardinality(String))')), '') AS lc, mapContainsValue(materialize(CAST(map('k', '', 'j', 'v'), 'Map(String, String)')), '') AS oracle; + +-- A constant the element type cannot represent equals no element, while one that survives the cast +-- still finds the default element. The timezone is pinned because the cast to Date drops the +-- needle's time of day and which day that lands on is offset-dependent. +SELECT has(materialize(CAST([0, 5], 'Array(LowCardinality(UInt8))')), 256) AS lc, has(materialize(CAST([0, 5], 'Array(UInt8)')), 256) AS oracle; +SELECT mapContainsKey(materialize(CAST(map(0, 'a', 5, 'b'), 'Map(LowCardinality(UInt8), String)')), 256) AS lc, mapContainsKey(materialize(CAST(map(0, 'a', 5, 'b'), 'Map(UInt8, String)')), 256) AS oracle; +SELECT has(materialize(CAST([toDate('1970-01-01'), toDate('2020-01-01')], 'Array(LowCardinality(Date))')), toDateTime('1970-01-01 00:00:05')) AS lc, has(materialize(CAST([toDate('1970-01-01'), toDate('2020-01-01')], 'Array(Date)')), toDateTime('1970-01-01 00:00:05')) AS oracle SETTINGS session_timezone = 'UTC'; +SELECT has(materialize(CAST([0, 5], 'Array(LowCardinality(UInt8))')), toUInt64(0)) AS widened_needle, has(materialize(CAST([0, 1.5], 'Array(LowCardinality(Float64))')), toUInt8(0)) AS integral_needle; +-- An IPv4 element represents a UInt32 needle exactly, and the two have no accurate cast between them, +-- so representability is decided by comparing the needle against its own cast image. +SELECT has(materialize(CAST([toIPv4('0.0.0.0'), toIPv4('1.2.3.4')], 'Array(LowCardinality(IPv4))')), toUInt32(0)) AS lc, has(materialize(CAST([toIPv4('0.0.0.0'), toIPv4('1.2.3.4')], 'Array(IPv4)')), toUInt32(0)) AS oracle; + +-- A FixedString needle is padded to its own width and equality ignores that padding. The plain-array +-- arrayFilter oracle of the original test is dropped on 25.8: the vectorized `String = FixedString` +-- comparison on this branch does not ignore the padding yet, so it reads 0 even though the +-- constant-folded `'' = CAST('', 'FixedString(4)')` reads 1. +SELECT has(materialize(CAST(['', 'xy'], 'Array(LowCardinality(String))')), CAST('', 'FixedString(4)')) AS lc; + +-- -0.0 and 0.0 are equal but a text format stores them apart, so either zero as a needle must match +-- either spelling, and a count must see both. stored_bits is asserted in the same row: if it ever +-- reads 0 the array no longer holds -0.0 and the arm is void. +SELECT arrayMap(x -> reinterpretAsUInt64(x), a) AS stored_bits, has(a, toFloat64(0)) AS positive_zero_needle, has(a, reinterpretAsFloat64(toUInt64(9223372036854775808))) AS negative_zero_needle, length(arrayFilter(x -> x = toFloat64(0), a)) AS oracle FROM format(JSONEachRow, 'a Array(LowCardinality(Float64))', '{"a":[-0.0,1.5]}'); +SELECT arrayMap(x -> reinterpretAsUInt64(x), a) AS stored_bits, indexOf(a, toFloat64(0)) AS positive_zero_needle, indexOf(a, reinterpretAsFloat64(toUInt64(9223372036854775808))) AS negative_zero_needle, indexOf(arrayMap(x -> toFloat64(x), a), toFloat64(0)) AS oracle FROM format(JSONEachRow, 'a Array(LowCardinality(Float64))', '{"a":[1.5,-0.0]}'); +SELECT arrayMap(x -> reinterpretAsUInt64(x), a) AS stored_bits, countEqual(a, toFloat64(0)) AS positive_zero_needle, countEqual(a, reinterpretAsFloat64(toUInt64(9223372036854775808))) AS negative_zero_needle, length(arrayFilter(x -> x = toFloat64(0), a)) AS oracle FROM format(JSONEachRow, 'a Array(LowCardinality(Float64))', '{"a":[0.0,-0.0,1.5]}'); +SELECT arrayMap(x -> reinterpretAsUInt32(x), a) AS stored_bits, has(a, toFloat32(0)) AS positive_zero_needle, has(a, reinterpretAsFloat32(toUInt32(2147483648))) AS negative_zero_needle, length(arrayFilter(x -> x = toFloat32(0), a)) AS oracle FROM format(JSONEachRow, 'a Array(LowCardinality(Float32))', '{"a":[-0.0,1.5]}'); + +-- A CAST array is folded onto the default slot, so its -0.0 element is stored as +0.0. +SELECT arrayMap(x -> reinterpretAsUInt64(x), materialize(CAST([reinterpretAsFloat64(toUInt64(9223372036854775808)), 1.5], 'Array(LowCardinality(Float64))'))) AS stored_bits, has(materialize(CAST([reinterpretAsFloat64(toUInt64(9223372036854775808)), 1.5], 'Array(LowCardinality(Float64))')), toFloat64(0)) AS positive_zero_needle, has(materialize(CAST([reinterpretAsFloat64(toUInt64(9223372036854775808)), 1.5], 'Array(LowCardinality(Float64))')), reinterpretAsFloat64(toUInt64(9223372036854775808))) AS negative_zero_needle; +SELECT has(materialize(CAST([0, NULL, 1.5], 'Array(LowCardinality(Nullable(Float64)))')), reinterpretAsFloat64(toUInt64(9223372036854775808))) AS lc, length(arrayFilter(x -> x = reinterpretAsFloat64(toUInt64(9223372036854775808)), materialize(CAST([0, NULL, 1.5], 'Array(LowCardinality(Nullable(Float64)))')))) AS oracle; + +-- An Enum needle reaches a float element through its underlying number, so it must be declined too. +SELECT arrayMap(x -> reinterpretAsUInt64(x), a) AS stored_bits, has(a, CAST('z', 'Enum8(\'z\' = 0, \'o\' = 1)')) AS enum8_needle, indexOf(a, CAST('z', 'Enum8(\'z\' = 0, \'o\' = 1)')) AS enum8_index, countEqual(a, CAST('z', 'Enum16(\'z\' = 0, \'o\' = 1)')) AS enum16_count, length(arrayFilter(x -> x = toFloat64(0), a)) AS oracle FROM format(JSONEachRow, 'a Array(LowCardinality(Float64))', '{"a":[-0.0,1.5]}'); + +-- Controls that must not move. +SELECT has(materialize(CAST(['', 'a'], 'Array(LowCardinality(String))')), 'zzz') AS absent_needle, has(materialize(CAST(['a', 'b'], 'Array(LowCardinality(String))')), '') AS default_absent; +SELECT has(materialize(CAST(['', 'a'], 'Array(LowCardinality(String))')), materialize('')) AS non_const_needle, indexOfAssumeSorted(materialize(CAST(['', 'a'], 'Array(LowCardinality(String))')), '') AS assume_sorted; + +-- Two answers that only the dictionary shortcut produces, so a build that stopped taking it would +-- move them. Both disagree with the plain-array oracle printed beside them, and both are known +-- defects of the value comparison tracked elsewhere, not of the lookup this test covers: a NaN is +-- one dictionary entry but never equal to itself, and a negative needle is compared to an unsigned +-- element as a raw number. +SELECT has(materialize(CAST([nan, 1.5], 'Array(LowCardinality(Float64))')), nan) AS nan_needle, has(materialize(CAST([nan, 1.5], 'Array(Float64)')), nan) AS oracle; +SELECT has(materialize(CAST([0, 255], 'Array(LowCardinality(UInt8))')), toInt8(-1)) AS negative_needle, has(materialize(CAST([0, 255], 'Array(UInt8)')), toInt8(-1)) AS oracle; + +-- LowCardinality(Nullable(T)): a NULL needle finds the NULL element and a default needle finds the default one. +SELECT indexOf(materialize(CAST(['a', NULL, ''], 'Array(LowCardinality(Nullable(String)))')), NULL) AS null_needle, indexOf(materialize(CAST(['a', NULL, ''], 'Array(LowCardinality(Nullable(String)))')), '') AS default_needle; + +-- Reached through a real table read rather than a constant-folded literal. +DROP TABLE IF EXISTS t_04881; +CREATE TABLE t_04881 (a Array(LowCardinality(String)), m Map(LowCardinality(String), String)) ENGINE = MergeTree ORDER BY tuple(); +INSERT INTO t_04881 VALUES (['', 'a'], map('', 'v_empty', 'k', 'v_k')), (['x', 'y'], map('x', 'v_x', 'y', 'v_y')); +SELECT a, has(a, '') AS has_empty, indexOf(a, '') AS idx_empty, mapContainsKey(m, '') AS key_empty, m[''] AS subscript_empty FROM t_04881 ORDER BY a; +DROP TABLE t_04881; diff --git a/tests/queries/0_stateless/04895_parquet_dictionary_source_prefetch_lifetime.reference b/tests/queries/0_stateless/04895_parquet_dictionary_source_prefetch_lifetime.reference new file mode 100644 index 000000000000..52790e8f4d2d --- /dev/null +++ b/tests/queries/0_stateless/04895_parquet_dictionary_source_prefetch_lifetime.reference @@ -0,0 +1,6 @@ +1 +1 +1 +1 +1 +alive 1 diff --git a/tests/queries/0_stateless/04895_parquet_dictionary_source_prefetch_lifetime.sh b/tests/queries/0_stateless/04895_parquet_dictionary_source_prefetch_lifetime.sh new file mode 100755 index 000000000000..bce61fb96aa0 --- /dev/null +++ b/tests/queries/0_stateless/04895_parquet_dictionary_source_prefetch_lifetime.sh @@ -0,0 +1,60 @@ +#!/usr/bin/env bash +# Tags: no-fasttest +# no-fasttest: needs the Parquet format which is not built in fasttest. + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +# A dictionary reading Parquet from a local file hands the ReadBuffer to the input format, which owns +# it. When the read throws, the pipeline releases that buffer on the way out while the format's +# background prefetch and decode tasks may still be reading through it. The error must surface every +# time and the server must stay alive; under a sanitizer build the buffer must not be read after +# release. + +DICT="d_${CLICKHOUSE_DATABASE}" +REL="${CLICKHOUSE_DATABASE}/prefetch_lifetime.parquet" + +# Small row groups so there are many read ranges, hence many queued tasks at throw time. +${CLICKHOUSE_CLIENT} --query=" + insert into function file('${REL}', Parquet, 'key UInt64, val String') + select number, repeat('y', 400) from numbers(2000000) + settings engine_file_truncate_on_insert = 1, output_format_parquet_row_group_size = 5000, + output_format_parquet_compression_method = 'none'; +" + +# The dictionary FILE source needs the actual absolute path. The `user_files_path` setting can be +# empty, so derive the resolved path from the table function instead of composing it from that +# setting. +ABS=$(${CLICKHOUSE_CLIENT} --query="select _path from file('${REL}', Parquet) limit 1") + +# `val` is Int64 in the dictionary but holds strings in the file, so the Parquet read throws +# mid-flight, which is what makes the pipeline tear down while tasks are still running. +# +# The settings have to be on the dictionary, not on the query: a dictionary loads in the global +# context, and the `file` source only picks up this SETTINGS clause. min_bytes_for_seek = 1 stops +# range coalescing, so each range becomes its own task, and the prefetch pool has to exist for those +# tasks to run in the background rather than inline on the decoding thread. +${CLICKHOUSE_CLIENT} --query=" + create dictionary ${DICT} (key UInt64, val Int64) primary key key + source(file(path '${ABS}' format 'Parquet')) + layout(flat(max_array_size 5000000)) lifetime(0) + settings(max_download_threads = 32, max_parsing_threads = 32, + input_format_parquet_use_native_reader_v3 = 1, + input_format_parquet_local_file_min_bytes_for_seek = 1, + input_format_parquet_enable_row_group_prefetch = 1); +" + +# A forced reload, because a plain dictGet would replay the first load's cached exception instead of +# reading the file again. +for _ in 1 2 3 4 5; do + ${CLICKHOUSE_CLIENT} --log_comment="${DICT}_reload" --query="system reload dictionary ${DICT}" 2>&1 \ + | grep -c -m1 -F 'CANNOT_PARSE_TEXT' +done + +# The server survived every attempt and still answers. +${CLICKHOUSE_CLIENT} --query="select 'alive', count() from system.dictionaries where database = currentDatabase() and name = '${DICT}'" + +${CLICKHOUSE_CLIENT} --query="drop dictionary ${DICT}" +${CLICKHOUSE_CLIENT} --query="select * from file('${REL}', Parquet) where 0 format Null" 2>/dev/null +rm -f "${ABS}" diff --git a/tests/queries/0_stateless/04927_uniq_injective_tuple_nullable.reference b/tests/queries/0_stateless/04927_uniq_injective_tuple_nullable.reference new file mode 100644 index 000000000000..a5eeb060a13f --- /dev/null +++ b/tests/queries/0_stateless/04927_uniq_injective_tuple_nullable.reference @@ -0,0 +1,28 @@ +tuple +3 3 3 3 3 +3 3 3 3 3 +nested, LowCardinality and multiple arguments +3 3 +3 3 +2 +2 +constant folding must not change the result +1 +1 +1 +the optimization still applies to a not Nullable argument +QUERY id: 0 + PROJECTION COLUMNS + uniqExact((number)) UInt64 + PROJECTION + LIST id: 1, nodes: 1 + FUNCTION id: 2, function_name: uniqExact, function_type: aggregate, result_type: UInt64 + ARGUMENTS + LIST id: 3, nodes: 1 + COLUMN id: 4, column_name: number, result_type: UInt64, source_id: 5 + JOIN TREE + TABLE_FUNCTION id: 5, alias: __table1, table_function_name: numbers + ARGUMENTS + LIST id: 6, nodes: 1 + CONSTANT id: 7, constant_value: UInt64_3, constant_value_type: UInt8 + SETTINGS optimize_injective_functions_inside_uniq=1 diff --git a/tests/queries/0_stateless/04927_uniq_injective_tuple_nullable.sql b/tests/queries/0_stateless/04927_uniq_injective_tuple_nullable.sql new file mode 100644 index 000000000000..8451cdd8357c --- /dev/null +++ b/tests/queries/0_stateless/04927_uniq_injective_tuple_nullable.sql @@ -0,0 +1,38 @@ +-- Tags: no-old-analyzer +-- The old analyzer rewrites the query before types are resolved and keeps the wrong results. +-- https://github.com/ClickHouse/ClickHouse/issues/114784 +-- `uniq*` skips the rows where an argument is NULL. An injective function whose result cannot be Nullable +-- hides that nullability - `tuple(NULL)` and `bitmaskToArray(NULL)` are values that get counted - so +-- `optimize_injective_functions_inside_uniq` must not remove it. + +SELECT 'tuple'; +SELECT uniq(tuple(x)), uniqExact(tuple(x)), uniqHLL12(tuple(x)), uniqCombined(tuple(x)), uniqCombined64(tuple(x)) +FROM values('x Nullable(Int32)', 1, 2, NULL) +SETTINGS optimize_injective_functions_inside_uniq = 1; + +SELECT uniq(tuple(x)), uniqExact(tuple(x)), uniqHLL12(tuple(x)), uniqCombined(tuple(x)), uniqCombined64(tuple(x)) +FROM values('x Nullable(Int32)', 1, 2, NULL) +SETTINGS optimize_injective_functions_inside_uniq = 0; + +SELECT 'nested, LowCardinality and multiple arguments'; +SELECT uniqExact(tuple(tuple(x))), uniqExact(tuple(toLowCardinality(x))) +FROM values('x Nullable(Int32)', 1, 2, NULL) +SETTINGS optimize_injective_functions_inside_uniq = 1; + +SELECT uniqExact(tuple(tuple(x))), uniqExact(tuple(toLowCardinality(x))) +FROM values('x Nullable(Int32)', 1, 2, NULL) +SETTINGS optimize_injective_functions_inside_uniq = 0; + +SELECT uniqExact(tuple(x), y) FROM values('x Nullable(Int32), y Nullable(Int32)', (1, 1), (2, NULL), (NULL, 3)) +SETTINGS optimize_injective_functions_inside_uniq = 1; + +SELECT uniqExact(tuple(x), y) FROM values('x Nullable(Int32), y Nullable(Int32)', (1, 1), (2, NULL), (NULL, 3)) +SETTINGS optimize_injective_functions_inside_uniq = 0; + +SELECT 'constant folding must not change the result'; +SELECT countDistinct(tuple(NULL)); +SELECT countDistinct(tuple(arrayJoin([NULL]))); +SELECT countDistinct(tuple(arrayJoin(emptyArrayToSingle([]::Array(Nullable(Int32)))))); + +SELECT 'the optimization still applies to a not Nullable argument'; +EXPLAIN QUERY TREE SELECT uniqExact(tuple(number)) FROM numbers(3) SETTINGS optimize_injective_functions_inside_uniq = 1; diff --git a/tests/queries/0_stateless/05024_binary_string_literal_uninitialized_byte.reference b/tests/queries/0_stateless/05024_binary_string_literal_uninitialized_byte.reference new file mode 100644 index 000000000000..5388d4505bca --- /dev/null +++ b/tests/queries/0_stateless/05024_binary_string_literal_uninitialized_byte.reference @@ -0,0 +1,6 @@ +01 +1 +01 +0101 +01 +0001 diff --git a/tests/queries/0_stateless/05024_binary_string_literal_uninitialized_byte.sql b/tests/queries/0_stateless/05024_binary_string_literal_uninitialized_byte.sql new file mode 100644 index 000000000000..7fe76d3ef33f --- /dev/null +++ b/tests/queries/0_stateless/05024_binary_string_literal_uninitialized_byte.sql @@ -0,0 +1,10 @@ +-- The buffer of a binary string literal is allocated for the worst case: ceil(bits / 8) bytes. +-- When the number of bits is not a multiple of eight, fewer bytes are written, and the rest of +-- the buffer must not end up in the result. + +SELECT hex(b'000000001'); +SELECT length(b'000000001'); +SELECT hex(b'1'); +SELECT hex(b'100000001'); +SELECT hex(b'00000001'); +SELECT hex(x'0001'); diff --git a/tests/queries/0_stateless/05028_lz4_empty_compressed_body.reference b/tests/queries/0_stateless/05028_lz4_empty_compressed_body.reference new file mode 100644 index 000000000000..c34302cbc7ef --- /dev/null +++ b/tests/queries/0_stateless/05028_lz4_empty_compressed_body.reference @@ -0,0 +1,2 @@ +Cannot decompress LZ4-encoded data +Ok. diff --git a/tests/queries/0_stateless/05028_lz4_empty_compressed_body.sh b/tests/queries/0_stateless/05028_lz4_empty_compressed_body.sh new file mode 100755 index 000000000000..20ef52e589d5 --- /dev/null +++ b/tests/queries/0_stateless/05028_lz4_empty_compressed_body.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +# A compressed block with an empty body: the compressed size is exactly the size of the header, while +# the uncompressed size is not zero. The decompressor has nothing to read from, so it must fail +# instead of reporting success and handing out the previous contents of the destination buffer. +# +# The layout of the block is: 16 bytes of the checksum (not verified here), 1 byte of the method +# (0x82 is LZ4), 4 bytes of the compressed size including the 9 bytes of the header, and 4 bytes of +# the uncompressed size. The numbers are little endian. + +echo -ne '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x82\x09\x00\x00\x00\x10\x00\x00\x00' | + ${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}&decompress=1&http_native_compression_disable_checksumming_on_decompress=1" --data-binary @- 2>&1 | + grep -oF 'Cannot decompress LZ4-encoded data' + +${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}" --data-binary "SELECT 'Ok.'" diff --git a/tests/queries/0_stateless/data_minio/03800_a.parquet b/tests/queries/0_stateless/data_minio/03800_a.parquet new file mode 100644 index 000000000000..c1c7f0ae7395 Binary files /dev/null and b/tests/queries/0_stateless/data_minio/03800_a.parquet differ diff --git a/tests/queries/0_stateless/data_minio/03800_b.parquet b/tests/queries/0_stateless/data_minio/03800_b.parquet new file mode 100644 index 000000000000..c1c7f0ae7395 Binary files /dev/null and b/tests/queries/0_stateless/data_minio/03800_b.parquet differ