Don't abort when export corrupted data

azevaykin · azevaykin · commit 7e96bc9836ec · 2023-10-25T17:55:34.000+03:00
diff --git a/ydb/core/scheme/scheme_tablecell.h b/ydb/core/scheme/scheme_tablecell.h
@@ -101,9 +101,19 @@ struct TCell {
         return ReadUnaligned<T>(Data());
     }
 
-    template<typename T, typename = TStdLayout<T>>
-    static inline TCell Make(const T &val) noexcept
-    {
+    template <typename T, typename = TStdLayout<T>>
+    bool ToStream(IOutputStream& out, TStringBuilder& err) const noexcept {
+        if(sizeof(T) != Size()) {
+            err << "AsValue<T>() type doesn't match TCell";
+            return false;
+        }
+
+        out << ReadUnaligned<T>(Data());
+        return true;
+    }
+
+    template <typename T, typename = TStdLayout<T>>
+    static inline TCell Make(const T& val) noexcept {
         auto *ptr = static_cast<const char*>(static_cast<const void*>(&val));
 
         return TCell{ ptr, sizeof(val) };
diff --git a/ydb/core/tx/datashard/export_common.cpp b/ydb/core/tx/datashard/export_common.cpp
@@ -77,9 +77,33 @@ TString DecimalToString(const std::pair<ui64, i64>& loHi) {
 }
 
 TString DyNumberToString(TStringBuf data) {
+    TString result;
+    TStringOutput out(result);
+    TStringBuilder err;
+
+    bool success = DyNumberToStream(data, out, err);
+    Y_ABORT_UNLESS(success);
+
+    return result;
+}
+
+bool DecimalToStream(const std::pair<ui64, i64>& loHi, IOutputStream& out, TStringBuilder& err) {
+    Y_UNUSED(err);
+    using namespace NYql::NDecimal;
+
+    TInt128 val = FromHalfs(loHi.first, loHi.second);
+    out << ToString(val, NScheme::DECIMAL_PRECISION, NScheme::DECIMAL_SCALE);
+    return true;
+}
+
+bool DyNumberToStream(TStringBuf data, IOutputStream& out, TStringBuilder& err) {
     auto result = NDyNumber::DyNumberToString(data);
-    Y_ABORT_UNLESS(result.Defined(), "Invalid DyNumber binary representation");
-    return *result;
+    if (!result.Defined()) {
+        err << "Invalid DyNumber binary representation";
+        return false;
+    }
+    out << *result;
+    return true;
 }
 
 } // NDataShard
diff --git a/ydb/core/tx/datashard/export_common.h b/ydb/core/tx/datashard/export_common.h
@@ -36,6 +36,8 @@ TMaybe<Ydb::Table::CreateTableRequest> GenYdbScheme(
 
 TString DecimalToString(const std::pair<ui64, i64>& loHi);
 TString DyNumberToString(TStringBuf data);
+bool DecimalToStream(const std::pair<ui64, i64>& loHi, IOutputStream& out, TStringBuilder& err);
+bool DyNumberToStream(TStringBuf data, IOutputStream& out, TStringBuilder& err);
 
 } // NDataShard
 } // NKikimr
diff --git a/ydb/core/tx/datashard/export_s3_buffer_raw.cpp b/ydb/core/tx/datashard/export_s3_buffer_raw.cpp
@@ -36,7 +36,7 @@ void TS3BufferRaw::ColumnsOrder(const TVector<ui32>& tags) {
     }
 }
 
