Skip to content

Commit acee834

Browse files
nikvas0nikvas0
authored
Fix sink index column order (#9190)
1 parent 7b31943 commit acee834

10 files changed

+134
-19
lines changed

‎.github/config/muted_ya.txt

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,10 @@ ydb/core/kqp/ut/service KqpService.CloseSessionsWithLoad
2727
ydb/core/kqp/ut/service KqpQueryService.TableSink_OlapRWQueries
2828
ydb/core/kqp/ut/service KqpQueryService.TableSink_Htap
2929
ydb/core/kqp/ut/tx KqpSnapshotRead.ReadOnlyTxWithIndexCommitsOnConcurrentWrite+withSink
30+
ydb/core/kqp/ut/tx KqpSinkMvcc.OltpNamedStatement
31+
ydb/core/kqp/ut/tx KqpSinkMvcc.OlapNamedStatement
32+
ydb/core/kqp/ut/tx KqpSinkMvcc.OltpMultiSinks
33+
ydb/core/kqp/ut/tx KqpSinkMvcc.OlapMultiSinks
3034
ydb/core/persqueue/ut [*/*]*
3135
ydb/core/persqueue/ut TPQTest.*DirectRead*
3236
ydb/core/persqueue/ut/ut_with_sdk [*/*]*

‎ydb/core/kqp/opt/physical/effects/kqp_opt_phy_delete_index.cpp

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -88,14 +88,18 @@ TExprBase KqpBuildDeleteIndexStages(TExprBase node, TExprContext& ctx, const TKq
8888
effects.emplace_back(tableDelete);
8989

9090
for (const auto& [tableNode, indexDesc] : indexes) {
91-
THashSet<TStringBuf> indexTableColumns;
91+
THashSet<TStringBuf> indexTableColumnsSet;
92+
TVector<TStringBuf> indexTableColumns;
9293

9394
for (const auto& column : indexDesc->KeyColumns) {
94-
YQL_ENSURE(indexTableColumns.emplace(column).second);
95+
YQL_ENSURE(indexTableColumnsSet.emplace(column).second);
96+
indexTableColumns.emplace_back(column);
9597
}
9698

9799
for (const auto& column : pk) {
98-
indexTableColumns.insert(column);
100+
if (indexTableColumnsSet.insert(column).second) {
101+
indexTableColumns.emplace_back(column);
102+
}
99103
}
100104

101105
auto deleteIndexKeys = MakeRowsFromDict(lookupDict.Cast(), pk, indexTableColumns, del.Pos(), ctx);

‎ydb/core/kqp/opt/physical/effects/kqp_opt_phy_effects_impl.h

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -62,11 +62,11 @@ NYql::NNodes::TCoLambda MakeRowsPayloadSelector(const NYql::NNodes::TCoAtomList&
6262
const NYql::TKikimrTableDescription& table, NYql::TPositionHandle pos, NYql::TExprContext& ctx);
6363

6464
NYql::NNodes::TExprBase MakeRowsFromDict(const NYql::NNodes::TDqPhyPrecompute& dict, const TVector<TString>& dictKeys,
65-
const THashSet<TStringBuf>& columns, NYql::TPositionHandle pos, NYql::TExprContext& ctx);
65+
const TVector<TStringBuf>& columns, NYql::TPositionHandle pos, NYql::TExprContext& ctx);
6666

6767
// Same as MakeRowsFromDict but skip rows which marked as non changed (true in second tuple)
6868
NYql::NNodes::TExprBase MakeRowsFromTupleDict(const NYql::NNodes::TDqPhyPrecompute& dict, const TVector<TString>& dictKeys,
69-
const THashSet<TStringBuf>& columns, NYql::TPositionHandle pos, NYql::TExprContext& ctx);
69+
const TVector<TStringBuf>& columns, NYql::TPositionHandle pos, NYql::TExprContext& ctx);
7070

7171
NYql::NNodes::TMaybeNode<NYql::NNodes::TDqCnUnionAll> MakeConditionalInsertRows(const NYql::NNodes::TExprBase& input,
7272
const NYql::TKikimrTableDescription& table, const TMaybe<THashSet<TStringBuf>>& inputColumn, bool abortOnError,

‎ydb/core/kqp/opt/physical/effects/kqp_opt_phy_indexes.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -238,7 +238,7 @@ TMaybeNode<TDqPhyPrecompute> PrecomputeTableLookupDict(const TDqPhyPrecompute& l
238238
}
239239

240240
TExprBase MakeRowsFromDict(const TDqPhyPrecompute& dict, const TVector<TString>& dictKeys,
241-
const THashSet<TStringBuf>& columns, TPositionHandle pos, TExprContext& ctx)
241+
const TVector<TStringBuf>& columns, TPositionHandle pos, TExprContext& ctx)
242242
{
243243
THashSet<TString> dictKeysSet(dictKeys.begin(), dictKeys.end());
244244
auto dictTupleArg = TCoArgument(ctx.NewArgument(pos, "dict_tuple"));
@@ -296,7 +296,7 @@ TExprBase MakeRowsFromDict(const TDqPhyPrecompute& dict, const TVector<TString>&
296296
}
297297

298298
TExprBase MakeRowsFromTupleDict(const TDqPhyPrecompute& dict, const TVector<TString>& dictKeys,
299-
const THashSet<TStringBuf>& columns, TPositionHandle pos, TExprContext& ctx)
299+
const TVector<TStringBuf>& columns, TPositionHandle pos, TExprContext& ctx)
300300
{
301301
THashSet<TString> dictKeysSet(dictKeys.begin(), dictKeys.end());
302302
auto dictTupleArg = TCoArgument(ctx.NewArgument(pos, "dict_tuple"));

‎ydb/core/kqp/opt/physical/effects/kqp_opt_phy_insert_index.cpp

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ using namespace NYql::NNodes;
1212
namespace {
1313

1414
TExprBase MakeInsertIndexRows(const TDqPhyPrecompute& inputRows, const TKikimrTableDescription& table,
15-
const THashSet<TStringBuf>& inputColumns, const THashSet<TStringBuf>& indexColumns,
15+
const THashSet<TStringBuf>& inputColumns, const TVector<TStringBuf>& indexColumns,
1616
TPositionHandle pos, TExprContext& ctx)
1717
{
1818
auto inputRowArg = TCoArgument(ctx.NewArgument(pos, "input_row"));
@@ -113,19 +113,24 @@ TExprBase KqpBuildInsertIndexStages(TExprBase node, TExprContext& ctx, const TKq
113113
effects.emplace_back(upsertTable);
114114

115115
for (const auto& [tableNode, indexDesc] : indexes) {
116-
THashSet<TStringBuf> indexTableColumns;
116+
THashSet<TStringBuf> indexTableColumnsSet;
117+
TVector<TStringBuf> indexTableColumns;
117118

118119
for (const auto& column : indexDesc->KeyColumns) {
119-
YQL_ENSURE(indexTableColumns.emplace(column).second);
120+
YQL_ENSURE(indexTableColumnsSet.emplace(column).second);
121+
indexTableColumns.emplace_back(column);
120122
}
121123

122124
for (const auto& column : table.Metadata->KeyColumnNames) {
123-
indexTableColumns.insert(column);
125+
if (indexTableColumnsSet.insert(column).second) {
126+
indexTableColumns.emplace_back(column);
127+
}
124128
}
125129

126130
for (const auto& column : indexDesc->DataColumns) {
127131
if (inputColumnsSet.contains(column)) {
128-
YQL_ENSURE(indexTableColumns.emplace(column).second);
132+
YQL_ENSURE(indexTableColumnsSet.emplace(column).second);
133+
indexTableColumns.emplace_back(column);
129134
}
130135
}
131136

‎ydb/core/kqp/opt/physical/effects/kqp_opt_phy_upsert_index.cpp

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -150,7 +150,7 @@ TExprBase MakeNonexistingRowsFilter(const TDqPhyPrecompute& inputRows, const TDq
150150

151151
TExprBase MakeUpsertIndexRows(TKqpPhyUpsertIndexMode mode, const TDqPhyPrecompute& inputRows,
152152
const TDqPhyPrecompute& lookupDict, const THashSet<TStringBuf>& inputColumns,
153-
const THashSet<TStringBuf>& indexColumns, const TKikimrTableDescription& table, TPositionHandle pos,
153+
const TVector<TStringBuf>& indexColumns, const TKikimrTableDescription& table, TPositionHandle pos,
154154
TExprContext& ctx, bool opt)
155155
{
156156
// Check if we can update index table from just input data
@@ -686,9 +686,11 @@ TMaybeNode<TExprList> KqpPhyUpsertIndexEffectsImpl(TKqpPhyUpsertIndexMode mode,
686686

687687
for (const auto& [tableNode, indexDesc] : indexes) {
688688
bool indexKeyColumnsUpdated = false;
689-
THashSet<TStringBuf> indexTableColumns;
689+
THashSet<TStringBuf> indexTableColumnsSet;
690+
TVector<TStringBuf> indexTableColumns;
690691
for (const auto& column : indexDesc->KeyColumns) {
691-
YQL_ENSURE(indexTableColumns.emplace(column).second);
692+
YQL_ENSURE(indexTableColumnsSet.emplace(column).second);
693+
indexTableColumns.emplace_back(column);
692694

693695
if (mode == TKqpPhyUpsertIndexMode::UpdateOn && table.GetKeyColumnIndex(column)) {
694696
// Table PK cannot be updated, so don't consider PK columns update as index update
@@ -701,7 +703,9 @@ TMaybeNode<TExprList> KqpPhyUpsertIndexEffectsImpl(TKqpPhyUpsertIndexMode mode,
701703
}
702704

703705
for (const auto& column : pk) {
704-
indexTableColumns.insert(column);
706+
if (indexTableColumnsSet.insert(column).second) {
707+
indexTableColumns.emplace_back(column);
708+
}
705709
}
706710

707711
auto indexTableColumnsWithoutData = indexTableColumns;
@@ -710,7 +714,8 @@ TMaybeNode<TExprList> KqpPhyUpsertIndexEffectsImpl(TKqpPhyUpsertIndexMode mode,
710714
bool optUpsert = true;
711715
for (const auto& column : indexDesc->DataColumns) {
712716
// TODO: Conder not fetching/updating data columns without input value.
713-
YQL_ENSURE(indexTableColumns.emplace(column).second);
717+
YQL_ENSURE(indexTableColumnsSet.emplace(column).second);
718+
indexTableColumns.emplace_back(column);
714719

715720
if (inputColumnsSet.contains(column)) {
716721
indexDataColumnsUpdated = true;

‎ydb/core/kqp/opt/physical/kqp_opt_phy_helpers.cpp

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,10 @@ TCoAtomList BuildColumnsList(const THashSet<TStringBuf>& columns, TPositionHandl
8383
return BuildColumnsListImpl(columns, pos, ctx);
8484
}
8585

86+
TCoAtomList BuildColumnsList(const TVector<TStringBuf>& columns, NYql::TPositionHandle pos,NYql::TExprContext& ctx) {
87+
return BuildColumnsListImpl(columns, pos, ctx);
88+
}
89+
8690
TCoAtomList BuildColumnsList(const TVector<TString>& columns, TPositionHandle pos, TExprContext& ctx) {
8791
return BuildColumnsListImpl(columns, pos, ctx);
8892
}

‎ydb/core/kqp/opt/physical/kqp_opt_phy_impl.h

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,9 @@ NYql::NNodes::TMaybeNode<NYql::NNodes::TDqPhyPrecompute> BuildLookupKeysPrecompu
1919
NYql::NNodes::TCoAtomList BuildColumnsList(const THashSet<TStringBuf>& columns, NYql::TPositionHandle pos,
2020
NYql::TExprContext& ctx);
2121

22+
NYql::NNodes::TCoAtomList BuildColumnsList(const TVector<TStringBuf>& columns, NYql::TPositionHandle pos,
23+
NYql::TExprContext& ctx);
24+
2225
NYql::NNodes::TCoAtomList BuildColumnsList(const TVector<TString>& columns, NYql::TPositionHandle pos,
2326
NYql::TExprContext& ctx);
2427

‎ydb/core/kqp/ut/tx/kqp_sink_common.h

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,12 +18,13 @@ class TTableDataModificationTester {
1818
std::unique_ptr<TKikimrRunner> Kikimr;
1919
YDB_ACCESSOR(bool, IsOlap, false);
2020
YDB_ACCESSOR(bool, FastSnapshotExpiration, false);
21+
YDB_ACCESSOR(bool, DisableSinks, false);
2122

2223
virtual void DoExecute() = 0;
2324
public:
2425
void Execute() {
25-
AppConfig.MutableTableServiceConfig()->SetEnableOlapSink(true);
26-
AppConfig.MutableTableServiceConfig()->SetEnableOltpSink(true);
26+
AppConfig.MutableTableServiceConfig()->SetEnableOlapSink(!DisableSinks);
27+
AppConfig.MutableTableServiceConfig()->SetEnableOltpSink(!DisableSinks);
2728
AppConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(true);
2829
auto settings = TKikimrSettings().SetAppConfig(AppConfig).SetWithSampleTables(false);
2930
if (FastSnapshotExpiration) {

‎ydb/core/kqp/ut/tx/kqp_sink_mvcc_ut.cpp

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -299,6 +299,95 @@ Y_UNIT_TEST_SUITE(KqpSinkMvcc) {
299299
tester.SetIsOlap(true);
300300
tester.Execute();
301301
}
302+
303+
class TNamedStatement : public TTableDataModificationTester {
304+
protected:
305+
void DoExecute() override {
306+
auto client = Kikimr->GetQueryClient();
307+
308+
auto session1 = client.GetSession().GetValueSync().GetSession();
309+
310+
{
311+
auto result = session1.ExecuteQuery(Q_(R"(
312+
$data = SELECT * FROM `/Root/KV`;
313+
DELETE FROM `/Root/KV` WHERE 1=1;
314+
SELECT COUNT(*) FROM `/Root/KV`;
315+
SELECT COUNT(*) FROM $data;
316+
DELETE FROM `/Root/KV` ON SELECT 424242u AS Key, "One" As Value;
317+
UPSERT INTO `/Root/KV` (Key, Value) VALUES (424242u, "One");
318+
SELECT COUNT(*) FROM `/Root/KV`;
319+
SELECT COUNT(*) FROM $data;
320+
)"), TTxControl::BeginTx(TTxSettings::SerializableRW()).CommitTx()).ExtractValueSync();
321+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
322+
CompareYson(R"([[0u]])", FormatResultSetYson(result.GetResultSet(0)));
323+
CompareYson(R"([[0u]])", FormatResultSetYson(result.GetResultSet(1)));
324+
CompareYson(R"([[1u]])", FormatResultSetYson(result.GetResultSet(2)));
325+
CompareYson(R"([[1u]])", FormatResultSetYson(result.GetResultSet(3)));
326+
}
327+
}
328+
};
329+
330+
Y_UNIT_TEST(OltpNamedStatementNoSink) {
331+
TNamedStatement tester;
332+
tester.SetDisableSinks(true);
333+
tester.SetIsOlap(false);
334+
tester.Execute();
335+
}
336+
337+
Y_UNIT_TEST(OltpNamedStatement) {
338+
TNamedStatement tester;
339+
tester.SetIsOlap(false);
340+
tester.Execute();
341+
}
342+
343+
Y_UNIT_TEST(OlapNamedStatement) {
344+
TNamedStatement tester;
345+
tester.SetIsOlap(true);
346+
tester.Execute();
347+
}
348+
349+
class TMultiSinks: public TTableDataModificationTester {
350+
protected:
351+
void DoExecute() override {
352+
auto client = Kikimr->GetQueryClient();
353+
354+
auto session1 = client.GetSession().GetValueSync().GetSession();
355+
356+
{
357+
auto result = session1.ExecuteQuery(Q_(R"(
358+
UPSERT INTO `/Root/KV` (Key, Value) VALUES (1u, "1");
359+
UPSERT INTO `/Root/KV` (Key, Value) VALUES (1u, "2");
360+
)"), TTxControl::BeginTx(TTxSettings::SerializableRW()).CommitTx()).ExtractValueSync();
361+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
362+
}
363+
{
364+
auto result = session1.ExecuteQuery(Q_(R"(
365+
SELECT Value FROM `/Root/KV` WHERE Key = 1u;
366+
)"), TTxControl::BeginTx(TTxSettings::SerializableRW()).CommitTx()).ExtractValueSync();
367+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
368+
CompareYson(R"([[["2"]]])", FormatResultSetYson(result.GetResultSet(0)));
369+
}
370+
}
371+
};
372+
373+
Y_UNIT_TEST(OltpMultiSinksNoSinks) {
374+
TMultiSinks tester;
375+
tester.SetDisableSinks(true);
376+
tester.SetIsOlap(false);
377+
tester.Execute();
378+
}
379+
380+
Y_UNIT_TEST(OltpMultiSinks) {
381+
TMultiSinks tester;
382+
tester.SetIsOlap(false);
383+
tester.Execute();
384+
}
385+
386+
Y_UNIT_TEST(OlapMultiSinks) {
387+
TMultiSinks tester;
388+
tester.SetIsOlap(true);
389+
tester.Execute();
390+
}
302391
}
303392

304393
} // namespace NKqp

0 commit comments

Comments
 (0)