Skip to content

Commit d4871a7

Browse files
evanevanevanevannnnevanevanevanevannnn
authored
Merge 970ebb2 into d7c05c0 
2 parents d7c05c0 + 970ebb2 commit d4871a7

File tree

6 files changed

+33
-17
lines changed

6 files changed

+33
-17
lines changed

‎ydb/library/yql/providers/s3/actors/yql_s3_raw_read_actor.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -284,7 +284,7 @@ class TS3ReadActor : public NActors::TActorBootstrapped<TS3ReadActor>, public ID
284284
IssuesFromMessage(result->Get()->Record.GetIssues(), issues);
285285
LOG_E("TS3ReadActor", "Error while object listing, details: TEvObjectPathReadError: " << issues.ToOneLineString());
286286
issues = NS3Util::AddParentIssue(TStringBuilder{} << "Error while object listing", std::move(issues));
287-
Send(ComputeActorId, new TEvAsyncInputError(InputIndex, issues, NYql::NDqProto::StatusIds::EXTERNAL_ERROR));
287+
Send(ComputeActorId, new TEvAsyncInputError(InputIndex, issues, result->Get()->Record.GetFatalCode()));
288288
}
289289

290290
void HandleAck(TEvS3Provider::TEvAck::TPtr& ev) {

‎ydb/library/yql/providers/s3/actors/yql_s3_read_actor.cpp‎

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -965,6 +965,7 @@ class TS3ReadCoroImpl : public TActorCoroImpl {
965965
message = ErrorText;
966966
}
967967
Issues.AddIssues(BuildIssues(HttpResponseCode, errorCode, message));
968+
FatalCode = StatusFromS3ErrorCode(errorCode);
968969
}
969970

