From 18d517202ec73242ce07163c192267453dd96bfb Mon Sep 17 00:00:00 2001 From: Ilnaz Nizametdinov Date: Thu, 14 Apr 2022 14:14:07 +0300 Subject: Request additional data, buffer expansions, improved logging KIKIMR-14714 ref:c4512da3e927ef4f57f6f5db6e4a0c21c2e6bbf4 --- ydb/core/protos/flat_scheme_op.proto | 1 + ydb/core/tx/datashard/import_s3.cpp | 128 +++++++++++++++++---------------- ydb/core/tx/schemeshard/ut_restore.cpp | 29 +++++++- ydb/core/wrappers/s3_wrapper.cpp | 34 +++++---- ydb/core/wrappers/s3_wrapper.h | 41 +++++++++-- 5 files changed, 153 insertions(+), 80 deletions(-) diff --git a/ydb/core/protos/flat_scheme_op.proto b/ydb/core/protos/flat_scheme_op.proto index e26c0bd845c..5790fa91f8b 100644 --- a/ydb/core/protos/flat_scheme_op.proto +++ b/ydb/core/protos/flat_scheme_op.proto @@ -869,6 +869,7 @@ message TS3Settings { message TLimits { optional uint32 ReadBatchSize = 1 [default = 8388608]; // 8 MB optional uint32 WriteBatchSize = 2 [default = 67108864]; // 64 MB + optional uint32 ReadBufferSizeLimit = 3 [default = 67108864]; // 64 MB }; optional TLimits Limits = 100; diff --git a/ydb/core/tx/datashard/import_s3.cpp b/ydb/core/tx/datashard/import_s3.cpp index ecabfe7bb8a..3d4f20c2e22 100644 --- a/ydb/core/tx/datashard/import_s3.cpp +++ b/ydb/core/tx/datashard/import_s3.cpp @@ -39,8 +39,9 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { } public: - explicit TReadController(ui32 rangeSize) + explicit TReadController(ui32 rangeSize, ui64 bufferSizeLimit) : RangeSize(rangeSize) + , BufferSizeLimit(bufferSizeLimit) { } @@ -56,6 +57,7 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { bool Feed(TString&& portion, TStringBuf& buf) { if (Buffer && Pos) { Buffer.remove(0, Pos); + Pos = 0; } if (!Buffer) { @@ -64,19 +66,36 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { Buffer.append(portion); } - Pos = Buffer.rfind('\n'); - if (TString::npos == Pos) { + const ui64 pos = Buffer.rfind('\n'); + if (TString::npos == pos) { return false; } - Pos += 1; // for '\n' + Pos = pos + 1; // for '\n' buf = TStringBuf(Buffer.data(), Pos); return true; } + bool CanExpandBuffer(ui64 contentLength, ui64 processedBytes, TString& reason) const { + const auto size = Buffer.size(); + + if (SumWithSaturation(processedBytes, size) >= contentLength) { + reason = "reached end of file"; + return false; + } + + if (size >= BufferSizeLimit) { + reason = "reached buffer size limit"; + return false; + } + + return true; + } + private: const ui32 RangeSize; + const ui64 BufferSizeLimit; TString Buffer; ui64 Pos = 0; @@ -126,8 +145,7 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { }; void AllocateResource() { - IMPORT_LOG_D("AllocateResource" - << ": self# " << SelfId()); + IMPORT_LOG_D("AllocateResource"); const auto* appData = AppData(); Send(MakeResourceBrokerID(), new TEvResourceBroker::TEvSubmitTask( @@ -142,18 +160,17 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { } void Handle(TEvResourceBroker::TEvResourceAllocated::TPtr& ev) { - IMPORT_LOG_D("Handle TEvResourceBroker::TEvResourceAllocated" - << ": self# " << SelfId() - << ", taskId# " << ev->Get()->TaskId); + IMPORT_LOG_I("Handle TEvResourceBroker::TEvResourceAllocated {" + << " TaskId: " << ev->Get()->TaskId + << " }"); TaskId = ev->Get()->TaskId; Restart(); } void Restart() { - IMPORT_LOG_D("Restart" - << ": self# " << SelfId() - << ", attempt# " << Attempt); + IMPORT_LOG_N("Restart" + << ": attempt# " << Attempt); if (const TActorId client = std::exchange(Client, TActorId())) { Send(client, new TEvents::TEvPoisonPill()); @@ -167,8 +184,7 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { void HeadObject(const TString& key) { IMPORT_LOG_D("HeadObject" - << ": self# " << SelfId() - << ", key# " << key); + << ": key# " << key); auto request = Model::HeadObjectRequest() .WithBucket(Settings.Bucket) @@ -179,8 +195,7 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { void GetObject(const TString& key, const std::pair& range) { IMPORT_LOG_D("GetObject" - << ": self# " << SelfId() - << ", key# " << key + << ": key# " << key << ", range# " << range.first << "-" << range.second); auto request = Model::GetObjectRequest() @@ -192,12 +207,9 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { } void Handle(TEvS3Wrapper::TEvHeadObjectResponse::TPtr& ev) { - const auto& result = ev->Get()->Result; - - IMPORT_LOG_D("Handle TEvS3Wrapper::TEvHeadObjectResponse" - << ": self# " << SelfId() - << ", result# " << result); + IMPORT_LOG_D("Handle " << ev->Get()->ToString()); + const auto& result = ev->Get()->Result; if (!CheckResult(result, TStringBuf("HeadObject"))) { return; } @@ -209,12 +221,9 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { } void Handle(TEvDataShard::TEvS3DownloadInfo::TPtr& ev) { - const auto& info = ev->Get()->Info; - - IMPORT_LOG_D("Handle TEvDataShard::TEvS3DownloadInfo" - << ": self# " << SelfId() - << ", info# " << info); + IMPORT_LOG_D("Handle " << ev->Get()->ToString()); + const auto& info = ev->Get()->Info; if (!info.DataETag) { Send(DataShard, new TEvDataShard::TEvStoreS3DownloadInfo(TxId, { ETag, ProcessedBytes, WrittenBytes, WrittenRows @@ -226,8 +235,10 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { } void ProcessDownloadInfo(const TS3Download& info, const TStringBuf marker) { - Y_VERIFY(info.DataETag); + IMPORT_LOG_N("Process download info at '" << marker << "'" + << ": info# " << info); + Y_VERIFY(info.DataETag); if (!CheckETag(*info.DataETag, ETag, marker)) { return; } @@ -244,13 +255,10 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { } void Handle(TEvS3Wrapper::TEvGetObjectResponse::TPtr& ev) { + IMPORT_LOG_D("Handle " << ev->Get()->ToString()); + auto& msg = *ev->Get(); const auto& result = msg.Result; - - IMPORT_LOG_D("Handle TEvS3Wrapper::TEvGetObjectResponse" - << ": self# " << SelfId() - << ", result# " << result); - const TStringBuf marker = "GetObject"; if (!CheckResult(result, marker)) { @@ -262,13 +270,21 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { } IMPORT_LOG_T("Content size" - << ": self# " << SelfId() - << ", processed-bytes# " << ProcessedBytes + << ": processed-bytes# " << ProcessedBytes << ", content-length# " << ContentLength << ", body-size# " << msg.Body.size()); if (!Reader.Feed(std::move(msg.Body), Buffer)) { - return Finish(false, "Cannot find new line symbol in data"); + TString reason; + if (!Reader.CanExpandBuffer(ContentLength, ProcessedBytes, reason)) { + return Finish(false, TStringBuilder() << "Cannot find new line symbol in data" + << ": " << reason); + } + + IMPORT_LOG_W("Cannot find new line symbol, request additional data" + << ": processed-bytes# " << ProcessedBytes + << ", content-length# " << ContentLength); + return GetObject(Settings.DataKey, Reader.NextRange(ContentLength, ProcessedBytes)); } ProcessedBytes += Buffer.size(); @@ -325,9 +341,8 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { void UploadRows() { const auto& record = RequestBuilder.GetRecord(); - IMPORT_LOG_D("Upload rows" - << ": self# " << SelfId() - << ", count# " << record->RowsSize() + IMPORT_LOG_I("Upload rows" + << ": count# " << record->RowsSize() << ", size# " << record->ByteSizeLong()); Send(DataShard, new TEvDataShard::TEvUnsafeUploadRowsRequest(TxId, record, { @@ -336,17 +351,12 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { } void Handle(TEvDataShard::TEvUnsafeUploadRowsResponse::TPtr& ev) { - const auto& record = ev->Get()->Record; - const auto& info = ev->Get()->Info; - - IMPORT_LOG_D("Handle TEvDataShard::TEvUnsafeUploadRowsResponse" - << ": self# " << SelfId() - << ", record# " << record.ShortDebugString() - << ", info# " << info); + IMPORT_LOG_D("Handle " << ev->Get()->ToString()); + const auto& record = ev->Get()->Record; switch (record.GetStatus()) { case NKikimrTxDataShard::TError::OK: - return ProcessDownloadInfo(info, TStringBuf("UploadResponse")); + return ProcessDownloadInfo(ev->Get()->Info, TStringBuf("UploadResponse")); case NKikimrTxDataShard::TError::WRONG_SHARD_STATE: // OVERLOADED return RetryUpload(); @@ -357,7 +367,7 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { default: return RestartOrFinish(record.GetErrorDescription()); - }; + } } template @@ -367,8 +377,7 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { } IMPORT_LOG_E("Error at '" << marker << "'" - << ": self# " << SelfId() - << ", error# " << result); + << ": error# " << result); RestartOrFinish(result.GetError().GetMessage().c_str()); return false; @@ -380,8 +389,7 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { } const TString error = TStringBuilder() << "ETag mismatch at '" << marker << "'" - << ": self# " << SelfId() - << ", expected# " << expected + << ": expected# " << expected << ", got# " << got; IMPORT_LOG_E(error); @@ -459,9 +467,8 @@ class TS3Downloader: public TActorBootstrapped, private TS3User { } void Finish(bool success = true, const TString& error = TString()) { - IMPORT_LOG_I("Finish" - << ": self# " << SelfId() - << ", success# " << success + IMPORT_LOG_N("Finish" + << ": success# " << success << ", error# " << error << ", writtenBytes# " << WrittenBytes << ", writtenRows# " << WrittenRows); @@ -493,8 +500,8 @@ public: return NKikimrServices::TActivity::IMPORT_S3_DOWNLOADER_ACTOR; } - static constexpr TStringBuf LogPrefix() { - return "s3"; + TStringBuf LogPrefix() const { + return LogPrefix_; } explicit TS3Downloader(const TActorId& dataShard, ui64 txId, const NKikimrSchemeOp::TRestoreTask& task, const TTableInfo& tableInfo) @@ -503,15 +510,15 @@ public: , Settings(TS3Settings::FromRestoreTask(task)) , TableInfo(tableInfo) , Scheme(task.GetTableDescription()) + , LogPrefix_(TStringBuilder() << "s3:" << TxId) , Retries(task.GetNumberOfRetries()) - , Reader(task.GetS3Settings().GetLimits().GetReadBatchSize()) + , Reader(task.GetS3Settings().GetLimits().GetReadBatchSize(), task.GetS3Settings().GetLimits().GetReadBufferSizeLimit()) { } void Bootstrap() { IMPORT_LOG_D("Bootstrap" - << ": self# " << SelfId() - << ", attempt# " << Attempt); + << ": attempt# " << Attempt); if (!CheckScheme()) { return; @@ -523,7 +530,7 @@ public: STATEFN(StateAllocateResource) { switch (ev->GetTypeRewrite()) { hFunc(TEvResourceBroker::TEvResourceAllocated, Handle); - cFunc(TEvents::TEvPoisonPill::EventType, NotifyDied); + sFunc(TEvents::TEvPoisonPill, NotifyDied); } } @@ -546,6 +553,7 @@ private: const TS3Settings Settings; const TTableInfo TableInfo; const NKikimrSchemeOp::TTableDescription Scheme; + const TString LogPrefix_; const ui32 Retries; ui32 Attempt = 0; diff --git a/ydb/core/tx/schemeshard/ut_restore.cpp b/ydb/core/tx/schemeshard/ut_restore.cpp index 5573748eb88..9a84fc19caa 100644 --- a/ydb/core/tx/schemeshard/ut_restore.cpp +++ b/ydb/core/tx/schemeshard/ut_restore.cpp @@ -326,6 +326,23 @@ Y_UNIT_TEST_SUITE(TRestoreTests) { NKqp::CompareYson(data.YsonStr, content); } + Y_UNIT_TEST(ShouldExpandBuffer) { + TTestBasicRuntime runtime; + + const auto data = GenerateTestData("a", 2); + const ui32 batchSize = 1; + + Restore(runtime, R"( + Name: "Table" + Columns { Name: "key" Type: "Utf8" } + Columns { Name: "value" Type: "Utf8" } + KeyColumnNames: ["key"] + )", {data}, batchSize); + + auto content = ReadTable(runtime, TTestTxConfig::FakeHiveTablets); + NKqp::CompareYson(data.YsonStr, content); + } + Y_UNIT_TEST(ShouldSucceedOnSupportedDatatypes) { TTestBasicRuntime runtime; @@ -596,7 +613,7 @@ Y_UNIT_TEST_SUITE(TRestoreTests) { NKqp::CompareYson(data.YsonStr, content); } - Y_UNIT_TEST(ShouldFailOnFileWithoutNewLines) { + void ShouldFailOnFileWithoutNewLines(ui32 batchSize) { TTestBasicRuntime runtime; const auto data = TTestData("\"a1\",\"value1\"", EmptyYsonStr); @@ -606,12 +623,20 @@ Y_UNIT_TEST_SUITE(TRestoreTests) { Columns { Name: "key" Type: "Utf8" } Columns { Name: "value" Type: "Utf8" } KeyColumnNames: ["key"] - )", {data}); + )", {data}, batchSize); auto content = ReadTable(runtime, TTestTxConfig::FakeHiveTablets); NKqp::CompareYson(data.YsonStr, content); } + Y_UNIT_TEST(ShouldFailOnFileWithoutNewLinesStandardBatch) { + ShouldFailOnFileWithoutNewLines(128); + } + + Y_UNIT_TEST(ShouldFailOnFileWithoutNewLinesSmallBatch) { + ShouldFailOnFileWithoutNewLines(1); + } + Y_UNIT_TEST(ShouldFailOnEmptyToken) { TTestBasicRuntime runtime; diff --git a/ydb/core/wrappers/s3_wrapper.cpp b/ydb/core/wrappers/s3_wrapper.cpp index 5d5a21cc033..e3e81b813ff 100644 --- a/ydb/core/wrappers/s3_wrapper.cpp +++ b/ydb/core/wrappers/s3_wrapper.cpp @@ -95,6 +95,10 @@ TS3User::~TS3User() { class TS3Wrapper: public TActor, private TS3User { template class TContextBase: public AsyncCallerContext { + protected: + using TRequest = typename TEvRequest::TRequest; + using TOutcome = typename TEvResponse::TOutcome; + public: explicit TContextBase(const TActorSystem* sys, const TActorId& sender) : AsyncCallerContext() @@ -108,16 +112,14 @@ class TS3Wrapper: public TActor, private TS3User { return ActorSystem; } - virtual const typename TEvRequest::TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) { + virtual const TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) { return ev->Get()->Request; } - void Reply(const typename TEvRequest::TRequest& request, - const typename TEvResponse::TOutcome& outcome) const - { + void Reply(const TRequest& request, const TOutcome& outcome) const { Y_VERIFY(!std::exchange(Replied, true), "Double-reply"); - std::optional key; + typename TEvResponse::TKey key; if (request.KeyHasBeenSet()) { key = request.GetKey(); } @@ -134,9 +136,7 @@ class TS3Wrapper: public TActor, private TS3User { Send(Sender, ev); } - virtual THolder MakeResponse(const std::optional& key, - const typename TEvResponse::TOutcome& outcome) const - { + virtual THolder MakeResponse(const typename TEvResponse::TKey& key, const TOutcome& outcome) const { return MakeHolder(key, outcome); } @@ -150,6 +150,11 @@ class TS3Wrapper: public TActor, private TS3User { template class TOutputStreamContext: public TContextBase { + protected: + using TRequest = typename TEvRequest::TRequest; + using TOutcome = typename TEvResponse::TOutcome; + + private: class TOutputStreamBuf: public PreallocatedStreamBuf { TOutputStreamBuf(char* data, size_t size) : PreallocatedStreamBuf(reinterpret_cast(data), size) @@ -186,7 +191,7 @@ class TS3Wrapper: public TActor, private TS3User { public: using TContextBase::TContextBase; - const typename TEvRequest::TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) override { + const TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) override { auto& request = ev->Get()->Request; std::pair range; @@ -202,9 +207,7 @@ class TS3Wrapper: public TActor, private TS3User { } protected: - THolder MakeResponse(const std::optional& key, - const typename TEvResponse::TOutcome& outcome) const override - { + THolder MakeResponse(const typename TEvResponse::TKey& key, const TOutcome& outcome) const override { if (outcome.IsSuccess()) { return MakeHolder(key, outcome, std::move(Buffer)); } else { @@ -219,6 +222,11 @@ class TS3Wrapper: public TActor, private TS3User { template class TInputStreamContext: public TContextBase { + protected: + using TRequest = typename TEvRequest::TRequest; + using TOutcome = typename TEvResponse::TOutcome; + + private: class TInputStreamBuf: public PreallocatedStreamBuf { TInputStreamBuf(char* data, size_t size) : PreallocatedStreamBuf(reinterpret_cast(data), size) @@ -240,7 +248,7 @@ class TS3Wrapper: public TActor, private TS3User { public: using TContextBase::TContextBase; - const typename TEvRequest::TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) override { + const TRequest& PrepareRequest(typename TEvRequest::TPtr& ev) override { auto& request = ev->Get()->Request; Buffer = std::move(ev->Get()->Body); diff --git a/ydb/core/wrappers/s3_wrapper.h b/ydb/core/wrappers/s3_wrapper.h index c623113e9a0..369621f29e2 100644 --- a/ydb/core/wrappers/s3_wrapper.h +++ b/ydb/core/wrappers/s3_wrapper.h @@ -16,6 +16,8 @@ #include +#include + namespace NKikimr { namespace NWrappers { @@ -35,6 +37,12 @@ struct TEvS3Wrapper { { } + TString ToString() const override { + return TStringBuilder() << this->ToStringHeader() << " {" + << " Request: " << Request + << " }"; + } + using TBase = TGenericRequest; }; @@ -50,6 +58,13 @@ struct TEvS3Wrapper { { } + TString ToString() const override { + return TStringBuilder() << this->ToStringHeader() << " {" + << " Request: " << this->Request + << " Body: " << Body.size() << "b" + << " }"; + } + using TBase = TRequestWithBody; }; @@ -57,11 +72,12 @@ struct TEvS3Wrapper { struct TGenericResponse: public TEventLocal { using TOutcome = Aws::Utils::Outcome; using TResult = Aws::Utils::Outcome; + using TKey = std::optional; - std::optional Key; + TKey Key; TResult Result; - explicit TGenericResponse(const std::optional& key, const TOutcome& outcome) + explicit TGenericResponse(const TKey& key, const TOutcome& outcome) : Key(key) , Result(TDerived::ResultFromOutcome(outcome)) { @@ -71,27 +87,42 @@ struct TEvS3Wrapper { return outcome; } + TString ToString() const override { + return TStringBuilder() << this->ToStringHeader() << " {" + << " Key: " << (Key ? "null" : *Key) + << " Result: " << Result + << " }"; + } + using TBase = TGenericResponse; }; template struct TResponseWithBody: public TGenericResponse { using TGeneric = TGenericResponse; + using TKey = typename TGeneric::TKey; TString Body; - explicit TResponseWithBody(const std::optional& key, const typename TGeneric::TOutcome& outcome) + explicit TResponseWithBody(const TKey& key, const typename TGeneric::TOutcome& outcome) : TGeneric(key, outcome) { } - explicit TResponseWithBody(const std::optional& key, const typename TGeneric::TOutcome& outcome, - TString&& body) + explicit TResponseWithBody(const TKey& key, const typename TGeneric::TOutcome& outcome, TString&& body) : TGeneric(key, outcome) , Body(std::move(body)) { } + TString ToString() const override { + return TStringBuilder() << this->ToStringHeader() << " {" + << " Key: " << (this->Key ? "null" : *this->Key) + << " Result: " << this->Result + << " Body: " << Body.size() << "b" + << " }"; + } + using TBase = TResponseWithBody; }; -- cgit v1.3