Skip to content

Commit f834783

Browse files
ivanmorozov333ivanmorozov333
authored
Merge cd4759f into 77a5f3a 
2 parents 77a5f3a + cd4759f commit f834783

26 files changed

+282
-130
lines changed

‎ydb/core/kqp/ut/common/kqp_ut_common.cpp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -123,6 +123,7 @@ TKikimrRunner::TKikimrRunner(const TKikimrSettings& settings) {
123123
appConfig.MutableTableServiceConfig()->SetEnableRowsDuplicationCheck(true);
124124
ServerSettings->SetAppConfig(appConfig);
125125
ServerSettings->SetFeatureFlags(settings.FeatureFlags);
126+
ServerSettings->FeatureFlags.SetEnableImmediateWritingOnBulkUpsert(true);
126127
ServerSettings->SetNodeCount(settings.NodeCount);
127128
ServerSettings->SetEnableKqpSpilling(enableSpilling);
128129
ServerSettings->SetEnableDataColumnForIndexTable(true);

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

Lines changed: 1 addition & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,7 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
8080

8181
{
8282
auto alterQuery = TStringBuilder() <<
83-
"ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=UPSERT_OPTIONS, SCHEME_NEED_ACTUALIZATION=`true`, EXTERNAL_GUARANTEE_EXCLUSIVE_PK=`true`);";
83+
"ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=UPSERT_OPTIONS, SCHEME_NEED_ACTUALIZATION=`true`);";
8484
auto session = tableClient.CreateSession().GetValueSync().GetSession();
8585
auto alterResult = session.ExecuteSchemeQuery(alterQuery).GetValueSync();
8686
UNIT_ASSERT_VALUES_EQUAL_C(alterResult.GetStatus(), NYdb::EStatus::SUCCESS, alterResult.GetIssues().ToString());
@@ -336,13 +336,6 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
336336
auto alterResult = session.ExecuteSchemeQuery(alterQuery).GetValueSync();
337337
UNIT_ASSERT_VALUES_EQUAL_C(alterResult.GetStatus(), NYdb::EStatus::SUCCESS, alterResult.GetIssues().ToString());
338338
}
339-
{
340-
auto alterQuery = TStringBuilder() <<
341-
"ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=UPSERT_OPTIONS, EXTERNAL_GUARANTEE_EXCLUSIVE_PK=`true`);";
342-
auto session = tableClient.CreateSession().GetValueSync().GetSession();
343-
auto alterResult = session.ExecuteSchemeQuery(alterQuery).GetValueSync();
344-
UNIT_ASSERT_VALUES_EQUAL_C(alterResult.GetStatus(), NYdb::EStatus::SUCCESS, alterResult.GetIssues().ToString());
345-
}
346339

347340
std::vector<TString> uids;
348341
std::vector<TString> resourceIds;

‎ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8172,7 +8172,7 @@ Y_UNIT_TEST_SUITE(KqpOlapScheme) {
81728172
csController->WaitCompactions(TDuration::Seconds(5));
81738173
}
81748174

8175-
testHelper.ReadData("SELECT value FROM `/Root/ColumnTableTest`", "[[#];[#];[[42u]];[[43u]]]");
8175+
testHelper.ReadData("SELECT value FROM `/Root/ColumnTableTest` ORDER BY value", "[[#];[#];[[42u]];[[43u]]]");
81768176
}
81778177