970971
if (ev->Get()->Issues) {
@@ -1115,15 +1116,15 @@ class TS3ReadCoroImpl : public TActorCoroImpl {
11151116
DecompressorActorId = Register(CreateS3DecompressorActor(SelfActorId, ReadSpec->Compression));
11161117
}
11171118

1118-
NYql::NDqProto::StatusIds::StatusCode fatalCode = NYql::NDqProto::StatusIds::EXTERNAL_ERROR;
1119+
FatalCode = NYql::NDqProto::StatusIds::EXTERNAL_ERROR;
11191120

11201121
StartCycleCount = GetCycleCountFast();
11211122

11221123
try {
11231124
if (ReadSpec->Arrow) {
11241125
if (ReadSpec->Compression) {
11251126
Issues.AddIssue(TIssue("Blocks optimisations are incompatible with external compression"));
1126-
fatalCode = NYql::NDqProto::StatusIds::BAD_REQUEST;
1127+
FatalCode = NYql::NDqProto::StatusIds::BAD_REQUEST;
11271128
} else {
11281129
try {
11291130
if (Url.StartsWith("file://")) {
@@ -1133,7 +1134,7 @@ class TS3ReadCoroImpl : public TActorCoroImpl {
11331134
}
11341135
} catch (const parquet::ParquetException& ex) {
11351136
Issues.AddIssue(TIssue(ex.what()));
1136-
fatalCode = NYql::NDqProto::StatusIds::BAD_REQUEST;
1137+
FatalCode = NYql::NDqProto::StatusIds::BAD_REQUEST;
11371138
RetryStuff->Cancel();
11381139
}
11391140
}
@@ -1149,7 +1150,7 @@ class TS3ReadCoroImpl : public TActorCoroImpl {
11491150
LOG_CORO_D("S3 read ERROR");
11501151
} catch (const NDB::Exception& ex) {
11511152
Issues.AddIssue(TIssue(ex.message()));
1152-
fatalCode = NYql::NDqProto::StatusIds::BAD_REQUEST;
1153+
FatalCode = NYql::NDqProto::StatusIds::BAD_REQUEST;
11531154
RetryStuff->Cancel();
11541155
}
11551156
}
@@ -1162,15 +1163,15 @@ class TS3ReadCoroImpl : public TActorCoroImpl {
11621163
return;
11631164
} catch (const std::exception& err) {
11641165
Issues.AddIssue(TIssue(err.what()));
1165-
fatalCode = NYql::NDqProto::StatusIds::INTERNAL_ERROR;
1166+
FatalCode = NYql::NDqProto::StatusIds::INTERNAL_ERROR;
11661167
RetryStuff->Cancel();
11671168
}
11681169

11691170
CpuTime += GetCpuTimeDelta();
11701171

11711172
auto issues = NS3Util::AddParentIssue(TStringBuilder{} << "Error while reading file " << Path, std::move(Issues));
11721173
if (issues)
1173-
Send(ComputeActorId, new IDqComputeActorAsyncInput::TEvAsyncInputError(InputIndex, std::move(issues), fatalCode));
1174+
Send(ComputeActorId, new IDqComputeActorAsyncInput::TEvAsyncInputError(InputIndex, std::move(issues), FatalCode));
11741175
else
11751176
Send(ParentActorId, new TEvS3Provider::TEvFileFinished(PathIndex, TakeIngressDelta(), TakeCpuTimeDelta(), RetryStuff->SizeLimit));
11761177
}
@@ -1230,6 +1231,7 @@ class TS3ReadCoroImpl : public TActorCoroImpl {
12301231
bool ServerReturnedError = false;
12311232
TString ErrorText;
12321233
TIssues Issues;
1234+
NYql::NDqProto::StatusIds::StatusCode FatalCode;
12331235

12341236
NActors::TActorId DecompressorActorId;
12351237
std::size_t LastOffset = 0;
@@ -1719,7 +1721,7 @@ class TS3StreamReadActor : public TActorBootstrapped<TS3StreamReadActor>, public
17191721
IssuesFromMessage(result->Get()->Record.GetIssues(), issues);
17201722
LOG_W("TS3StreamReadActor", "Error while object listing, details: TEvObjectPathReadError: " << issues.ToOneLineString());
17211723
issues = NS3Util::AddParentIssue(TStringBuilder{} << "Error while object listing", std::move(issues));
1722-
Send(ComputeActorId, new TEvAsyncInputError(InputIndex, std::move(issues), NYql::NDqProto::StatusIds::EXTERNAL_ERROR));
1724+
Send(ComputeActorId, new TEvAsyncInputError(InputIndex, std::move(issues), result->Get()->Record.GetFatalCode()));
17231725
}
17241726

17251727
void HandleRetry(TEvS3Provider::TEvRetryEventFunc::TPtr& retry) {

‎ydb/library/yql/providers/s3/actors/yql_s3_source_queue.cpp‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -183,6 +183,7 @@ class TS3FileQueueActor : public NActors::TActorBootstrapped<TS3FileQueueActor>
183183
, FileSizeLimit(fileSizeLimit)
184184
, ReadLimit(readLimit)
185185
, MaybeIssues(Nothing())
186+
, FatalCode(NYql::NDqProto::StatusIds::EXTERNAL_ERROR)
186187
, UseRuntimeListing(useRuntimeListing)
187188
, ConsumersCount(consumersCount)
188189
, BatchSizeLimit(batchSizeLimit)
@@ -302,6 +303,7 @@ class TS3FileQueueActor : public NActors::TActorBootstrapped<TS3FileQueueActor>
302303
<< " and exceeds limit = " << FileSizeLimit;
303304
LOG_E("TS3FileQueueActor", errorMessage);
304305
MaybeIssues = TIssues{TIssue{errorMessage}};
306+
FatalCode = NYql::NDqProto::StatusIds::PRECONDITION_FAILED;
305307
return false;
306308
}
307309
LOG_T("TS3FileQueueActor", "SaveRetrievedResults adding path: " << object.Path);
@@ -377,7 +379,7 @@ class TS3FileQueueActor : public NActors::TActorBootstrapped<TS3FileQueueActor>
377379
LOG_D(
378380
"TS3FileQueueActor",
379381
"HandleGetNextBatchForErrorState Giving away rest of Objects");
380-
Send(ev->Sender, new TEvS3Provider::TEvObjectPathReadError(*MaybeIssues, ev->Get()->Record.GetTransportMeta()));
382+
Send(ev->Sender, new TEvS3Provider::TEvObjectPathReadError(*MaybeIssues, FatalCode, ev->Get()->Record.GetTransportMeta()));
381383
TryFinish(ev->Sender, ev->Get()->Record.GetTransportMeta().GetSeqNo());
382384
}
383385

@@ -558,7 +560,7 @@ class TS3FileQueueActor : public NActors::TActorBootstrapped<TS3FileQueueActor>
558560
if (!MaybeIssues.Defined()) {
559561
SendObjects(consumer, requests.front());
560562
} else {
561-
Send(consumer, new TEvS3Provider::TEvObjectPathReadError(*MaybeIssues, requests.front()));
563+
Send(consumer, new TEvS3Provider::TEvObjectPathReadError(*MaybeIssues, FatalCode, requests.front()));
562564
TryFinish(consumer, requests.front().GetSeqNo());
563565
}
564566
requests.pop_front();
@@ -603,6 +605,7 @@ class TS3FileQueueActor : public NActors::TActorBootstrapped<TS3FileQueueActor>
603605
size_t CurrentDirectoryPathIndex = 0;
604606
THashMap<NActors::TActorId, TDeque<NDqProto::TMessageTransportMeta>> PendingRequests;
605607
TMaybe<TIssues> MaybeIssues;
608+
NYql::NDqProto::StatusIds::StatusCode FatalCode;
606609
bool UseRuntimeListing;
607610
ui64 ConsumersCount;
608611
ui64 BatchSizeLimit;

‎ydb/library/yql/providers/s3/events/events.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -100,9 +100,10 @@ struct TEvS3Provider {
100100

101101
TEvObjectPathReadError() = default;
102102

103-
TEvObjectPathReadError(TIssues issues, const NDqProto::TMessageTransportMeta& transportMeta) {
103+
TEvObjectPathReadError(TIssues issues, NYql::NDqProto::StatusIds::StatusCode code, const NDqProto::TMessageTransportMeta& transportMeta) {
104104
NYql::IssuesToMessage(issues, Record.MutableIssues());
105105
Record.MutableTransportMeta()->CopyFrom(transportMeta);
106+
Record.SetFatalCode(code);
106107
}
107108
};
108109

‎ydb/library/yql/providers/s3/object_listers/yql_s3_list.cpp‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -133,11 +133,15 @@ TS3ListObjectV2Response ParseListObjectV2Response(
133133
if (const auto& root = xml.Root(); root.Name() == "Error") {
134134
const auto& code = root.Node("Code", true).Value<TString>();
135135
const auto& message = root.Node("Message", true).Value<TString>();
136-
ythrow yexception() << message << ", error: code: " << code << ", request id: ["
137-
<< requestId << "]";
136+
const auto errorMessage = TStringBuilder{} << message << ", error: code: " << code
137+
<< ", request id: [" << requestId << "]";
138+
YQL_CLOG(DEBUG, ProviderS3) << "[TS3Lister::ParseListObjectV2Response] " << errorMessage;
139+
throw yexception() << errorMessage;
138140
} else if (root.Name() != "ListBucketResult") {
139-
ythrow yexception() << "Unexpected response '" << root.Name()
141+
const auto errorMessage = TStringBuilder{} << "Unexpected response '" << root.Name()
140142
<< "' on discovery, request id: [" << requestId << "]";
143+
YQL_CLOG(DEBUG, ProviderS3) << "[TS3Lister::ParseListObjectV2Response] " << errorMessage;
144+
throw yexception() << errorMessage;
141145
} else {
142146
const NXml::TNamespacesForXPath nss(
143147
1U, {"s3", "http://s3.amazonaws.com/doc/2006-03-01/"});
@@ -313,7 +317,8 @@ class TS3Lister : public IS3Lister {
313317

314318
auto gateway = ctx.GatewayWeak.lock();
315319
if (!gateway) {
316-
ythrow yexception() << "Gateway disappeared";
320+
YQL_CLOG(DEBUG, ProviderS3) << "[TS3Lister::SubmitRequestIntoGateway] Gateway disappeared";
321+
throw yexception() << "Gateway disappeared";
317322
}
318323

319324
auto sharedCtx = ctx.SharedCtx;
@@ -360,7 +365,8 @@ class TS3Lister : public IS3Lister {
360365
static void OnDiscovery(TListingContext ctx, IHTTPGateway::TResult&& result) try {
361366
auto gateway = ctx.GatewayWeak.lock();
362367
if (!gateway) {
363-
ythrow yexception() << "Gateway disappeared";
368+
YQL_CLOG(DEBUG, ProviderS3) << "[TS3Lister::OnDiscovery] Gateway disappeared";
369+
throw yexception() << "Gateway disappeared";
364370
}
365371
if (!result.Issues) {
366372
auto xmlString = result.Content.Extract();
@@ -539,8 +545,10 @@ IS3Lister::TPtr MakeS3Lister(
539545
}
540546

541547
if (!allowLocalFiles) {
542-
ythrow yexception() << "Using local files as DataSource isn't allowed, but trying access "
548+
const auto errorMessage = TStringBuilder{} << "Using local files as DataSource isn't allowed, but trying access "
543549
<< listingRequest.Url;
550+
YQL_CLOG(DEBUG, ProviderS3) << "[IS3Lister::MakeS3Lister] " << errorMessage;
551+
throw yexception() << errorMessage;
544552
}
545553
return std::make_shared<TLocalS3Lister>(listingRequest, delimiter);
546554
}

‎ydb/library/yql/providers/s3/proto/file_queue.proto‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ option cc_enable_arenas = true;
44
package NYql.NS3.FileQueue;
55

66
import "ydb/library/yql/dq/actors/protos/dq_events.proto";
7+
import "ydb/library/yql/dq/actors/protos/dq_status_codes.proto";
78
import "ydb/public/api/protos/ydb_issue_message.proto";
89

910
message TEvUpdateConsumersCount {
@@ -29,6 +30,7 @@ message TEvObjectPathBatch {
2930

3031
message TEvObjectPathReadError {
3132
repeated Ydb.Issue.IssueMessage Issues = 1;
33+
optional NYql.NDqProto.StatusIds.StatusCode FatalCode = 2;
3234

3335
optional NYql.NDqProto.TMessageTransportMeta TransportMeta = 100;
3436
}

0 commit comments

Comments
 (0)