-void TS3BufferRaw::Collect(const NTable::IScan::TRow& row, IOutputStream& out) {
+bool TS3BufferRaw::Collect(const NTable::IScan::TRow& row, IOutputStream& out) {
     bool needsComma = false;
     for (const auto& [tag, column] : Columns) {
         auto it = Indices.find(tag);
@@ -57,18 +57,19 @@ void TS3BufferRaw::Collect(const NTable::IScan::TRow& row, IOutputStream& out) {
             continue;
         }
 
+        bool serialized = true;
         switch (column.Type.GetTypeId()) {
         case NScheme::NTypeIds::Int32:
-            out << cell.AsValue<i32>();
+            serialized = cell.ToStream<i32>(out, ErrorString);
             break;
         case NScheme::NTypeIds::Uint32:
-            out << cell.AsValue<ui32>();
+            serialized = cell.ToStream<ui32>(out, ErrorString);
             break;
         case NScheme::NTypeIds::Int64:
-            out << cell.AsValue<i64>();
+            serialized = cell.ToStream<i64>(out, ErrorString);
             break;
         case NScheme::NTypeIds::Uint64:
-            out << cell.AsValue<ui64>();
+            serialized = cell.ToStream<ui64>(out, ErrorString);
             break;
         case NScheme::NTypeIds::Uint8:
         //case NScheme::NTypeIds::Byte:
@@ -78,19 +79,19 @@ void TS3BufferRaw::Collect(const NTable::IScan::TRow& row, IOutputStream& out) {
             out << static_cast<i32>(cell.AsValue<i8>());
             break;
         case NScheme::NTypeIds::Int16:
-            out << cell.AsValue<i16>();
+            serialized = cell.ToStream<i16>(out, ErrorString);
             break;
         case NScheme::NTypeIds::Uint16:
-            out << cell.AsValue<ui16>();
+            serialized = cell.ToStream<ui16>(out, ErrorString);
             break;
         case NScheme::NTypeIds::Bool:
-            out << cell.AsValue<bool>();
+            serialized = cell.ToStream<bool>(out, ErrorString);
             break;
         case NScheme::NTypeIds::Double:
-            out << cell.AsValue<double>();
+            serialized = cell.ToStream<double>(out, ErrorString);
             break;
         case NScheme::NTypeIds::Float:
-            out << cell.AsValue<float>();
+            serialized = cell.ToStream<float>(out, ErrorString);
             break;
         case NScheme::NTypeIds::Date:
             out << TInstant::Days(cell.AsValue<ui16>());
@@ -102,13 +103,13 @@ void TS3BufferRaw::Collect(const NTable::IScan::TRow& row, IOutputStream& out) {
             out << TInstant::MicroSeconds(cell.AsValue<ui64>());
             break;
         case NScheme::NTypeIds::Interval:
-            out << cell.AsValue<i64>();
+            serialized = cell.ToStream<i64>(out, ErrorString);
             break;
         case NScheme::NTypeIds::Decimal:
-            out << DecimalToString(cell.AsValue<std::pair<ui64, i64>>());
+            serialized = DecimalToStream(cell.AsValue<std::pair<ui64, i64>>(), out, ErrorString);
             break;
         case NScheme::NTypeIds::DyNumber:
-            out << DyNumberToString(cell.AsBuf());
+            serialized = DyNumberToStream(cell.AsBuf(), out, ErrorString);
             break;
         case NScheme::NTypeIds::String:
         case NScheme::NTypeIds::String4k:
@@ -127,16 +128,22 @@ void TS3BufferRaw::Collect(const NTable::IScan::TRow& row, IOutputStream& out) {
         default:
             Y_ABORT("Unsupported type");
         }
+
+        if (!serialized) {
+            return false;
+        }
     }
 
     out << "\n";
     ++Rows;
+
+    return true;
 }
 
 bool TS3BufferRaw::Collect(const NTable::IScan::TRow& row) {
     TBufferOutput out(Buffer);
-    Collect(row, out);
-    return true;
+    ErrorString.clear();
+    return Collect(row, out);
 }
 
 IEventBase* TS3BufferRaw::PrepareEvent(bool last, NExportScan::IBuffer::TStats& stats) {
@@ -161,7 +168,7 @@ bool TS3BufferRaw::IsFilled() const {
 }
 
 TString TS3BufferRaw::GetError() const {
-    Y_ABORT("unreachable");
+    return ErrorString;
 }
 
 TMaybe<TBuffer> TS3BufferRaw::Flush(bool) {
diff --git a/ydb/core/tx/datashard/export_s3_buffer_raw.h b/ydb/core/tx/datashard/export_s3_buffer_raw.h
@@ -27,7 +27,7 @@ class TS3BufferRaw: public NExportScan::IBuffer {
     inline ui64 GetRowsLimit() const { return RowsLimit; }
     inline ui64 GetBytesLimit() const { return BytesLimit; }
 
-    void Collect(const NTable::IScan::TRow& row, IOutputStream& out);
+    bool Collect(const NTable::IScan::TRow& row, IOutputStream& out);
     virtual TMaybe<TBuffer> Flush(bool prepare);
 
 private:
@@ -42,6 +42,7 @@ class TS3BufferRaw: public NExportScan::IBuffer {
     ui64 BytesRead;
     TBuffer Buffer;
 
+    TStringBuilder ErrorString;
 }; // TS3BufferRaw
 
 } // NDataShard
diff --git a/ydb/core/tx/datashard/export_s3_buffer_zstd.cpp b/ydb/core/tx/datashard/export_s3_buffer_zstd.cpp
@@ -62,7 +62,9 @@ class TS3BufferZstd: public TS3BufferRaw {
     bool Collect(const NTable::IScan::TRow& row) override {
         BufferRaw.clear();
         TStringOutput out(BufferRaw);
-        TS3BufferRaw::Collect(row, out);
+        if (!TS3BufferRaw::Collect(row, out)) {
+            return false;
+        }
 
         BytesRaw += BufferRaw.size();
 
diff --git a/ydb/core/tx/schemeshard/ut_cdc_stream_reboots/ut_cdc_stream_reboots.cpp b/ydb/core/tx/schemeshard/ut_cdc_stream_reboots/ut_cdc_stream_reboots.cpp
@@ -570,38 +570,6 @@ Y_UNIT_TEST_SUITE(TCdcStreamWithRebootsTests) {
         }
     }
 
-    void UploadRows(TTestActorRuntime& runtime, const TString& tablePath, int partitionIdx,
-            const TVector<ui32>& keyTags, const TVector<ui32>& valueTags, const TVector<ui32>& recordIds)
-    {
-        auto tableDesc = DescribePath(runtime, tablePath, true, true);
-        const auto& tablePartitions = tableDesc.GetPathDescription().GetTablePartitions();
-        UNIT_ASSERT(partitionIdx < tablePartitions.size());
-
-        auto ev = MakeHolder<TEvDataShard::TEvUploadRowsRequest>();
-        ev->Record.SetTableId(tableDesc.GetPathId());
-
-        auto& scheme = *ev->Record.MutableRowScheme();
-        for (ui32 tag : keyTags) {
-            scheme.AddKeyColumnIds(tag);
-        }
-        for (ui32 tag : valueTags) {
-            scheme.AddValueColumnIds(tag);
-        }
-
-        for (ui32 i : recordIds) {
-            auto key = TVector<TCell>{TCell::Make(i)};
-            auto value = TVector<TCell>{TCell::Make(i)};
-
-            auto& row = *ev->Record.AddRows();
-            row.SetKeyColumns(TSerializedCellVec::Serialize(key));
-            row.SetValueColumns(TSerializedCellVec::Serialize(value));
-        }
-
-        const auto& sender = runtime.AllocateEdgeActor();
-        ForwardToTablet(runtime, tablePartitions[partitionIdx].GetDatashardId(), sender, ev.Release());
-        runtime.GrabEdgeEvent<TEvDataShard::TEvUploadRowsResponse>(sender);
-    }
-
     Y_UNIT_TEST(SplitTable) {
         TTestWithReboots t;
         t.Run([&](TTestActorRuntime& runtime, bool& activeZone) {
diff --git a/ydb/core/tx/schemeshard/ut_export/ut_export.cpp b/ydb/core/tx/schemeshard/ut_export/ut_export.cpp
@@ -1372,4 +1372,41 @@ partitioning_settings {
         env.TestWaitNotification(runtime, exportId);
         TestGetExport(runtime, exportId, "/MyRoot");
     }
+
+    Y_UNIT_TEST(CorruptedBadDyNumber) {
+        TTestBasicRuntime runtime;
+        TTestEnv env(runtime, TTestEnvOptions().DisableStatsBatching(true));
+        ui64 txId = 100;
+
+        TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+                Name: "Table"
+                Columns { Name: "key" Type: "Utf8" }
+                Columns { Name: "value" Type: "DyNumber" }
+                KeyColumnNames: ["key"]
+            )");
+        env.TestWaitNotification(runtime, txId);
+
+        // Write bad DyNumber
+        UploadRows(runtime, "/MyRoot/Table", 0, {1}, {2}, {1});
+
+        TPortManager portManager;
+        const ui16 port = portManager.GetPort();
+
+        TS3Mock s3Mock({}, TS3Mock::TSettings(port));
+        UNIT_ASSERT(s3Mock.Start());
+
+        TestExport(runtime, ++txId, "/MyRoot", Sprintf(R"(
+                ExportToS3Settings {
+                endpoint: "localhost:%d"
+                scheme: HTTP
+                items {
+                    source_path: "/MyRoot/Table"
+                    destination_prefix: ""
+                }
+                }
+            )", port));
+        env.TestWaitNotification(runtime, txId);
+
+        TestGetExport(runtime, txId, "/MyRoot", Ydb::StatusIds::CANCELLED);
+    }
 }
diff --git a/ydb/core/tx/schemeshard/ut_helpers/helpers.cpp b/ydb/core/tx/schemeshard/ut_helpers/helpers.cpp
@@ -2258,4 +2258,35 @@ namespace NSchemeShardUT_Private {
         data.push_back({1, message});
         NKikimr::NPQ::CmdWrite(&runtime, tabletId, edge, partitionId, "sourceid0", msgSeqNo, data, false, {}, true, cookie, 0);
     }
+
+    void UploadRows(TTestActorRuntime& runtime, const TString& tablePath, int partitionIdx, const TVector<ui32>& keyTags, const TVector<ui32>& valueTags, const TVector<ui32>& recordIds)
+    {
+        auto tableDesc = DescribePath(runtime, tablePath, true, true);
+        const auto& tablePartitions = tableDesc.GetPathDescription().GetTablePartitions();
+        UNIT_ASSERT(partitionIdx < tablePartitions.size());
+
+        auto ev = MakeHolder<TEvDataShard::TEvUploadRowsRequest>();
+        ev->Record.SetTableId(tableDesc.GetPathId());
+
+        auto& scheme = *ev->Record.MutableRowScheme();
+        for (ui32 tag : keyTags) {
+            scheme.AddKeyColumnIds(tag);
+        }
+        for (ui32 tag : valueTags) {
+            scheme.AddValueColumnIds(tag);
+        }
+
+        for (ui32 i : recordIds) {
+            auto key = TVector<TCell>{TCell::Make(i)};
+            auto value = TVector<TCell>{TCell::Make(i)};
+
+            auto& row = *ev->Record.AddRows();
+            row.SetKeyColumns(TSerializedCellVec::Serialize(key));
+            row.SetValueColumns(TSerializedCellVec::Serialize(value));
+        }
+
+        const auto& sender = runtime.AllocateEdgeActor();
+        ForwardToTablet(runtime, tablePartitions[partitionIdx].GetDatashardId(), sender, ev.Release());
+        runtime.GrabEdgeEvent<TEvDataShard::TEvUploadRowsResponse>(sender);
+    }
 }
diff --git a/ydb/core/tx/schemeshard/ut_helpers/helpers.h b/ydb/core/tx/schemeshard/ut_helpers/helpers.h
@@ -542,4 +542,6 @@ namespace NSchemeShardUT_Private {
     void SendTEvPeriodicTopicStats(TTestActorRuntime& runtime, ui64 topicId, ui64 generation, ui64 round, ui64 dataSize, ui64 usedReserveSize);
     void WriteToTopic(TTestActorRuntime& runtime, const TString& path, ui32& msgSeqNo, const TString& message);
 
+    void UploadRows(TTestActorRuntime& runtime, const TString& tablePath, int partitionIdx, const TVector<ui32>& keyTags, const TVector<ui32>& valueTags, const TVector<ui32>& recordIds);
+
 } //NSchemeShardUT_Private
