Add export checksums (#12728)

Author: pixcc

ydb/core/protos/feature_flags.proto

@@ -184,4 +184,5 @@ message TFeatureFlags {
     optional bool EnableDataShardInMemoryStateMigration = 159 [default = true];
     optional bool EnableDataShardInMemoryStateMigrationAcrossGenerations = 160 [default = false];
     optional bool DisableLocalDBEraseCache = 161 [default = false];
+    optional bool EnableExportChecksums = 162 [default = false];
 }

ydb/core/protos/flat_scheme_op.proto

@@ -1246,6 +1246,7 @@ message TBackupTask {
 
     optional uint64 SnapshotStep = 14;
     optional uint64 SnapshotTxId = 15;
+    optional bool EnableChecksums = 16; // currently available for s3
 }
 
 message TRestoreTask {

ydb/core/testlib/basics/feature_flags.h

@@ -71,6 +71,7 @@ class TTestFeatureFlagsHolder {
     FEATURE_FLAG_SETTER(EnableParameterizedDecimal)
     FEATURE_FLAG_SETTER(EnableTopicAutopartitioningForCDC)
     FEATURE_FLAG_SETTER(EnableFollowerStats)
+    FEATURE_FLAG_SETTER(EnableExportChecksums)
 
     #undef FEATURE_FLAG_SETTER
 };

ydb/core/tx/datashard/backup_restore_common.cpp

@@ -6,9 +6,13 @@ void TMetadata::AddFullBackup(TFullBackupMetadata::TPtr fb) {
     FullBackups.emplace(fb->SnapshotVts, fb);
 }
 
+void TMetadata::SetVersion(ui64 version) {
+    Version = version;
+}
+
 TString TMetadata::Serialize() const {
     NJson::TJsonMap m;
-    m["version"] = 0;
+    m["version"] = Version;
     NJson::TJsonArray fullBackups;
     for (auto &[tp, _] : FullBackups) {
         NJson::TJsonMap backupMap;

ydb/core/tx/datashard/backup_restore_common.h

@@ -175,13 +175,15 @@ class TMetadata {
     void AddFullBackup(TFullBackupMetadata::TPtr fullBackup);
     void AddLog(TLogMetadata::TPtr log);
     void SetConsistencyKey(const TString& key);
+    void SetVersion(ui64 version);
 
     TString Serialize() const;
     static TMetadata Deserialize(const TString& metadata);
 private:
     TString ConsistencyKey;
     TMap<TVirtualTimestamp, TFullBackupMetadata::TPtr> FullBackups;
     TMap<TVirtualTimestamp, TLogMetadata::TPtr> Logs;
+    ui64 Version = 0;
 };
 
 } // NBackupRestore

ydb/core/tx/datashard/backup_restore_traits.cpp

@@ -72,6 +72,27 @@ TString DataFileExtension(EDataFormat format, ECompressionCodec codec) {
     return Sprintf("%s%s", fit->second.c_str(), cit->second.c_str());
 }
 
+TString PermissionsKeySuffix() {
+    return "permissions.pb";
+}
+
+TString SchemeKeySuffix() {
+    return "scheme.pb";
+}
+
+TString MetadataKeySuffix() {
+    return "metadata.json";
+}
+
+TString DataKeySuffix(ui32 n, EDataFormat format, ECompressionCodec codec) {
+    const auto ext = DataFileExtension(format, codec);
+    return Sprintf("data_%02d%s", n, ext.c_str());
+}
+
+TString ChecksumKey(const TString& objKey) {
+    return objKey + ".sha256";
+}
+
 } // NBackupRestoreTraits
 } // NDataShard
 } // NKikimr

ydb/core/tx/datashard/backup_restore_traits.h

@@ -30,22 +30,12 @@ ECompressionCodec NextCompressionCodec(ECompressionCodec cur);
 
 TString DataFileExtension(EDataFormat format, ECompressionCodec codec);
 
-inline TString SchemeKey(const TString& objKeyPattern) {
-    return Sprintf("%s/scheme.pb", objKeyPattern.c_str());
-}
-
-inline TString PermissionsKey(const TString& objKeyPattern) {
-    return Sprintf("%s/permissions.pb", objKeyPattern.c_str());
-}
+TString PermissionsKeySuffix();
+TString SchemeKeySuffix();
+TString MetadataKeySuffix();
+TString DataKeySuffix(ui32 n, EDataFormat format, ECompressionCodec codec);
 
-inline TString MetadataKey(const TString& objKeyPattern) {
-    return Sprintf("%s/metadata.json", objKeyPattern.c_str());
-}
-
-inline TString DataKey(const TString& objKeyPattern, ui32 n, EDataFormat format, ECompressionCodec codec) {
-    const auto ext = DataFileExtension(format, codec);
-    return Sprintf("%s/data_%02d%s", objKeyPattern.c_str(), n, ext.c_str());
-}
+TString ChecksumKey(const TString& objKey);
 
 } // NBackupRestoreTraits
 } // NDataShard

ydb/core/tx/datashard/export_checksum.cpp

@@ -0,0 +1,39 @@
+#include "export_checksum.h"
+
+#include <openssl/sha.h>
+
+#include <util/string/hex.h>
+
+namespace NKikimr::NDataShard {
+
+class TSHA256 : public IExportChecksum {
+public:
+    TSHA256() {
+        SHA256_Init(&Context);
+    }
+
+    void AddData(TStringBuf data) override {
+        SHA256_Update(&Context, data.data(), data.size());
+    }
+
+    TString Serialize() override {
+        unsigned char hash[SHA256_DIGEST_LENGTH];
+        SHA256_Final(hash, &Context);
+        return to_lower(HexEncode(hash, SHA256_DIGEST_LENGTH));
+    }
+
+private:
+    SHA256_CTX Context;
+};
+
+TString ComputeExportChecksum(TStringBuf data) {
+    IExportChecksum::TPtr checksum(CreateExportChecksum());
+    checksum->AddData(data);
+    return checksum->Serialize();
+}
+
+IExportChecksum* CreateExportChecksum() {
+    return new TSHA256();
+}
+
+} // NKikimr::NDataShard

ydb/core/tx/datashard/export_checksum.h

@@ -0,0 +1,20 @@
+#pragma once
+
+#include <util/generic/string.h>
+
+namespace NKikimr::NDataShard {
+
+class IExportChecksum {
+public:
+    using TPtr = std::unique_ptr<IExportChecksum>;
+
+    virtual ~IExportChecksum() = default;
+
+    virtual void AddData(TStringBuf data) = 0;
+    virtual TString Serialize() = 0;
+};
+
+IExportChecksum* CreateExportChecksum();
+TString ComputeExportChecksum(TStringBuf data);
+
+} // NKikimr::NDataShard

ydb/core/tx/datashard/export_s3.h

@@ -31,9 +31,10 @@ class TS3Export: public IExport {
 
         switch (CodecFromTask(Task)) {
         case ECompressionCodec::None:
-            return CreateS3ExportBufferRaw(Columns, maxRows, maxBytes);
+            return CreateS3ExportBufferRaw(Columns, maxRows, maxBytes, Task.GetEnableChecksums());
         case ECompressionCodec::Zstd:
-            return CreateS3ExportBufferZstd(Task.GetCompression().GetLevel(), Columns, maxRows, maxBytes, minBytes);
+            return CreateS3ExportBufferZstd(Task.GetCompression().GetLevel(), Columns, maxRows,
+                maxBytes, minBytes, Task.GetEnableChecksums());
         case ECompressionCodec::Invalid:
             Y_ABORT("unreachable");
         }