81788178
Y_UNIT_TEST(DropThenAddColumn) {

‎ydb/core/protos/data_events.proto‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,7 @@ message TEvWrite {
9595
repeated uint32 ColumnIds = 3 [packed = true];
9696
optional uint64 PayloadIndex = 4;
9797
optional EDataFormat PayloadFormat = 5;
98+
optional string PayloadSchema = 6;
9899
}
99100

100101
// Transaction operations

‎ydb/core/protos/feature_flags.proto‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -162,4 +162,5 @@ message TFeatureFlags {
162162
optional bool EnableExternalDataSourcesOnServerless = 143 [default = true];
163163
optional bool EnableSparsedColumns = 144 [default = false];
164164
optional bool EnableParameterizedDecimal = 145 [default = false];
165+
optional bool EnableImmediateWritingOnBulkUpsert = 146 [default = false];
165166
}

‎ydb/core/tx/columnshard/blobs_action/transaction/tx_write.cpp‎

Lines changed: 27 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -6,28 +6,11 @@
66
namespace NKikimr::NColumnShard {
77

88
bool TTxWrite::InsertOneBlob(TTransactionContext& txc, const NOlap::TWideSerializedBatch& batch, const TInsertWriteId writeId) {
9-
NKikimrTxColumnShard::TLogicalMetadata meta;
10-
meta.SetNumRows(batch->GetRowsCount());
11-
meta.SetRawBytes(batch->GetRawBytes());
12-
meta.SetDirtyWriteTimeSeconds(batch.GetStartInstant().Seconds());
13-
meta.SetSpecialKeysRawData(batch->GetSpecialKeysFullSafe());
14-
meta.SetSpecialKeysPayloadData(batch->GetSpecialKeysPayloadSafe());
15-
16-
const auto& blobRange = batch.GetRange();
17-
Y_ABORT_UNLESS(blobRange.GetBlobId().IsValid());
9+
auto userData = batch.BuildInsertionUserData(*Self);
10+
NOlap::TInsertedData insertData(writeId, userData);
1811

19-
// First write wins
2012
TBlobGroupSelector dsGroupSelector(Self->Info());
2113
NOlap::TDbWrapper dbTable(txc.DB, &dsGroupSelector);
22-
23-
const auto& writeMeta = batch.GetAggregation().GetWriteMeta();
24-
meta.SetModificationType(TEnumOperator<NEvWrite::EModificationType>::SerializeToProto(writeMeta.GetModificationType()));
25-
*meta.MutableSchemaSubset() = batch.GetAggregation().GetSchemaSubset().SerializeToProto();
26-
auto schemeVersion = batch.GetAggregation().GetSchemaVersion();
27-
auto tableSchema = Self->TablesManager.GetPrimaryIndex()->GetVersionedIndex().GetSchemaVerified(schemeVersion);
28-
29-
auto userData = std::make_shared<NOlap::TUserData>(writeMeta.GetTableId(), blobRange, meta, tableSchema->GetVersion(), batch->GetData());
30-
NOlap::TInsertedData insertData(writeId, userData);
3114
bool ok = Self->InsertTable->Insert(dbTable, std::move(insertData));
3215
if (ok) {
3316
Self->UpdateInsertTableCounters();
@@ -36,6 +19,18 @@ bool TTxWrite::InsertOneBlob(TTransactionContext& txc, const NOlap::TWideSeriali
3619
return false;
3720
}
3821

22+
bool TTxWrite::CommitOneBlob(TTransactionContext& txc, const NOlap::TWideSerializedBatch& batch, const TInsertWriteId writeId) {
23+
auto userData = batch.BuildInsertionUserData(*Self);
24+
TBlobGroupSelector dsGroupSelector(Self->Info());
25+
NOlap::TDbWrapper dbTable(txc.DB, &dsGroupSelector);
26+
NOlap::TCommittedData commitData(userData, Self->GetLastPlannedSnapshot(), Self->Generation(), writeId);
27+
if (Self->TablesManager.HasTable(userData->GetPathId())) {
28+
Self->InsertTable->CommitEphemeral(dbTable, std::move(commitData));
29+
}
30+
Self->UpdateInsertTableCounters();
31+
return true;
32+
}
33+
3934
bool TTxWrite::Execute(TTransactionContext& txc, const TActorContext&) {
4035
TMemoryProfileGuard mpg("TTxWrite::Execute");
4136
NActors::TLogContextGuard logGuard =
@@ -65,10 +60,17 @@ bool TTxWrite::Execute(TTransactionContext& txc, const TActorContext&) {
6560
operation = Self->OperationsManager->GetOperationVerified((TOperationWriteId)writeMeta.GetWriteId());
6661
Y_ABORT_UNLESS(operation->GetStatus() == EOperationStatus::Started);
6762
for (auto&& i : aggr->GetSplittedBlobs()) {
68-
const TInsertWriteId insertWriteId = Self->InsertTable->BuildNextWriteId(txc);
69-
aggr->AddInsertWriteId(insertWriteId);
70-
AFL_VERIFY(InsertOneBlob(txc, i, insertWriteId))("write_id", writeMeta.GetWriteId())("insert_write_id", insertWriteId)(
71-
"size", aggr->GetSplittedBlobs().size());
63+
if (operation->GetBehaviour() == EOperationBehaviour::NoTxWrite) {
64+
static TAtomicCounter Counter = 0;
65+
const TInsertWriteId insertWriteId = (TInsertWriteId)Counter.Inc();
66+
AFL_VERIFY(CommitOneBlob(txc, i, insertWriteId))("write_id", writeMeta.GetWriteId())("insert_write_id", insertWriteId)(
67+
"size", aggr->GetSplittedBlobs().size());
68+
} else {
69+
const TInsertWriteId insertWriteId = Self->InsertTable->BuildNextWriteId(txc);
70+
aggr->AddInsertWriteId(insertWriteId);
71+
AFL_VERIFY(InsertOneBlob(txc, i, insertWriteId))("write_id", writeMeta.GetWriteId())("insert_write_id", insertWriteId)(
72+
"size", aggr->GetSplittedBlobs().size());
73+
}
7274
}
7375
}
7476
}
@@ -92,8 +94,6 @@ bool TTxWrite::Execute(TTransactionContext& txc, const TActorContext&) {
9294
if (operation->GetBehaviour() == EOperationBehaviour::NoTxWrite) {
9395
auto ev = NEvents::TDataEvents::TEvWriteResult::BuildCompleted(Self->TabletID());
9496
Results.emplace_back(std::move(ev), writeMeta.GetSource(), operation->GetCookie());
95-
Self->OperationsManager->AddTemporaryTxLink(operation->GetLockId());
96-
Self->OperationsManager->CommitTransactionOnExecute(*Self, operation->GetLockId(), txc, Self->GetLastTxSnapshot());
9797
} else if (operation->GetBehaviour() == EOperationBehaviour::InTxWrite) {
9898
NKikimrTxColumnShard::TCommitWriteTxBody proto;
9999
proto.SetLockId(operation->GetLockId());
@@ -156,13 +156,15 @@ void TTxWrite::Complete(const TActorContext& ctx) {
156156
Self->GetOperationsManager().AddEventForLock(*Self, op->GetLockId(), evWrite);
157157
}
158158
if (op->GetBehaviour() == EOperationBehaviour::NoTxWrite) {
159+
Self->OperationsManager->AddTemporaryTxLink(op->GetLockId());
159160
Self->OperationsManager->CommitTransactionOnComplete(*Self, op->GetLockId(), Self->GetLastTxSnapshot());
160161
}
161162
}
162163
Self->Counters.GetCSCounters().OnWriteTxComplete(now - writeMeta.GetWriteStartInstant());
163164
Self->Counters.GetCSCounters().OnSuccessWriteResponse();
164165
}
165166
Self->Counters.GetTabletCounters()->IncCounter(COUNTER_IMMEDIATE_TX_COMPLETED);
167+
Self->SetupIndexation();
166168
}
167169

168170
} // namespace NKikimr::NColumnShard

‎ydb/core/tx/columnshard/blobs_action/transaction/tx_write.h‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,9 @@ class TTxWrite : public NTabletFlatExecutor::TTransactionBase<TColumnShard> {
2020
TEvPrivate::TEvWriteBlobsResult::TPtr PutBlobResult;
2121
const ui32 TabletTxNo;
2222

23+
bool CommitOneBlob(TTransactionContext& txc, const NOlap::TWideSerializedBatch& batch, const TInsertWriteId writeId);
24+
bool InsertOneBlob(TTransactionContext& txc, const NOlap::TWideSerializedBatch& batch, const TInsertWriteId writeId);
25+
2326
class TReplyInfo {
2427
private:
2528
std::unique_ptr<NActors::IEventBase> Event;
@@ -43,8 +46,6 @@ class TTxWrite : public NTabletFlatExecutor::TTransactionBase<TColumnShard> {
4346
std::vector<std::shared_ptr<TTxController::ITransactionOperator>> ResultOperators;
4447

4548

46-
bool InsertOneBlob(TTransactionContext& txc, const NOlap::TWideSerializedBatch& batch, const TInsertWriteId writeId);
47-
4849
TStringBuilder TxPrefix() const {
4950
return TStringBuilder() << "TxWrite[" << ToString(TabletTxNo) << "] ";
5051
}

‎ydb/core/tx/columnshard/columnshard_impl.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -544,6 +544,10 @@ class TColumnShard
544544
public:
545545
ui64 TabletTxCounter = 0;
546546

547+
const TTablesManager& GetTablesManager() const {
548+
return TablesManager;
549+
}
550+
547551
bool HasLongTxWrites(const TInsertWriteId insertWriteId) const {
548552
return LongTxWrites.contains(insertWriteId);
549553
}

‎ydb/core/tx/columnshard/engines/insert_table/committed.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,10 +25,10 @@ class TCommittedData: public TUserDataContainer {
2525
, DedupId(dedupId) {
2626
}
2727

28-
TCommittedData(const std::shared_ptr<TUserData>& userData, const TSnapshot& ss, const TInsertWriteId insertWriteId)
28+
TCommittedData(const std::shared_ptr<TUserData>& userData, const TSnapshot& ss, const ui64 generation, const TInsertWriteId ephemeralWriteId)
2929
: TBase(userData)
3030
, Snapshot(ss)
31-
, DedupId(ToString(ss.GetPlanStep()) + ":" + ToString((ui64)insertWriteId)) {
31+
, DedupId(ToString(generation) + ":" + ToString(ephemeralWriteId)) {
3232
}
3333

3434
void SetRemove() {

‎ydb/core/tx/columnshard/engines/insert_table/insert_table.cpp‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,22 @@ TInsertionSummary::TCounters TInsertTable::Commit(
5757
return counters;
5858
}
5959

60+
TInsertionSummary::TCounters TInsertTable::CommitEphemeral(IDbWrapper& dbTable, TCommittedData&& data) {
61+
TInsertionSummary::TCounters counters;
62+
counters.Rows += data.GetMeta().GetNumRows();
63+
counters.RawBytes += data.GetMeta().GetRawBytes();
64+
counters.Bytes += data.BlobSize();
65+
66+
AddBlobLink(data.GetBlobRange().BlobId);
67+
const ui64 pathId = data.GetPathId();
68+
auto& pathInfo = Summary.GetPathInfo(pathId);
69+
AFL_TRACE(NKikimrServices::TX_COLUMNSHARD)("event", "commit_insertion")("path_id", pathId)("blob_range", data.GetBlobRange().ToString());
70+
dbTable.Commit(data);
71+
pathInfo.AddCommitted(std::move(data));
72+
73+
return counters;
74+
}
75+
6076
void TInsertTable::Abort(IDbWrapper& dbTable, const THashSet<TInsertWriteId>& writeIds) {
6177
Y_ABORT_UNLESS(!writeIds.empty());
6278

0 commit comments

Comments
 (0)