Skip to content

Commit be3a726

Browse files
GrigoriyPAGrigoriyPA
committed
Removed unnecessary fields and simplified create or replace
1 parent c70d179 commit be3a726

File tree

6 files changed

+65
-44
lines changed

6 files changed

+65
-44
lines changed

‎ydb/core/protos/flat_scheme_op.proto‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1786,6 +1786,7 @@ message TModifyScheme {
17861786
optional EOperationType OperationType = 2;
17871787
optional bool Internal = 36 [default = false]; // internal operations are not generated directly by the user
17881788
optional bool FailOnExist = 50 [default = false]; // as a replacement for TEvModifySchemeTransaction.FailOnExist
1789+
optional bool ReplaceIfExists = 89 [default = false];
17891790
optional bool AllowAccessToPrivatePaths = 53 [default = false];
17901791

17911792
optional TMkDir MkDir = 3;
@@ -2362,8 +2363,5 @@ message TStreamingQueryProperties {
23622363

23632364
message TStreamingQueryDescription {
23642365
optional string Name = 1;
2365-
optional NKikimrProto.TPathID PathId = 2;
2366-
optional uint64 Version = 3;
2367-
optional TStreamingQueryProperties Properties = 4;
2368-
optional bool ReplaceIfExists = 5;
2366+
optional TStreamingQueryProperties Properties = 2;
23692367
}

‎ydb/core/tx/schemeshard/schemeshard__operation.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1620,7 +1620,7 @@ TVector<ISubOperation::TPtr> TDefaultOperationFactory::MakeOperationParts(
16201620

16211621
// StreamingQuery
16221622
case NKikimrSchemeOp::EOperationType::ESchemeOpCreateStreamingQuery:
1623-
return {CreateNewStreamingQuery(op.NextPartId(), tx, context)};
1623+
return {CreateNewStreamingQuery(op.NextPartId(), tx)};
16241624
case NKikimrSchemeOp::EOperationType::ESchemeOpDropStreamingQuery:
16251625
return {CreateDropStreamingQuery(op.NextPartId(), tx)};
16261626
case NKikimrSchemeOp::EOperationType::ESchemeOpAlterStreamingQuery:

‎ydb/core/tx/schemeshard/schemeshard__operation_create_streaming_query.cpp‎

Lines changed: 58 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,9 @@ namespace {
1212

1313
class TPropose : public TSubOperationState {
1414
public:
15-
explicit TPropose(TOperationId id)
15+
TPropose(TOperationId id, bool replacePath)
1616
: OperationId(std::move(id))
17+
, ReplacePath(replacePath)
1718
{}
1819

1920
bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr& ev, TOperationContext& context) override {
@@ -33,7 +34,9 @@ class TPropose : public TSubOperationState {
3334
path->StepCreated = step;
3435
context.SS->PersistCreateStep(db, pathId, step);
3536

36-
context.SS->TabletCounters->Simple()[COUNTER_STREAMING_QUERY_COUNT].Add(1);
37+
if (!ReplacePath) {
38+
context.SS->TabletCounters->Simple()[COUNTER_STREAMING_QUERY_COUNT].Add(1);
39+
}
3740

3841
IncParentDirAlterVersionWithRepublish(OperationId, path, context);
3942
context.SS->ClearDescribePathCaches(pathPtr);
@@ -61,6 +64,7 @@ class TPropose : public TSubOperationState {
6164

6265
private:
6366
const TOperationId OperationId;
67+
const bool ReplacePath = false;
6468
};
6569

6670
class TCreateStreamingQuery : public TSubOperation {
@@ -82,29 +86,53 @@ class TCreateStreamingQuery : public TSubOperation {
8286
switch (state) {
8387
case TTxState::Waiting:
8488
case TTxState::Propose:
85-
return MakeHolder<TPropose>(OperationId);
89+
return MakeHolder<TPropose>(OperationId, ReplacePath);
8690
case TTxState::Done:
8791
return MakeHolder<TDone>(OperationId);
8892
default:
8993
return nullptr;
9094
}
9195
}
9296

93-
static bool IsDestinationPathValid(const THolder<TProposeResponse>& result, const TOperationContext& context, const TPath& dstPath, const TString& acl, bool acceptExisted) {
97+
ui64 GetAlterVersion(const TPath& dstPath, const TOperationContext& context) {
98+
ui64 alterVersion = 1;
99+
100+
if (Transaction.GetReplaceIfExists()) {
101+
ReplacePath = static_cast<bool>(dstPath.Check()
102+
.IsResolved()
103+
.NotUnderDeleting());
104+
105+
if (ReplacePath) {
106+
const auto& oldStreamingQueryInfo = context.SS->StreamingQueries.Value(dstPath->PathId, nullptr);
107+
Y_ABORT_UNLESS(oldStreamingQueryInfo);
108+
alterVersion = oldStreamingQueryInfo->AlterVersion + 1;
109+
}
110+
}
111+
112+
return alterVersion;
113+
}
114+
115+
bool IsDestinationPathValid(const THolder<TProposeResponse>& result, const TOperationContext& context, const TPath& dstPath, const TString& acl, bool acceptExisted) const {
94116
const auto checks = dstPath.Check();
95117

96118
checks.IsAtLocalSchemeShard();
97119

98120
if (dstPath.IsResolved()) {
99121
checks.IsResolved()
100-
.NotUnderDeleting()
101-
.FailOnExist(TPathElement::EPathType::EPathTypeStreamingQuery, acceptExisted);
122+
.NotUnderDeleting();
123+
124+
if (ReplacePath) {
125+
checks.NotUnderOperation()
126+
.FailOnWrongType(TPathElement::EPathType::EPathTypeStreamingQuery);
127+
} else {
128+
checks.FailOnExist(TPathElement::EPathType::EPathTypeStreamingQuery, acceptExisted);
129+
}
102130
} else {
103131
checks.NotEmpty()
104132
.NotResolved();
105133
}
106134

107-
if (checks) {
135+
if (!ReplacePath && checks) {
108136
checks.IsValidLeafName(context.UserToken.Get())
109137
.DepthLimit()
110138
.PathsLimit()
@@ -123,8 +151,11 @@ class TCreateStreamingQuery : public TSubOperation {
123151
return static_cast<bool>(checks);
124152
}
125153

126-
static void AddPathInSchemeShard(const THolder<TProposeResponse>& result, TPath& dstPath, const TString& owner) {
127-
dstPath.MaterializeLeaf(owner);
154+
void AddPathInSchemeShard(const THolder<TProposeResponse>& result, TPath& dstPath, const TString& owner) const {
155+
if (!ReplacePath) {
156+
dstPath.MaterializeLeaf(owner);
157+
}
158+
128159
result->SetPathId(dstPath.Base()->PathId.LocalPathId);
129160
}
130161

@@ -139,6 +170,16 @@ class TCreateStreamingQuery : public TSubOperation {
139170
return streamingQuery;
140171
}
141172

173+
void IncPathCounters(const TPath& parentPath, const TPath& dstPath, TOperationContext& context) const {
174+
IncParentDirAlterVersionWithRepublishSafeWithUndo(OperationId, dstPath, context.SS, context.OnComplete);
175+
176+
if (!ReplacePath) {
177+
dstPath.DomainInfo()->IncPathsInside(context.SS);
178+
}
179+
180+
IncAliveChildrenDirect(OperationId, parentPath, context);
181+
}
182+
142183
public:
143184
using TSubOperation::TSubOperation;
144185

@@ -156,12 +197,14 @@ class TCreateStreamingQuery : public TSubOperation {
156197
RETURN_RESULT_UNLESS(IsParentPathValid(result, parentPath, /* isCreate */ true));
157198

158199
TPath dstPath = parentPath.Child(name);
200+
const ui64 alterVersion = GetAlterVersion(dstPath, context);
201+
159202
const TString& acl = Transaction.GetModifyACL().GetDiffACL();
160203
RETURN_RESULT_UNLESS(IsDestinationPathValid(result, context, dstPath, acl, !Transaction.GetFailOnExist()));
161204
RETURN_RESULT_UNLESS(IsApplyIfChecksPassed(result, Transaction, context));
162205
RETURN_RESULT_UNLESS(IsDescriptionValid(result, streamingQueryDescription));
163206

164-
const auto streamingQueryInfo = CreateNewStreamingQuery(streamingQueryDescription, 1);
207+
const auto streamingQueryInfo = CreateNewStreamingQuery(streamingQueryDescription, alterVersion);
165208
Y_ABORT_UNLESS(streamingQueryInfo);
166209

167210
AddPathInSchemeShard(result, dstPath, owner);
@@ -172,11 +215,7 @@ class TCreateStreamingQuery : public TSubOperation {
172215
NIceDb::TNiceDb db(context.GetDB());
173216
AdvanceTransactionStateToPropose(OperationId, context, db);
174217
PersistStreamingQuery(OperationId, context, db, streamingQuery, streamingQueryInfo, acl);
175-
176-
IncParentDirAlterVersionWithRepublishSafeWithUndo(OperationId, dstPath, context.SS, context.OnComplete);
177-
178-
dstPath.DomainInfo()->IncPathsInside(context.SS);
179-
IncAliveChildrenDirect(OperationId, parentPath, context); // for correct discard of ChildrenExist prop
218+
IncPathCounters(parentPath, dstPath, context);
180219

181220
SetState(NextState());
182221
return result;
@@ -191,6 +230,9 @@ class TCreateStreamingQuery : public TSubOperation {
191230
LOG_N("TCreateStreamingQuery AbortUnsafe: opId# " << OperationId << ", txId# " << forceDropTxId);
192231
context.OnComplete.DoneOperation(OperationId);
193232
}
233+
234+
private:
235+
bool ReplacePath = false;
194236
};
195237

196238
using TTag = TSchemeTxTraits<NKikimrSchemeOp::EOperationType::ESchemeOpCreateStreamingQuery>;
@@ -215,27 +257,7 @@ bool SetName<NStreamingQuery::TTag>(NStreamingQuery::TTag, TTxTransaction& tx, c
215257

216258
} // namespace NOperation
217259

218-
ISubOperation::TPtr CreateNewStreamingQuery(TOperationId id, const TTxTransaction& tx, TOperationContext& context) {
219-
Y_ABORT_UNLESS(tx.GetOperationType() == NKikimrSchemeOp::ESchemeOpCreateStreamingQuery);
220-
221-
LOG_I("CreateNewStreamingQuery, opId# " << id << ", tx# " << tx.ShortDebugString());
222-
223-
const TPath parentPath = TPath::Resolve(tx.GetWorkingDir(), context.SS);
224-
if (const auto checks = NStreamingQuery::IsParentPathValid(parentPath, /* isCreate */ true); !checks) {
225-
return CreateReject(id, checks.GetStatus(), TStringBuilder() << "Invalid CreateStreamingQuery request: " << checks.GetError());
226-
}
227-
228-
if (const auto& operation = tx.GetCreateStreamingQuery(); operation.GetReplaceIfExists()) {
229-
const TPath dstPath = parentPath.Child(operation.GetName());
230-
const auto isAlreadyExists = dstPath.Check()
231-
.IsResolved()
232-
.NotUnderDeleting();
233-
234-
if (isAlreadyExists) {
235-
return CreateAlterStreamingQuery(id, tx);
236-
}
237-
}
238-
260+
ISubOperation::TPtr CreateNewStreamingQuery(TOperationId id, const TTxTransaction& tx) {
239261
return MakeSubOperation<NStreamingQuery::TCreateStreamingQuery>(id, tx);
240262
}
241263

‎ydb/core/tx/schemeshard/schemeshard__operation_part.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -733,7 +733,7 @@ ISubOperation::TPtr CreateDropSysView(TOperationId id, TTxState::ETxState state)
733733

734734
// Streaming Query
735735
// Create
736-
ISubOperation::TPtr CreateNewStreamingQuery(TOperationId id, const TTxTransaction& tx, TOperationContext& context);
736+
ISubOperation::TPtr CreateNewStreamingQuery(TOperationId id, const TTxTransaction& tx);
737737
ISubOperation::TPtr CreateNewStreamingQuery(TOperationId id, TTxState::ETxState state);
738738
// Alter
739739
ISubOperation::TPtr CreateAlterStreamingQuery(TOperationId id, const TTxTransaction& tx);

‎ydb/core/tx/schemeshard/schemeshard_impl.cpp‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1470,6 +1470,9 @@ bool TSchemeShard::CheckApplyIf(const NKikimrSchemeOp::TModifyScheme& scheme, TS
14701470
case NKikimrSchemeOp::EPathType::EPathTypeSysView:
14711471
actualVersion = pathVersion.GetSysViewVersion();
14721472
break;
1473+
case NKikimrSchemeOp::EPathType::EPathTypeStreamingQuery:
1474+
actualVersion = pathVersion.GetStreamingQueryVersion();
1475+
break;
14731476
default:
14741477
actualVersion = pathVersion.GetGeneralVersion();
14751478
break;

‎ydb/core/tx/schemeshard/schemeshard_path_describer.cpp‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1132,8 +1132,6 @@ void TPathDescriber::DescribeStreamingQuery(TPathId pathId, TPathElement::TPtr p
11321132

11331133
auto& entry = *Result->Record.MutablePathDescription()->MutableStreamingQueryDescription();
11341134
entry.SetName(pathEl->Name);
1135-
pathId.ToProto(entry.MutablePathId());
1136-
entry.SetVersion(streamingQueryInfo->AlterVersion);
11371135
*entry.MutableProperties() = streamingQueryInfo->Properties;
11381136
}
11391137

0 commit comments

Comments
 (0)