Skip to content

Commit c7e36ae

Browse files
author
Vladislav Gogov
authored
Fix #11186 (#11631)
1 parent cb324c9 commit c7e36ae

File tree

7 files changed

+62
-30
lines changed

7 files changed

+62
-30
lines changed

‎ydb/core/kqp/ut/olap/kqp_olap_ut.cpp

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2907,6 +2907,31 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
29072907
UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
29082908
}
29092909
}
2910+
2911+
Y_UNIT_TEST(ScanFailedSnapshotTooOld) {
2912+
NKikimrConfig::TAppConfig appConfig;
2913+
appConfig.MutableTableServiceConfig()->SetEnableOlapSink(true);
2914+
appConfig.MutableColumnShardConfig()->SetMaxReadStaleness_ms(5000);
2915+
auto settings = TKikimrSettings().SetAppConfig(appConfig).SetWithSampleTables(false);
2916+
TTestHelper testHelper(settings);
2917+
2918+
TTestHelper::TColumnTable cnt;
2919+
TVector<TTestHelper::TColumnSchema> schema = {
2920+
TTestHelper::TColumnSchema().SetName("key").SetType(NScheme::NTypeIds::Int32).SetNullable(false),
2921+
TTestHelper::TColumnSchema().SetName("c").SetType(NScheme::NTypeIds::Int32).SetNullable(true)
2922+
};
2923+
cnt.SetName("/Root/cnt").SetPrimaryKey({ "key" }).SetSchema(schema);
2924+
testHelper.CreateTable(cnt);
2925+
Sleep(TDuration::Seconds(10));
2926+
auto client = testHelper.GetKikimr().GetQueryClient();
2927+
auto result =
2928+
client
2929+
.ExecuteQuery(
2930+
TStringBuilder() << "$v = SELECT CAST(COUNT(*) AS INT32) FROM `/Root/cnt`; INSERT INTO `/Root/cnt` (key, c) values(1, $v);",
2931+
NYdb::NQuery::TTxControl::BeginTx().CommitTx())
2932+
.GetValueSync();
2933+
UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::SUCCESS);
2934+
}
29102935
}
29112936

29122937
}

‎ydb/core/tx/columnshard/columnshard__write.cpp

