diff --git a/docs/en/antalya/cas/index.md b/docs/en/antalya/cas/index.md index cb1d563eebf0..377f03b7d40f 100644 --- a/docs/en/antalya/cas/index.md +++ b/docs/en/antalya/cas/index.md @@ -94,4 +94,5 @@ per disk, so adopting it never requires migrating an existing deployment. | [Architecture overview](/antalya/cas/architecture/) | The object model, the Git analogy, and the safety invariants | | [Correctness](/antalya/cas/architecture/correctness) | How the design was verified: TLA+ models, counterexamples, soak methodology | | [Design history](/antalya/cas/architecture/design-history) | What earlier designs were tried and rejected, and why | +| [Backup](/antalya/cas/operations/backup) | How `BACKUP` and `RESTORE` behave on a content-addressed disk | | [Roadmap](/antalya/cas/roadmap) | What is shipped, planned, and deliberately not pursued | diff --git a/docs/en/antalya/cas/operations/backup.md b/docs/en/antalya/cas/operations/backup.md new file mode 100644 index 000000000000..25c019dce0c9 --- /dev/null +++ b/docs/en/antalya/cas/operations/backup.md @@ -0,0 +1,99 @@ +--- +description: 'How BACKUP and RESTORE work for a table on a content-addressed disk: what holds the data during a backup, when the copy runs inside S3, and what is not supported yet.' +sidebar_label: 'Backup' +sidebar_position: 5 +slug: /antalya/cas/operations/backup +title: 'CAS Operations — Backup' +doc_type: 'guide' +--- + +# Operations — backup {#backup} + +Ordinary `BACKUP` and `RESTORE` work for a table on a content-addressed (`CAS`) disk. This page +covers what happens during one, how it differs from a plain disk, and what the limits are. + +The `CAS`-native backup model — `snapshot` / `mirror` / `fetch` / `restore` — is designed but not +implemented and is not wired into the SQL surface. See the [roadmap](/antalya/cas/roadmap#backups). + +## What is supported {#supported} + +```sql +BACKUP TABLE t TO S3('https://bucket.s3.amazonaws.com/backups/b1', 'key', 'secret'); +RESTORE TABLE t AS t_restored FROM S3('https://bucket.s3.amazonaws.com/backups/b1', 'key', 'secret'); +``` + +The destination can be anything: `S3`, `Disk`, `File`, or an archive. A backup can be restored onto a +disk of any type, because it holds the table's files rather than the pool's objects. + +**An `Atomic` database is required.** That has been the default since 20.x. On the deprecated +`Ordinary` engine the backup fails with `SUPPORT_IS_DISABLED`: that path pins files with temporary +hard links, which object storage does not have. + +## What holds the data during a backup {#holding} + +On a plain disk a backup pins files against deletion with a hard link. `CAS` uses a different +mechanism — pointer holding: + +- the backup holds a `shared_ptr` to the table and to each part; +- the outdated-part cleanup skips those parts; +- while a part is alive so is its [ref](/antalya/cas/architecture/manifests-and-refs#ref-table) — the + name under which the part is registered in its namespace's ref table, and through which it points + at its manifest; +- while the ref is alive, garbage collection sees the manifest and its blobs as reachable. + +This mechanism lives in the process's memory and **does not survive a server restart**. An +interrupted backup leaves nothing behind in the pool, but it also stops protecting the data once the +process is gone. For durable pinning there is `FREEZE`, which publishes a real ref. + +## How the bytes move {#copy-path} + +Which path runs depends on the destination. + +**A destination outside the pool** — the common case: another bucket, a local disk, an archive. Files +are read through the `CAS` read path and written to the destination. Pool deduplication is lost: +what was one blob shared by several replicas becomes ordinary files in the backup. + +**A destination on the same `S3` endpoint as the pool** — the copy then runs inside the `S3` store +itself: the ClickHouse server issues one "copy these bytes" command, and `S3` moves the bytes +internally without sending them through ClickHouse. The files of a part fall into two categories: + +| Category | Example | How it is copied | +|---|---|---| +| Blob | `data.bin`, marks, `primary.idx` | an `UploadPartCopy` naming a byte range — only the payload moves, without the blob's internal header | +| Inside the manifest | `checksums.txt`, `count.txt`, `columns.txt` | through ClickHouse's buffers: they have no object of their own | + +If the destination cannot copy a byte range, `CAS` does not fall back to copying the whole object — +the file is read and written through ClickHouse instead. That is slower, but correct. + +Copying inside `S3` can be turned off: + +```sql +BACKUP TABLE t TO S3(...) SETTINGS allow_s3_native_copy = 0; +``` + +## Restore {#restore} + +Each part is materialized in **one disk transaction** and published as one manifest and one ref. A +partially restored part can never appear in the pool: either the whole part is published or nothing +is. + +Restored data is packed afresh — on a `CAS` disk it gets new blobs and new refs. Deduplication +against data already in the pool works as usual: identical content hashes to the same blob and is +not written twice. + +## `FREEZE` is not a backup {#freeze} + +`ALTER TABLE ... FREEZE` works on `CAS` and publishes parts into a separate shadow namespace, which +is a garbage-collection root in its own right. `DROP PARTITION` removes the live refs and leaves the +snapshot alone; `SYSTEM UNFREEZE` removes only the shadow refs. + +It is still not a snapshot of a table: there is no SQL metadata, no single commit marker, no +portable object with a listing and a restore API, and its lifetime is tied to a manual `UNFREEZE`. +It is a useful building block, not a replacement for `BACKUP`. + +## Limitations {#limitations} + +- The `CAS`-native backup model (`snapshot` / `mirror` / `fetch`) is not implemented. +- The `Ordinary` database engine is not supported. +- Pool deduplication is lost in the backup: its size follows the logical files, not the unique blobs. +- Pointer holding does not survive a server restart. diff --git a/docs/en/antalya/cas/roadmap.md b/docs/en/antalya/cas/roadmap.md index 4df03af9ab3c..14578bd42058 100644 --- a/docs/en/antalya/cas/roadmap.md +++ b/docs/en/antalya/cas/roadmap.md @@ -91,6 +91,10 @@ positioning. ## Backups {#backups} +Ordinary `BACKUP` and `RESTORE` already work for a table on a `CAS` disk — see +[backup](/antalya/cas/operations/backup) for how they behave and what the limits are. What follows is +about the `CAS`-native model, which is a different thing. + A `snapshot` / `mirror` / `fetch` / `restore` design is **approved but not implemented**. The model is deliberately git-shaped: `snapshot` is instant and free (like `git tag` — it references existing manifests, copies nothing); `mirror` is a continuous pull from a production pool into a diff --git a/src/Backups/BackupIO_S3.cpp b/src/Backups/BackupIO_S3.cpp index df43076d8957..2a92e5e862a4 100644 --- a/src/Backups/BackupIO_S3.cpp +++ b/src/Backups/BackupIO_S3.cpp @@ -317,6 +317,7 @@ void BackupReaderS3::copyFileToDisk(const String & path_in_backup, size_t file_s fs::path(s3_uri.key) / path_in_backup, 0, file_size, + /* src_object_offset= */ 0, /* dest_s3_client= */ destination_disk->getS3StorageClient(), /* dest_bucket= */ blob_path[1], /* dest_key= */ blob_path[0], @@ -392,12 +393,23 @@ void BackupWriterS3::copyFileFromDisk( if (auto blob_path = src_disk->getBlobPath(src_path); blob_path.size() == 2) { LOG_TRACE(log, "Copying file {} from disk {} to S3", src_path, src_disk->getName()); + + if (blob_path[0].empty()) + { + LOG_TRACE(log, "File {} has no object of its own, copying through buffers", src_path); + BackupWriterDefault::copyFileFromDisk(path_in_backup, src_disk, src_path, copy_encrypted, start_pos, length); + return; + } + + const size_t src_object_offset = src_disk->getObjectPayloadOffset(src_path); + copyS3File( /* src_s3_client */ disk_client_factory.getOrCreate(src_disk), /* src_bucket */ blob_path[1], /* src_key */ blob_path[0], start_pos, length, + src_object_offset, /* dest_s3_client */ client, /* dest_bucket */ s3_uri.bucket, /* dest_key */ fs::path(s3_uri.key) / path_in_backup, @@ -433,6 +445,7 @@ void BackupWriterS3::copyFile(const String & destination, const String & source, /* src_key= */ source_key, 0, size, + /* src_object_offset= */ 0, /* dest_s3_client= */ client, /* dest_bucket= */ s3_uri.bucket, /* dest_key= */ fs::path(s3_uri.key) / destination, diff --git a/src/Disks/DiskBackup.cpp b/src/Disks/DiskBackup.cpp index 7f5b1ec9b0d4..93efc93f0e60 100644 --- a/src/Disks/DiskBackup.cpp +++ b/src/Disks/DiskBackup.cpp @@ -170,6 +170,11 @@ std::vector DiskBackup::getBlobPath(const String &) const throw Exception(ErrorCodes::UNSUPPORTED_METHOD, "DiskBackup does not support getBlobPath method"); } +size_t DiskBackup::getObjectPayloadOffset(const String &) const +{ + throw Exception(ErrorCodes::UNSUPPORTED_METHOD, "DiskBackup does not support getObjectPayloadOffset method"); +} + void DiskBackup::writeFileUsingBlobWritingFunction(const String &, WriteMode, WriteBlobFunction &&) { throw Exception(ErrorCodes::UNSUPPORTED_METHOD, "DiskBackup does not support writeFileUsingBlobWritingFunction method"); diff --git a/src/Disks/DiskBackup.h b/src/Disks/DiskBackup.h index b2f67e17ab20..5468a7bac7aa 100644 --- a/src/Disks/DiskBackup.h +++ b/src/Disks/DiskBackup.h @@ -91,6 +91,7 @@ class DiskBackup final : public IDisk const WriteSettings & settings) override; Strings getBlobPath(const String & path) const override; + size_t getObjectPayloadOffset(const String & path) const override; bool areBlobPathsRandom() const override { return false; } void writeFileUsingBlobWritingFunction(const String & path, WriteMode mode, WriteBlobFunction && write_blob_function) override; diff --git a/src/Disks/DiskEncrypted.h b/src/Disks/DiskEncrypted.h index ece91f310985..9640c2e5740c 100644 --- a/src/Disks/DiskEncrypted.h +++ b/src/Disks/DiskEncrypted.h @@ -205,6 +205,12 @@ class DiskEncrypted : public IDisk return delegate->getBlobPath(wrapped_path); } + size_t getObjectPayloadOffset(const String & path) const override + { + auto wrapped_path = wrappedPath(path); + return delegate->getObjectPayloadOffset(wrapped_path); + } + bool areBlobPathsRandom() const override { return delegate->areBlobPathsRandom(); diff --git a/src/Disks/DiskLocal.cpp b/src/Disks/DiskLocal.cpp index 472944b64432..9eb3ef2ba3a0 100644 --- a/src/Disks/DiskLocal.cpp +++ b/src/Disks/DiskLocal.cpp @@ -447,6 +447,11 @@ std::vector DiskLocal::getBlobPath(const String & path) const return {fs_path}; } +size_t DiskLocal::getObjectPayloadOffset(const String &) const +{ + return 0; +} + void DiskLocal::writeFileUsingBlobWritingFunction(const String & path, WriteMode mode, WriteBlobFunction && write_blob_function) { auto fs_path = fs::path(disk_path) / path; diff --git a/src/Disks/DiskLocal.h b/src/Disks/DiskLocal.h index a262752e4e12..83b43687f17b 100644 --- a/src/Disks/DiskLocal.h +++ b/src/Disks/DiskLocal.h @@ -91,6 +91,7 @@ class DiskLocal : public IDisk const WriteSettings & settings) override; Strings getBlobPath(const String & path) const override; + size_t getObjectPayloadOffset(const String & path) const override; bool areBlobPathsRandom() const override { return false; } void writeFileUsingBlobWritingFunction(const String & path, WriteMode mode, WriteBlobFunction && write_blob_function) override; diff --git a/src/Disks/DiskObjectStorage/DiskObjectStorage.cpp b/src/Disks/DiskObjectStorage/DiskObjectStorage.cpp index 2c2d071a7fbb..fef09747814e 100644 --- a/src/Disks/DiskObjectStorage/DiskObjectStorage.cpp +++ b/src/Disks/DiskObjectStorage/DiskObjectStorage.cpp @@ -54,6 +54,7 @@ namespace ErrorCodes { extern const int INCORRECT_DISK_INDEX; extern const int CANNOT_RMDIR; + extern const int LOGICAL_ERROR; } namespace @@ -824,12 +825,16 @@ void DiskObjectStorage::prepareRead( if (metadata_storage->isContentAddressed()) { const auto * ca = dynamic_cast(metadata_storage.get()); - if (ca) - { - if (ca->prepareInManifestRead(path, settings, pipeline)) - return; - ca_blob_view = ca->getBlobViewPlan(path); - } + if (!ca) + throw Exception( + ErrorCodes::LOGICAL_ERROR, + "Metadata storage of disk {} reports itself content-addressed but does not implement " + "IContentAddressedExchange, so the payload window of {} cannot be resolved", + getName(), path); + + if (ca->prepareInManifestRead(path, settings, pipeline)) + return; + ca_blob_view = ca->getBlobViewPlan(path); } const auto storage_objects = ca_blob_view @@ -957,6 +962,24 @@ Strings DiskObjectStorage::getBlobPath(const String & path) const return res; } +size_t DiskObjectStorage::getObjectPayloadOffset(const String & path) const +{ + if (!metadata_storage->isContentAddressed()) + return 0; + + const auto * ca = dynamic_cast(metadata_storage.get()); + if (!ca) + throw Exception( + ErrorCodes::LOGICAL_ERROR, + "Metadata storage of disk {} reports itself content-addressed but does not implement " + "IContentAddressedExchange, so the payload offset of {} cannot be resolved", + getName(), path); + + if (auto plan = ca->getBlobViewPlan(path)) + return plan->payload_offset; + return 0; +} + bool DiskObjectStorage::areBlobPathsRandom() const { return metadata_storage->areBlobPathsRandom(); diff --git a/src/Disks/DiskObjectStorage/DiskObjectStorage.h b/src/Disks/DiskObjectStorage/DiskObjectStorage.h index 09ec478b02d0..95bf6ef366d7 100644 --- a/src/Disks/DiskObjectStorage/DiskObjectStorage.h +++ b/src/Disks/DiskObjectStorage/DiskObjectStorage.h @@ -179,6 +179,8 @@ friend class DiskObjectStorageReservation; const WriteSettings & settings) override; Strings getBlobPath(const String & path) const override; + + size_t getObjectPayloadOffset(const String & path) const override; bool areBlobPathsRandom() const override; void writeFileUsingBlobWritingFunction(const String & path, WriteMode mode, WriteBlobFunction && write_blob_function) override; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp index 8fb5a31b467c..0e88ad24e93e 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp @@ -2103,10 +2103,10 @@ std::optional ContentAddressedMet /// `partAccess()` then `store()` pair (each an independent `pointer_mutex` acquisition) -- see /// `poolAccess()`. const auto snap = poolAccess(); - auto view = snap.part_access->getView(r->refKey(), Cas::Freshness::CachedForLoad); - if (!view) + const auto manifest_view = snap.part_access->getView(r->refKey(), Cas::Freshness::CachedForLoad); + if (!manifest_view) return std::nullopt; - if (const auto * entry = view->findFile(r->file)) + if (const auto * entry = manifest_view->findFile(r->file)) { const auto location = snap.pool->locate(*entry); BlobViewPlan plan; diff --git a/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp b/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp index 6eb155dd4c44..2f74ad9aa8f8 100644 --- a/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp +++ b/src/Disks/DiskObjectStorage/ObjectStorages/S3/S3ObjectStorage.cpp @@ -988,6 +988,7 @@ void S3ObjectStorage::copyObjectToAnotherObjectStorage( // NOLINT /*src_key=*/object_from.remote_path, /*src_offset=*/0, /*src_size=*/size, + /*src_object_offset=*/0, /*dest_s3_client=*/current_client, /*dest_bucket=*/dest_s3->uri.bucket, /*dest_key=*/object_to.remote_path, @@ -1062,6 +1063,7 @@ void S3ObjectStorage::copyObject( // NOLINT /*src_key=*/object_from.remote_path, /*src_offset=*/0, /*src_size=*/size, + /*src_object_offset=*/0, /*dest_s3_client=*/current_client, /*dest_bucket=*/uri.bucket, /*dest_key=*/object_to.remote_path, diff --git a/src/Disks/IDisk.h b/src/Disks/IDisk.h index b79a3171bde7..f1133e0326d5 100644 --- a/src/Disks/IDisk.h +++ b/src/Disks/IDisk.h @@ -318,6 +318,9 @@ class IDisk : public Space /// StoredObject::remote_path for each stored object combined with the name of the objects' namespace. virtual Strings getBlobPath(const String & path) const = 0; + /// Where the file's bytes begin inside the object `getBlobPath` names. + virtual size_t getObjectPayloadOffset(const String & path) const = 0; + /// Returns whether the blob paths this disk uses are randomly generated. virtual bool areBlobPathsRandom() const = 0; diff --git a/src/Disks/ReadOnlyDiskWrapper.h b/src/Disks/ReadOnlyDiskWrapper.h index 9a38e85cde77..f82a562ca6b8 100644 --- a/src/Disks/ReadOnlyDiskWrapper.h +++ b/src/Disks/ReadOnlyDiskWrapper.h @@ -29,6 +29,7 @@ class ReadOnlyDiskWrapper : public IDisk size_t getFileSize(const String & path) const override { return delegate->getFileSize(path); } Strings getBlobPath(const String & path) const override { return delegate->getBlobPath(path); } + size_t getObjectPayloadOffset(const String & path) const override { return delegate->getObjectPayloadOffset(path); } bool areBlobPathsRandom() const override { return delegate->areBlobPathsRandom(); } void writeFileUsingBlobWritingFunction(const String & path, WriteMode mode, WriteBlobFunction && write_blob_function) override { diff --git a/src/IO/S3/copyS3File.cpp b/src/IO/S3/copyS3File.cpp index 4b1f5e14ece5..df5429a2a980 100644 --- a/src/IO/S3/copyS3File.cpp +++ b/src/IO/S3/copyS3File.cpp @@ -640,8 +640,27 @@ namespace void performCopy() { LOG_TEST(log, "Copy object {} to {} using native copy", src_key, dest_key); - bool use_single_operation_copy = !supports_multipart_copy || !request_settings[S3RequestSetting::allow_multipart_copy] - || (size <= request_settings[S3RequestSetting::max_single_operation_copy_size]); + + const bool ranged = offset != 0; + const bool multipart_copy_available + = supports_multipart_copy && request_settings[S3RequestSetting::allow_multipart_copy]; + + if (ranged && !multipart_copy_available) + { + if (!allow_fallback) + throw Exception( + ErrorCodes::NOT_IMPLEMENTED, + "Native copy of a byte range requires multipart copy, which is unavailable for {}", + src_key); + + LOG_TRACE(log, "Ranged native copy needs multipart copy, falling back for {}", src_key); + fallback_method(); + return; + } + + const bool use_single_operation_copy = !ranged + && (!multipart_copy_available + || (size <= request_settings[S3RequestSetting::max_single_operation_copy_size])); if (use_single_operation_copy) performSingleOperationCopy(); @@ -863,6 +882,7 @@ void copyS3File( const String & src_key, size_t src_offset, size_t src_size, + size_t src_object_offset, std::shared_ptr dest_s3_client, const String & dest_bucket, const String & dest_key, @@ -908,7 +928,7 @@ void copyS3File( src_s3_client, src_bucket, src_key, - src_offset, + src_offset + src_object_offset, src_size, dest_bucket, dest_key, diff --git a/src/IO/S3/copyS3File.h b/src/IO/S3/copyS3File.h index d4f728377130..8c37c4187739 100644 --- a/src/IO/S3/copyS3File.h +++ b/src/IO/S3/copyS3File.h @@ -40,6 +40,7 @@ void copyS3File( const String & src_key, size_t src_offset, size_t src_size, + size_t src_object_offset, std::shared_ptr dest_s3_client, const String & dest_bucket, const String & dest_key, diff --git a/src/Storages/ObjectStorageQueue/ObjectStorageQueuePostProcessor.cpp b/src/Storages/ObjectStorageQueue/ObjectStorageQueuePostProcessor.cpp index aa748eacaedc..020904ae73c3 100644 --- a/src/Storages/ObjectStorageQueue/ObjectStorageQueuePostProcessor.cpp +++ b/src/Storages/ObjectStorageQueue/ObjectStorageQueuePostProcessor.cpp @@ -331,6 +331,7 @@ void ObjectStorageQueuePostProcessor::moveS3Objects(const StoredObjects & object /*src_key=*/ object_from.remote_path, /*src_offset=*/ 0, /*src_size=*/ object_size, + /*src_object_offset=*/ 0, /*dest_s3_client=*/ dst_client, /*dest_bucket=*/ dst_uri.bucket, /*dest_key=*/ object_to.remote_path, diff --git a/tests/integration/test_cas_backup_s3_native_copy/__init__.py b/tests/integration/test_cas_backup_s3_native_copy/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/integration/test_cas_backup_s3_native_copy/configs/storage_conf.xml b/tests/integration/test_cas_backup_s3_native_copy/configs/storage_conf.xml new file mode 100644 index 000000000000..f50227cedd3c --- /dev/null +++ b/tests/integration/test_cas_backup_s3_native_copy/configs/storage_conf.xml @@ -0,0 +1,30 @@ + + + + + object_storage + s3 + cas + 30 + 10000 + itest-cas-backup-s3-native-copy + + http://rustfs1:11121/test/cas_backup_data/ + clickhouse + clickhouse + + + + + +
+ disk_cas_backup_s3 +
+ + + + + diff --git a/tests/integration/test_cas_backup_s3_native_copy/test.py b/tests/integration/test_cas_backup_s3_native_copy/test.py new file mode 100644 index 000000000000..96900347c830 --- /dev/null +++ b/tests/integration/test_cas_backup_s3_native_copy/test.py @@ -0,0 +1,211 @@ +"""BACKUP/RESTORE of a CAS table to an S3 destination sharing the pool's authority. + +`sameKind` then matches and `BackupWriterS3` copies objects server-side instead of reading +through the CAS read path. That path must handle two shapes: a blob, whose object is +`[envelope][payload]` so the file starts at a non-zero offset, and an inline manifest entry, +which has no object at all. `getStorageObjects` reports neither -- it drops the offset and +returns an empty key. + +Columns are chosen so one table yields both shapes: placement keys on the file name, so every +column makes blobs while per-part metadata stays inline. `n` and `arr` add the `.null.bin` and +`.size0.bin` substreams. +""" + +import uuid + +import pytest + +from helpers.cluster import ClickHouseCluster + +cluster = ClickHouseCluster(__file__) + +STORAGE_POLICY = "cas_backup_s3" + +S3_AUTHORITY = "http://rustfs1:11121" +S3_CREDENTIALS = "'clickhouse', 'clickhouse'" + +NUM_ROWS = 100000 + +COLUMNS = ["k", "s", "n", "arr"] + +RUN_TOKEN = uuid.uuid4().hex + + +@pytest.fixture(scope="module", autouse=True) +def start_cluster(): + cluster.add_instance( + "node", + main_configs=["configs/storage_conf.xml"], + with_rustfs=True, + stay_alive=True, + ) + try: + cluster.start() + yield + finally: + cluster.shutdown() + + +def backup_destination(name): + return f"S3('{S3_AUTHORITY}/test/backups/{RUN_TOKEN}/{name}', {S3_CREDENTIALS})" + + +def create_and_fill(node, table): + node.query(f"DROP TABLE IF EXISTS {table} SYNC") + node.query( + f""" + CREATE TABLE {table} (k UInt64, s String, n Nullable(Int64), arr Array(UInt32)) + ENGINE = MergeTree ORDER BY k + SETTINGS storage_policy = '{STORAGE_POLICY}', min_bytes_for_wide_part = 0 + """ + ) + node.query( + f""" + INSERT INTO {table} + SELECT + number, + randomPrintableASCII(64), + if(number % 7 = 0, NULL, toInt64(number)), + [toUInt32(number), toUInt32(number + 1)] + FROM numbers({NUM_ROWS}) + """ + ) + + +def column_fingerprints(node, table): + """Order-independent per-column hash; reading every column fetches every blob.""" + exprs = ", ".join( + f"sum(cityHash64(ifNull(toString({column}), '')))" for column in COLUMNS + ) + row = node.query(f"SELECT count(), {exprs} FROM {table}").strip().split("\t") + return dict(zip(["count"] + COLUMNS, row)) + + +@pytest.mark.parametrize("allow_native_copy", [True, False]) +def test_native_copy_round_trip(allow_native_copy): + node = cluster.instances["node"] + suffix = "native" if allow_native_copy else "buffered" + table = f"cas_backup_{suffix}" + restored = f"{table}_restored" + destination = backup_destination(suffix) + + create_and_fill(node, table) + expected = column_fingerprints(node, table) + + node.query( + f"BACKUP TABLE {table} TO {destination} " + f"SETTINGS allow_s3_native_copy = {int(allow_native_copy)}" + ) + + node.query(f"DROP TABLE IF EXISTS {restored} SYNC") + node.query( + f"RESTORE TABLE {table} AS {restored} FROM {destination} " + f"SETTINGS allow_s3_native_copy = {int(allow_native_copy)}" + ) + + actual = column_fingerprints(node, restored) + + assert actual["count"] == expected["count"] + differing = [c for c in COLUMNS if actual[c] != expected[c]] + assert not differing, f"columns differ after restore: {differing}" + + assert ( + node.query( + f"CHECK TABLE {restored} SETTINGS check_query_single_value_result = 1" + ).strip() + == "1" + ) + + node.query(f"DROP TABLE {table} SYNC") + node.query(f"DROP TABLE {restored} SYNC") + + +def test_blobs_use_ranged_copy_and_inline_falls_back(): + """A blob is `[envelope][payload]`, so its copy must be ranged: `UploadPartCopy`, never + `CopyObject`. Inline entries have no object and must go through buffers. + """ + node = cluster.instances["node"] + table = "cas_backup_mechanism" + destination = backup_destination("mechanism") + query_id = f"cas_backup_mechanism_{RUN_TOKEN}" + + create_and_fill(node, table) + node.query( + f"BACKUP TABLE {table} TO {destination} SETTINGS allow_s3_native_copy = 1", + query_id=query_id, + ) + node.query("SYSTEM FLUSH LOGS query_log") + + events = node.query( + f""" + SELECT ProfileEvents['S3UploadPartCopy'], ProfileEvents['S3CopyObject'] + FROM system.query_log + WHERE type = 'QueryFinish' AND query_id = '{query_id}' + ORDER BY event_time DESC LIMIT 1 + """ + ).strip() + assert events, "no query_log row for the backup query" + upload_part_copy, copy_object = (int(value) for value in events.split("\t")) + + assert upload_part_copy > 0, "no ranged server-side copy happened" + assert copy_object == 0, "CopyObject has no range: the envelope would land in the backup" + assert node.contains_in_log( + "has no object of its own, copying through buffers" + ), "inline entries did not fall back" + + node.query(f"DROP TABLE {table} SYNC") + + +def test_ranged_copy_falls_back_without_multipart(): + """Only `UploadPartCopy` can express a range. With multipart copy off there is no server-side + operation left, so the copy must go through buffers instead of taking the whole object. + """ + node = cluster.instances["node"] + table = "cas_backup_no_multipart" + restored = f"{table}_restored" + destination = backup_destination("no_multipart") + query_id = f"cas_backup_no_multipart_{RUN_TOKEN}" + no_multipart = {"s3_allow_multipart_copy": 0} + + create_and_fill(node, table) + expected = column_fingerprints(node, table) + + node.query( + f"BACKUP TABLE {table} TO {destination} SETTINGS allow_s3_native_copy = 1", + query_id=query_id, + settings=no_multipart, + ) + node.query("SYSTEM FLUSH LOGS query_log") + + events = node.query( + f""" + SELECT + ProfileEvents['S3UploadPartCopy'], + ProfileEvents['S3CopyObject'], + ProfileEvents['S3PutObject'] + ProfileEvents['S3UploadPart'] + FROM system.query_log + WHERE type = 'QueryFinish' AND query_id = '{query_id}' + ORDER BY event_time DESC LIMIT 1 + """ + ).strip() + assert events, "no query_log row for the backup query" + upload_part_copy, copy_object, uploaded = (int(value) for value in events.split("\t")) + + assert upload_part_copy == 0, "multipart copy was disabled but UploadPartCopy still ran" + assert copy_object == 0, "a ranged copy fell back to CopyObject, which would take the envelope" + assert uploaded > 0, "nothing was uploaded through the server, so nothing was copied at all" + assert node.contains_in_log( + "Ranged native copy needs multipart copy" + ), "the copy did not reach the ranged-copy fallback" + + node.query(f"DROP TABLE IF EXISTS {restored} SYNC") + node.query( + f"RESTORE TABLE {table} AS {restored} FROM {destination}", settings=no_multipart + ) + + actual = column_fingerprints(node, restored) + assert actual["count"] == expected["count"] + assert not [c for c in COLUMNS if actual[c] != expected[c]] + + node.query(f"DROP TABLE {table} SYNC") + node.query(f"DROP TABLE {restored} SYNC")