Skip to content

Commit 86695c8

Browse files
vlad-gogovvlad-gogov
authored
Merge 990dc4d into 5b3cd09 
2 parents 5b3cd09 + 990dc4d commit 86695c8

File tree

6 files changed

+175
-18
lines changed

6 files changed

+175
-18
lines changed

‎.github/config/muted_ya.txt‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ ydb/core/kqp/ut/query KqpQuery.QueryTimeout
1919
ydb/core/kqp/ut/scan KqpRequestContext.TraceIdInErrorMessage
2020
ydb/core/kqp/ut/scheme [*/*]*
2121
ydb/core/kqp/ut/scheme KqpOlapScheme.TenThousandColumns
22-
ydb/core/kqp/ut/scheme KqpOlapScheme.DropColumnAfterInsert
2322
ydb/core/kqp/ut/scheme KqpScheme.AlterAsyncReplication
2423
ydb/core/kqp/ut/scheme KqpScheme.QueryWithAlter
2524
ydb/core/kqp/ut/service [*/*]*

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

Lines changed: 140 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8354,13 +8354,152 @@ Y_UNIT_TEST_SUITE(KqpOlapScheme) {
83548354
testHelper.BulkUpsert(testTable, batch);
83558355

83568356
auto alterQueryAdd = TStringBuilder() << "ALTER TABLE `" << testTable.GetName() << "` DROP COLUMN int_column;";
8357-
Cerr << alterQueryAdd << Endl;
83588357
auto alterAddResult = testHelper.GetSession().ExecuteSchemeQuery(alterQueryAdd).GetValueSync();
83598358
UNIT_ASSERT_VALUES_EQUAL_C(alterAddResult.GetStatus(), EStatus::SUCCESS, alterAddResult.GetIssues().ToString());
83608359

83618360
csController->EnableBackground(NYDBTest::ICSController::EBackground::Indexation);
83628361
csController->WaitIndexation(TDuration::Seconds(5));
83638362
}
8363+
8364+
void TestInsertAddInsertDrop(
8365+
bool autoIndexation, bool indexationAfterInsertAddColumn, bool indexationAfterInsertDropColumn, bool indexationInEnd) {
8366+
using namespace NArrow;
8367+
8368+
auto csController = NYDBTest::TControllers::RegisterCSControllerGuard<NOlap::TWaitCompactionController>();
8369+
if (!autoIndexation) {
8370+
csController->DisableBackground(NYDBTest::ICSController::EBackground::Indexation);
8371+
}
8372+
8373+
TKikimrSettings runnerSettings;
8374+
runnerSettings.WithSampleTables = false;
8375+
TTestHelper testHelper(runnerSettings);
8376+
8377+
TVector<TTestHelper::TColumnSchema> schema = {
8378+
TTestHelper::TColumnSchema().SetName("id").SetType(NScheme::NTypeIds::Uint64).SetNullable(false),
8379+
TTestHelper::TColumnSchema().SetName("int_column").SetType(NScheme::NTypeIds::Int32).SetNullable(true)
8380+
};
8381+
8382+
TTestHelper::TColumnTable testTable;
8383+
testTable.SetName("/Root/ColumnTableTest").SetPrimaryKey({ "id" }).SetSchema(schema);
8384+
testHelper.CreateTable(testTable);
8385+
8386+
TVector<NConstruction::IArrayBuilder::TPtr> dataBuilders;
8387+
dataBuilders.push_back(
8388+
NConstruction::TSimpleArrayConstructor<NConstruction::TIntSeqFiller<arrow::UInt64Type>>::BuildNotNullable("id", false));
8389+
dataBuilders.push_back(
8390+
std::make_shared<NConstruction::TSimpleArrayConstructor<NConstruction::TIntSeqFiller<arrow::Int32Type>>>("int_column"));
8391+
auto batch = NConstruction::TRecordBatchConstructor(dataBuilders).BuildBatch(100);
8392+
8393+
for (ui32 i = 0; i < 5; i++) {
8394+
testHelper.BulkUpsert(testTable, batch);
8395+
auto alterQueryAdd = TStringBuilder() << "ALTER TABLE `" << testTable.GetName() << "` ADD COLUMN column" << i << " Uint64;";
8396+
auto alterAddResult = testHelper.GetSession().ExecuteSchemeQuery(alterQueryAdd).GetValueSync();
8397+
UNIT_ASSERT_VALUES_EQUAL_C(alterAddResult.GetStatus(), EStatus::SUCCESS, alterAddResult.GetIssues().ToString());
8398+
8399+
if (!autoIndexation && indexationAfterInsertAddColumn) {
8400+
csController->EnableBackground(NYDBTest::ICSController::EBackground::Indexation);
8401+
csController->WaitIndexation(TDuration::Seconds(5));
8402+
csController->DisableBackground(NYDBTest::ICSController::EBackground::Indexation);
8403+
}
8404+
8405+
testHelper.BulkUpsert(testTable, batch);
8406+
auto alterQueryDrop = TStringBuilder() << "ALTER TABLE `" << testTable.GetName() << "` DROP COLUMN column" << i << ";";
8407+
auto alterDropResult = testHelper.GetSession().ExecuteSchemeQuery(alterQueryDrop).GetValueSync();
8408+
UNIT_ASSERT_VALUES_EQUAL_C(alterDropResult.GetStatus(), EStatus::SUCCESS, alterDropResult.GetIssues().ToString());
8409+
8410+
if (!autoIndexation && indexationAfterInsertDropColumn) {
8411+
csController->EnableBackground(NYDBTest::ICSController::EBackground::Indexation);
8412+
csController->WaitIndexation(TDuration::Seconds(5));
8413+
csController->DisableBackground(NYDBTest::ICSController::EBackground::Indexation);
8414+
}
8415+
}
8416+
8417+
if (!autoIndexation && indexationInEnd) {
8418+
csController->EnableBackground(NYDBTest::ICSController::EBackground::Indexation);
8419+
csController->WaitIndexation(TDuration::Seconds(5));
8420+
}
8421+
}
8422+
8423+
Y_UNIT_TEST(InsertAddInsertDrop) {
8424+
TestInsertAddInsertDrop(true, false, false, false);
8425+
for (i32 i = 0; i < 8; i++) {
8426+
TestInsertAddInsertDrop(false, i & 1, i & 2, i & 3);
8427+
}
8428+
}
8429+
8430+
Y_UNIT_TEST(DropTableAfterInsert) {
8431+
using namespace NArrow;
8432+
8433+
auto csController = NYDBTest::TControllers::RegisterCSControllerGuard<NOlap::TWaitCompactionController>();
8434+
csController->DisableBackground(NYDBTest::ICSController::EBackground::Indexation);
8435+
8436+
TKikimrSettings runnerSettings;
8437+
runnerSettings.WithSampleTables = false;
8438+
TTestHelper testHelper(runnerSettings);
8439+
8440+
TVector<TTestHelper::TColumnSchema> schema = {
8441+
TTestHelper::TColumnSchema().SetName("id").SetType(NScheme::NTypeIds::Uint64).SetNullable(false),
8442+
TTestHelper::TColumnSchema().SetName("int_column").SetType(NScheme::NTypeIds::Int32).SetNullable(true)
8443+
};
8444+
8445+
TTestHelper::TColumnTable testTable;
8446+
testTable.SetName("/Root/ColumnTableTest").SetPrimaryKey({ "id" }).SetSchema(schema);
8447+
testHelper.CreateTable(testTable);
8448+
8449+
TVector<NConstruction::IArrayBuilder::TPtr> dataBuilders;
8450+
dataBuilders.push_back(
8451+
NConstruction::TSimpleArrayConstructor<NConstruction::TIntSeqFiller<arrow::UInt64Type>>::BuildNotNullable("id", false));
8452+
dataBuilders.push_back(
8453+
std::make_shared<NConstruction::TSimpleArrayConstructor<NConstruction::TIntSeqFiller<arrow::Int32Type>>>("int_column"));
8454+
auto batch = NConstruction::TRecordBatchConstructor(dataBuilders).BuildBatch(100);
8455+
8456+
testHelper.BulkUpsert(testTable, batch);
8457+
8458+
auto alterQueryDrop = TStringBuilder() << "DROP TABLE `" << testTable.GetName() << "`;";
8459+
auto alterDropResult = testHelper.GetSession().ExecuteSchemeQuery(alterQueryDrop).GetValueSync();
8460+
UNIT_ASSERT_VALUES_EQUAL_C(alterDropResult.GetStatus(), EStatus::SUCCESS, alterDropResult.GetIssues().ToString());
8461+
8462+
csController->EnableBackground(NYDBTest::ICSController::EBackground::Indexation);
8463+
csController->WaitIndexation(TDuration::Seconds(5));
8464+
}
8465+
8466+
Y_UNIT_TEST(InsertDropAddColumn) {
8467+
using namespace NArrow;
8468+
8469+
auto csController = NYDBTest::TControllers::RegisterCSControllerGuard<NOlap::TWaitCompactionController>();
8470+
csController->DisableBackground(NYDBTest::ICSController::EBackground::Indexation);
8471+
8472+
TKikimrSettings runnerSettings;
8473+
runnerSettings.WithSampleTables = false;
8474+
TTestHelper testHelper(runnerSettings);
8475+
8476+
TVector<TTestHelper::TColumnSchema> schema = {
8477+
TTestHelper::TColumnSchema().SetName("id").SetType(NScheme::NTypeIds::Uint64).SetNullable(false),
8478+
TTestHelper::TColumnSchema().SetName("int_column").SetType(NScheme::NTypeIds::Int32).SetNullable(true)
8479+
};
8480+
8481+
TTestHelper::TColumnTable testTable;
8482+
testTable.SetName("/Root/ColumnTableTest").SetPrimaryKey({ "id" }).SetSchema(schema);
8483+
testHelper.CreateTable(testTable);
8484+
8485+
TVector<NConstruction::IArrayBuilder::TPtr> dataBuilders;
8486+
dataBuilders.push_back(
8487+
NConstruction::TSimpleArrayConstructor<NConstruction::TIntSeqFiller<arrow::UInt64Type>>::BuildNotNullable("id", false));
8488+
dataBuilders.push_back(
8489+
std::make_shared<NConstruction::TSimpleArrayConstructor<NConstruction::TIntSeqFiller<arrow::Int32Type>>>("int_column"));
8490+
auto batch = NConstruction::TRecordBatchConstructor(dataBuilders).BuildBatch(100);
8491+
8492+
testHelper.BulkUpsert(testTable, batch);
8493+
8494+
auto alterQueryDrop = TStringBuilder() << "ALTER TABLE `" << testTable.GetName() << "` DROP COLUMN int_column;";
8495+
auto alterDropResult = testHelper.GetSession().ExecuteSchemeQuery(alterQueryDrop).GetValueSync();
8496+
8497+
auto alterQueryAdd = TStringBuilder() << "ALTER TABLE `" << testTable.GetName() << "` ADD COLUMN int_column Int32;";
8498+
auto alterAddResult = testHelper.GetSession().ExecuteSchemeQuery(alterQueryAdd).GetValueSync();
8499+
8500+
csController->EnableBackground(NYDBTest::ICSController::EBackground::Indexation);
8501+
csController->WaitIndexation(TDuration::Seconds(5));
8502+
}
83648503
}
83658504

