Skip to content

WIP #18

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 5 commits into
base: main
Choose a base branch
from
Open

WIP #18

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d57cb16
fix: index schema sync race
Enjection Nov 12, 2025
8db836c
WIP
Enjection Nov 12, 2025
bbe06e6
Implement operation order testing framework for SchemeShard
claude Nov 12, 2025
ae8cf33
Add SchemeShard runtime integration for automatic operation batching
claude Nov 12, 2025
044fff2
Update progress: Phase 2 (SchemeShard Integration) completed
claude Nov 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
885 changes: 885 additions & 0 deletions PARALLEL_OPERATIONS_TESTING_PLAN.md
View file
Open in desktop

Large diffs are not rendered by default.

52 changes: 38 additions & 14 deletions ydb/core/tx/schemeshard/schemeshard__operation_common_cdc_stream.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,18 +118,28 @@ void SyncImplTableVersion(
{
Y_ABORT_UNLESS(context.SS->Tables.contains(versionCtx.GrandParentPathId));
auto parentTable = context.SS->Tables.at(versionCtx.GrandParentPathId);

ui64 currentImplVersion = table->AlterVersion;
ui64 currentParentVersion = parentTable->AlterVersion;

if (currentImplVersion <= currentParentVersion) {
table->AlterVersion = currentParentVersion;
// Also check the index entity version to avoid race conditions
// Use the maximum of parent version and index entity version
ui64 targetVersion = currentParentVersion;
if (context.SS->Indexes.contains(versionCtx.ParentPathId)) {
auto index = context.SS->Indexes.at(versionCtx.ParentPathId);
// This handles cases where parent operation has already synced entity
targetVersion = Max(currentParentVersion, index->AlterVersion);
}

if (currentImplVersion <= targetVersion) {
table->AlterVersion = targetVersion;
LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
"Synchronized index impl table version to parent table"
"Synchronized index impl table version"
<< ", implTablePathId: " << versionCtx.PathId
<< ", parentTablePathId: " << versionCtx.GrandParentPathId
<< ", oldImplVersion: " << currentImplVersion
<< ", parentVersion: " << currentParentVersion
<< ", targetVersion: " << targetVersion
<< ", newImplVersion: " << table->AlterVersion
<< ", at schemeshard: " << context.SS->SelfTabletId());
} else {
Expand All @@ -139,6 +149,7 @@ void SyncImplTableVersion(
<< ", implTablePathId: " << versionCtx.PathId
<< ", implVersion: " << currentImplVersion
<< ", parentVersion: " << currentParentVersion
<< ", targetVersion: " << targetVersion
<< ", newImplVersion: " << table->AlterVersion
<< ", at schemeshard: " << context.SS->SelfTabletId());
}
Expand All @@ -156,19 +167,32 @@ void SyncIndexEntityVersion(
}

auto index = context.SS->Indexes.at(indexPathId);
index->AlterVersion = targetVersion;
ui64 oldIndexVersion = index->AlterVersion;

context.SS->PersistTableIndexAlterVersion(db, indexPathId, index);
// Only update if we're increasing the version (prevent downgrade due to race conditions)
if (targetVersion > oldIndexVersion) {
index->AlterVersion = targetVersion;

auto indexPath = context.SS->PathsById.at(indexPathId);
context.SS->ClearDescribePathCaches(indexPath);
context.OnComplete.PublishToSchemeBoard(operationId, indexPathId);
context.SS->PersistTableIndexAlterVersion(db, indexPathId, index);

LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
"Synced index entity version"
<< ", indexPathId: " << indexPathId
<< ", newVersion: " << index->AlterVersion
<< ", at schemeshard: " << context.SS->SelfTabletId());
auto indexPath = context.SS->PathsById.at(indexPathId);
context.SS->ClearDescribePathCaches(indexPath);
context.OnComplete.PublishToSchemeBoard(operationId, indexPathId);

LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
"Synced index entity version"
<< ", indexPathId: " << indexPathId
<< ", oldVersion: " << oldIndexVersion
<< ", newVersion: " << index->AlterVersion
<< ", at schemeshard: " << context.SS->SelfTabletId());
} else {
LOG_DEBUG_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD,
"Skipping index entity sync - already at higher version"
<< ", indexPathId: " << indexPathId
<< ", currentVersion: " << oldIndexVersion
<< ", targetVersion: " << targetVersion
<< ", at schemeshard: " << context.SS->SelfTabletId());
}
}

void SyncChildIndexes(
Expand Down
Loading