Lines changed: 14 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,8 @@ namespace NKikimr::NColumnShard {
1919

2020
using namespace NTabletFlatExecutor;
2121

22-
void TColumnShard::OverloadWriteFail(const EOverloadStatus overloadReason, const NEvWrite::TWriteMeta& writeMeta, const ui64 writeSize, const ui64 cookie,
23-
std::unique_ptr<NActors::IEventBase>&& event, const TActorContext& ctx) {
22+
void TColumnShard::OverloadWriteFail(const EOverloadStatus overloadReason, const NEvWrite::TWriteMeta& writeMeta, const ui64 writeSize,
23+
const ui64 cookie, std::unique_ptr<NActors::IEventBase>&& event, const TActorContext& ctx) {
2424
Counters.GetTabletCounters()->IncCounter(COUNTER_WRITE_FAIL);
2525
switch (overloadReason) {
2626
case EOverloadStatus::Disk:
@@ -262,8 +262,8 @@ void TColumnShard::Handle(TEvColumnShard::TEvWrite::TPtr& ev, const TActorContex
262262
<< (writeMeta.GetWriteId() ? (" writeId " + ToString(writeMeta.GetWriteId())).c_str() : " ")
263263
<< Counters.GetWritesMonitor()->DebugString() << " at tablet " << TabletID());
264264
writeData.MutableWriteMeta().SetWriteMiddle1StartInstant(TMonotonic::Now());
265-
std::shared_ptr<NConveyor::ITask> task = std::make_shared<NOlap::TBuildBatchesTask>(
266-
TabletID(), SelfId(), BufferizationWriteActorId, std::move(writeData), snapshotSchema, GetLastTxSnapshot(), Counters.GetCSCounters().WritingCounters);
265+
std::shared_ptr<NConveyor::ITask> task = std::make_shared<NOlap::TBuildBatchesTask>(TabletID(), SelfId(), BufferizationWriteActorId,
266+
std::move(writeData), snapshotSchema, GetLastTxSnapshot(), Counters.GetCSCounters().WritingCounters);
267267
NConveyor::TInsertServiceOperator::AsyncTaskToExecute(task);
268268
}
269269
}
@@ -285,7 +285,8 @@ class TCommitOperation {
285285
}
286286

287287
TCommitOperation(const ui64 tabletId)
288-
: TabletId(tabletId) {
288+
: TabletId(tabletId)
289+
{
289290
}
290291

291292
TConclusionStatus Parse(const NEvents::TDataEvents::TEvWrite& evWrite) {
@@ -357,7 +358,8 @@ class TProposeWriteTransaction: public NTabletFlatExecutor::TTransactionBase<TCo
357358
: TBase(self)
358359
, WriteCommit(op)
359360
, Source(source)
360-
, Cookie(cookie) {
361+
, Cookie(cookie)
362+
{
361363
}
362364

363365
virtual bool Execute(TTransactionContext& txc, const TActorContext&) override {
@@ -402,7 +404,8 @@ class TAbortWriteTransaction: public NTabletFlatExecutor::TTransactionBase<TColu
402404
: TBase(self)
403405
, TxId(txId)
404406
, Source(source)
405-
, Cookie(cookie) {
407+
, Cookie(cookie)
408+
{
406409
}
407410

408411
virtual bool Execute(TTransactionContext& txc, const TActorContext&) override {
@@ -466,9 +469,8 @@ void TColumnShard::Handle(NEvents::TDataEvents::TEvWrite::TPtr& ev, const TActor
466469
NKikimrDataEvents::TEvWriteResult::STATUS_ABORTED);
467470
} else {
468471
if (lockInfo->GetGeneration() != commitOperation->GetGeneration()) {
469-
sendError("tablet lock have another generation: " + ::ToString(lockInfo->GetGeneration()) +
470-
" != " + ::ToString(commitOperation->GetGeneration()),
471-
NKikimrDataEvents::TEvWriteResult::STATUS_LOCKS_BROKEN);
472+
sendError("tablet lock have another generation: " + ::ToString(lockInfo->GetGeneration()) + " != " +
473+
::ToString(commitOperation->GetGeneration()), NKikimrDataEvents::TEvWriteResult::STATUS_LOCKS_BROKEN);
472474
} else if (lockInfo->GetInternalGenerationCounter() != commitOperation->GetInternalGenerationCounter()) {
473475
sendError(
474476
"tablet lock have another internal generation counter: " + ::ToString(lockInfo->GetInternalGenerationCounter()) +
@@ -567,7 +569,7 @@ void TColumnShard::Handle(NEvents::TDataEvents::TEvWrite::TPtr& ev, const TActor
567569
auto writeOperation = OperationsManager->RegisterOperation(lockId, cookie, granuleShardingVersionId, *mType);
568570
Y_ABORT_UNLESS(writeOperation);
569571
writeOperation->SetBehaviour(behaviour);
570-
writeOperation->Start(*this, tableId, arrowData, source, schema, ctx);
572+
writeOperation->Start(*this, tableId, arrowData, source, schema, ctx, NOlap::TSnapshot::Max());
571573
}
572574

573-
} // namespace NKikimr::NColumnShard
575+
} // namespace NKikimr::NColumnShard

‎ydb/core/tx/columnshard/data_reader/actor.cpp

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ void TActor::HandleExecute(NKqp::TEvKqpCompute::TEvScanData::TPtr& ev) {
2323
} else {
2424
AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "restore_task_finished")("reason", status.GetErrorMessage());
2525
}
26+
PassAway();
2627
}
2728
}
2829

@@ -35,10 +36,11 @@ void TActor::HandleExecute(NKqp::TEvKqpCompute::TEvScanInitActor::TPtr& ev) {
3536
}
3637

3738
void TActor::HandleExecute(NKqp::TEvKqpCompute::TEvScanError::TPtr& ev) {
38-
SwitchStage(EStage::WaitData, EStage::Finished);
39+
SwitchStage(std::nullopt, EStage::Finished);
3940
AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "problem_on_restore_data")(
4041
"reason", NYql::IssuesFromMessageAsString(ev->Get()->Record.GetIssues()));
4142
RestoreTask->OnError(NYql::IssuesFromMessageAsString(ev->Get()->Record.GetIssues()));
43+
PassAway();
4244
}
4345

4446
void TActor::Bootstrap(const TActorContext& /*ctx*/) {

‎ydb/core/tx/columnshard/data_reader/actor.h

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -59,8 +59,10 @@ class TActor: public NActors::TActorBootstrapped<TActor> {
5959

6060
EStage Stage = EStage::Initialization;
6161
static inline const ui64 FreeSpace = ((ui64)8) << 20;
62-
void SwitchStage(const EStage from, const EStage to) {
63-
AFL_VERIFY(Stage == from)("from", (ui32)from)("real", (ui32)Stage)("to", (ui32)to);
62+
void SwitchStage(const std::optional<EStage> from, const EStage to) {
63+
if (from) {
64+
AFL_VERIFY(Stage == *from)("from", (ui32)*from)("real", (ui32)Stage)("to", (ui32)to);
65+
}
6466
Stage = to;
6567
}
6668

‎ydb/core/tx/columnshard/operations/batch_builder/builder.h

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,8 @@ class TBuildBatchesTask: public NConveyor::ITask {
3636
, BufferActorId(bufferActorId)
3737
, ActualSchema(actualSchema)
3838
, ActualSnapshot(actualSnapshot)
39-
, WritingCounters(writingCounters) {
39+
, WritingCounters(writingCounters)
40+
{
4041
}
4142
};
42-
} // namespace NKikimr::NOlap
43+
} // namespace NKikimr::NOlap

‎ydb/core/tx/columnshard/operations/write.cpp

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,12 @@ TWriteOperation::TWriteOperation(const TOperationWriteId writeId, const ui64 loc
2020
, LockId(lockId)
2121
, Cookie(cookie)
2222
, GranuleShardingVersionId(granuleShardingVersionId)
23-
, ModificationType(mType) {
23+
, ModificationType(mType)
24+
{
2425
}
2526

2627
void TWriteOperation::Start(TColumnShard& owner, const ui64 tableId, const NEvWrite::IDataContainer::TPtr& data, const NActors::TActorId& source,
27-
const std::shared_ptr<NOlap::ISnapshotSchema>& schema, const TActorContext& ctx) {
28+
const std::shared_ptr<NOlap::ISnapshotSchema>& schema, const TActorContext& ctx, const NOlap::TSnapshot& applyToSnapshot) {
2829
Y_ABORT_UNLESS(Status == EOperationStatus::Draft);
2930

3031
NEvWrite::TWriteMeta writeMeta((ui64)WriteId, tableId, source, GranuleShardingVersionId);
@@ -34,13 +35,14 @@ void TWriteOperation::Start(TColumnShard& owner, const ui64 tableId, const NEvWr
3435
std::make_shared<NOlap::TBuildBatchesTask>(owner.TabletID(), ctx.SelfID, owner.BufferizationWriteActorId,
3536
NEvWrite::TWriteData(writeMeta, data, owner.TablesManager.GetPrimaryIndex()->GetReplaceKey(),
3637
owner.StoragesManager->GetInsertOperator()->StartWritingAction(NOlap::NBlobOperations::EConsumer::WRITING_OPERATOR)),
37-
schema, owner.GetLastTxSnapshot(), owner.Counters.GetCSCounters().WritingCounters);
38-
NConveyor::TInsertServiceOperator::AsyncTaskToExecute(task);
38+
schema, applyToSnapshot, owner.Counters.GetCSCounters().WritingCounters);
39+
NConveyor::TCompServiceOperator::SendTaskToExecute(task);
3940

4041
Status = EOperationStatus::Started;
4142
}
4243

43-
void TWriteOperation::CommitOnExecute(TColumnShard& owner, NTabletFlatExecutor::TTransactionContext& txc, const NOlap::TSnapshot& snapshot) const {
44+
void TWriteOperation::CommitOnExecute(
45+
TColumnShard& owner, NTabletFlatExecutor::TTransactionContext& txc, const NOlap::TSnapshot& snapshot) const {
4446
Y_ABORT_UNLESS(Status == EOperationStatus::Prepared);
4547

4648
TBlobGroupSelector dsGroupSelector(owner.Info());
@@ -78,12 +80,10 @@ void TWriteOperation::OnWriteFinish(
7880
TString metadata;
7981
Y_ABORT_UNLESS(proto.SerializeToString(&metadata));
8082

81-
db.Table<Schema::Operations>()
82-
.Key((ui64)WriteId)
83-
.Update(NIceDb::TUpdate<Schema::Operations::Status>((ui32)Status), NIceDb::TUpdate<Schema::Operations::CreatedAt>(CreatedAt.Seconds()),
84-
NIceDb::TUpdate<Schema::Operations::Metadata>(metadata), NIceDb::TUpdate<Schema::Operations::LockId>(LockId),
85-
NIceDb::TUpdate<Schema::Operations::Cookie>(Cookie),
86-
NIceDb::TUpdate<Schema::Operations::GranuleShardingVersionId>(GranuleShardingVersionId.value_or(0)));
83+
db.Table<Schema::Operations>().Key((ui64)WriteId).Update(NIceDb::TUpdate<Schema::Operations::Status>((ui32)Status),
84+
NIceDb::TUpdate<Schema::Operations::CreatedAt>(CreatedAt.Seconds()), NIceDb::TUpdate<Schema::Operations::Metadata>(metadata),
85+
NIceDb::TUpdate<Schema::Operations::LockId>(LockId), NIceDb::TUpdate<Schema::Operations::Cookie>(Cookie),
86+
NIceDb::TUpdate<Schema::Operations::GranuleShardingVersionId>(GranuleShardingVersionId.value_or(0)));
8787
}
8888

8989
void TWriteOperation::ToProto(NKikimrTxColumnShard::TInternalOperationData& proto) const {
@@ -119,4 +119,4 @@ void TWriteOperation::AbortOnComplete(TColumnShard& /*owner*/) const {
119119
Y_ABORT_UNLESS(Status == EOperationStatus::Prepared);
120120
}
121121

122-
} // namespace NKikimr::NColumnShard
122+
} // namespace NKikimr::NColumnShard

‎ydb/core/tx/columnshard/operations/write.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,7 @@ class TWriteOperation {
6262
const std::optional<ui32> granuleShardingVersionId, const NEvWrite::EModificationType mType);
6363

6464
void Start(TColumnShard& owner, const ui64 tableId, const NEvWrite::IDataContainer::TPtr& data, const NActors::TActorId& source,
65-
const std::shared_ptr<NOlap::ISnapshotSchema>& schema, const TActorContext& ctx);
65+
const std::shared_ptr<NOlap::ISnapshotSchema>& schema, const TActorContext& ctx, const NOlap::TSnapshot& applyToSnapshot);
6666
void OnWriteFinish(NTabletFlatExecutor::TTransactionContext& txc, const std::vector<TInsertWriteId>& insertWriteIds, const bool ephemeralFlag);
6767
void CommitOnExecute(TColumnShard& owner, NTabletFlatExecutor::TTransactionContext& txc, const NOlap::TSnapshot& snapshot) const;
6868
void CommitOnComplete(TColumnShard& owner, const NOlap::TSnapshot& snapshot) const;

0 commit comments

Comments
 (0)