83668505
Y_UNIT_TEST_SUITE(KqpOlapTypes) {

‎ydb/core/tx/columnshard/engines/changes/indexation.cpp‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -104,14 +104,19 @@ class TPathFieldsInfo {
104104
return;
105105
}
106106
auto blobSchema = context.SchemaVersions.GetSchemaVerified(data.GetSchemaVersion());
107+
std::set<ui32> columnIdsToDelete = blobSchema->GetColumnIdsToDelete(ResultSchema);
107108
if (!Schemas.contains(data.GetSchemaVersion())) {
108109
Schemas.emplace(data.GetSchemaVersion(), blobSchema);
109110
}
110111
std::vector<ui32> filteredIds = data.GetMeta().GetSchemaSubset().Apply(blobSchema->GetIndexInfo().GetColumnIds(false));
111112
if (data.GetMeta().GetModificationType() == NEvWrite::EModificationType::Delete) {
112113
filteredIds.emplace_back((ui32)IIndexInfo::ESpecialColumn::DELETE_FLAG);
113114
}
114-
UsageColumnIds.insert(filteredIds.begin(), filteredIds.end());
115+
for (const auto& filteredId : filteredIds) {
116+
if (!columnIdsToDelete.contains(filteredId)) {
117+
UsageColumnIds.insert(filteredId);
118+
}
119+
}
115120
}
116121
};
117122

@@ -242,7 +247,10 @@ TConclusionStatus TInsertColumnEngineChanges::DoConstructBlobs(TConstructionCont
242247
auto batchSchema =
243248
std::make_shared<arrow::Schema>(inserted.GetMeta().GetSchemaSubset().Apply(blobSchema->GetIndexInfo().ArrowSchema()->fields()));
244249
batch = std::make_shared<NArrow::TGeneralContainer>(NArrow::DeserializeBatch(blobData, batchSchema));
245-
blobSchema->AdaptBatchToSchema(*batch, resultSchema);
250+
std::set<ui32> columnIdsToDelete = blobSchema->GetColumnIdsToDelete(resultSchema);
251+
if (!columnIdsToDelete.empty()) {
252+
batch->DeleteFieldsByIndex(blobSchema->ConvertColumnIdsToIndexes(columnIdsToDelete));
253+
}
246254
}
247255
IIndexInfo::AddSnapshotColumns(*batch, inserted.GetSnapshot(), (ui64)inserted.GetInsertWriteId());
248256

‎ydb/core/tx/columnshard/engines/reader/plain_reader/iterator/source.cpp‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -224,7 +224,10 @@ void TCommittedDataSource::DoAssembleColumns(const std::shared_ptr<TColumnsSet>&
224224
auto rBatch = NArrow::DeserializeBatch(bData, std::make_shared<arrow::Schema>(CommittedBlob.GetSchemaSubset().Apply(schema->fields())));
225225
AFL_VERIFY(rBatch)("schema", schema->ToString());
226226
auto batch = std::make_shared<NArrow::TGeneralContainer>(rBatch);
227-
batchSchema->AdaptBatchToSchema(*batch, resultSchema);
227+
std::set<ui32> columnIdsToDelete = batchSchema->GetColumnIdsToDelete(resultSchema);
228+
if (!columnIdsToDelete.empty()) {
229+
batch->DeleteFieldsByIndex(batchSchema->ConvertColumnIdsToIndexes(columnIdsToDelete));
230+
}
228231
TSnapshot ss = TSnapshot::Zero();
229232
if (CommittedBlob.IsCommitted()) {
230233
ss = CommittedBlob.GetCommittedSnapshotVerified();

‎ydb/core/tx/columnshard/engines/scheme/versions/abstract_scheme.cpp‎

Lines changed: 19 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -133,20 +133,27 @@ TConclusion<std::shared_ptr<arrow::RecordBatch>> ISnapshotSchema::PrepareForModi
133133
return batch;
134134
}
135135

136-
void ISnapshotSchema::AdaptBatchToSchema(NArrow::TGeneralContainer& batch, const ISnapshotSchema::TPtr& targetSchema) const {
137-
if (targetSchema->GetVersion() != GetVersion()) {
138-
std::vector<ui32> columnIdxToDelete;
139-
for (size_t columnIdx = 0; columnIdx < batch.GetSchema()->GetFields().size(); ++columnIdx) {
140-
const std::optional<ui32> targetColumnId = targetSchema->GetColumnIdOptional(batch.GetSchema()->field(columnIdx)->name());
141-
const ui32 batchColumnId = GetColumnIdVerified(GetFieldByIndex(columnIdx)->name());
142-
if (!targetColumnId || *targetColumnId != batchColumnId) {
143-
columnIdxToDelete.emplace_back(columnIdx);
144-
}
145-
}
146-
if (!columnIdxToDelete.empty()) {
147-
batch.DeleteFieldsByIndex(columnIdxToDelete);
136+
std::set<ui32> ISnapshotSchema::GetColumnIdsToDelete(const ISnapshotSchema::TPtr& targetSchema) const {
137+
if (targetSchema->GetVersion() == GetVersion()) {
138+
return {};
139+
}
140+
std::set<ui32> columnIdxsToDelete;
141+
for (const auto& columnIdx : GetColumnIds()) {
142+
const std::optional<ui32> targetColumnId = targetSchema->GetColumnIdOptional(GetFieldByColumnIdOptional(columnIdx)->name());
143+
if (!targetColumnId || *targetColumnId != columnIdx) {
144+
columnIdxsToDelete.emplace(columnIdx);
148145
}
149146
}
147+
return columnIdxsToDelete;
148+
}
149+
150+
std::vector<ui32> ISnapshotSchema::ConvertColumnIdsToIndexes(const std::set<ui32>& idxs) const {
151+
std::vector<ui32> columnIndexes;
152+
for (const auto& id : idxs) {
153+
AFL_VERIFY(HasColumnId(id));
154+
columnIndexes.emplace_back(GetFieldIndex(id));
155+
}
156+
return columnIndexes;
150157
}
151158

152159
ui32 ISnapshotSchema::GetColumnId(const std::string& columnName) const {

‎ydb/core/tx/columnshard/engines/scheme/versions/abstract_scheme.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,8 @@ class ISnapshotSchema {
7777
const ISnapshotSchema& dataSchema, const std::shared_ptr<NArrow::TGeneralContainer>& batch, const std::set<ui32>& restoreColumnIds) const;
7878
[[nodiscard]] TConclusion<std::shared_ptr<arrow::RecordBatch>> PrepareForModification(
7979
const std::shared_ptr<arrow::RecordBatch>& incomingBatch, const NEvWrite::EModificationType mType) const;
80-
void AdaptBatchToSchema(NArrow::TGeneralContainer& batch, const ISnapshotSchema::TPtr& targetSchema) const;
80+
std::set<ui32> GetColumnIdsToDelete(const ISnapshotSchema::TPtr& targetSchema) const;
81+
std::vector<ui32> ConvertColumnIdsToIndexes(const std::set<ui32>& idxs) const;
8182
};
8283

8384
} // namespace NKikimr::NOlap

0 commit comments

Comments
 (0)