Skip to content

Commit 5f29865

Browse files
vlad-gogovvlad-gogov
authored
Merge e677ae3 into a179d17 
2 parents a179d17 + e677ae3 commit 5f29865

File tree

5 files changed

+131
-78
lines changed

5 files changed

+131
-78
lines changed

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

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
#include "columnshard.h"
2-
#include <ydb/core/testlib/cs_helper.h>
2+
33
#include <ydb/core/base/tablet_pipecache.h>
4+
#include <ydb/core/formats/arrow/serializer/parsing.h>
5+
#include <ydb/core/testlib/cs_helper.h>
46

57
extern "C" {
68
#include <ydb/library/yql/parser/pg_wrapper/postgresql/src/include/catalog/pg_type_d.h>
@@ -143,6 +145,13 @@ namespace NKqp {
143145
}
144146
}
145147

148+
void TTestHelper::SetCompression(
149+
const TColumnTableBase& columnTable, const TString& columnName, const TCompression& compression, const NYdb::EStatus expectedStatus) {
150+
auto alterQuery = columnTable.BuildAlterCompressionQuery(columnName, compression);
151+
auto result = GetSession().ExecuteSchemeQuery(alterQuery).GetValueSync();
152+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), expectedStatus, result.GetIssues().ToString());
153+
}
154+
146155
TString TTestHelper::TColumnSchema::BuildQuery() const {
147156
TStringBuilder str;
148157
str << Name << ' ';
@@ -185,6 +194,16 @@ namespace NKqp {
185194
return str;
186195
}
187196

197+
TString TTestHelper::TColumnTableBase::BuildAlterCompressionQuery(const TString& columnName, const TCompression& compression) const {
198+
auto str = TStringBuilder() << "ALTER OBJECT `" << Name << "` (TYPE " << GetObjectType() << ") SET";
199+
str << " (ACTION=ALTER_COLUMN, NAME=" << columnName << ", `SERIALIZER.CLASS_NAME`=`" << compression.GetSerializerName() << "`,";
200+
str << " `COMPRESSION.TYPE`=`" << NArrow::CompressionToString(compression.GetType()) << "`";
201+
if (compression.GetCompressionLevel() != Max<i32>()) {
202+
str << "`COMPRESSION.LEVEL`=" << compression.GetCompressionLevel();
203+
}
204+
str << ");";
205+
return str;
206+
}
188207

189208
std::shared_ptr<arrow::Schema> TTestHelper::TColumnTableBase::GetArrowSchema(const TVector<TColumnSchema>& columns) {
190209
std::vector<std::shared_ptr<arrow::Field>> result;

‎ydb/core/kqp/ut/common/columnshard.h‎

Lines changed: 80 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -1,88 +1,102 @@
11
#pragma once
22

33
#include "kqp_ut_common.h"
4+
5+
#include <ydb/core/tx/columnshard/test_helper/columnshard_ut_common.h>
6+
47
#include <ydb/library/accessor/accessor.h>
5-
#include <ydb/library/formats/arrow/simple_builder/filler.h>
68
#include <ydb/library/formats/arrow/simple_builder/array.h>
79
#include <ydb/library/formats/arrow/simple_builder/batch.h>
10+
#include <ydb/library/formats/arrow/simple_builder/filler.h>
811
#include <ydb/public/lib/scheme_types/scheme_type_id.h>
912
#include <ydb/public/sdk/cpp/client/ydb_table/table.h>
1013
#include <ydb/public/sdk/cpp/client/ydb_types/status_codes.h>
11-
#include <ydb/core/tx/columnshard/test_helper/columnshard_ut_common.h>
1214

1315
#include <contrib/libs/apache/arrow/cpp/src/arrow/type.h>
1416

1517
namespace NKikimr {
1618
namespace NKqp {
17-
class TTestHelper {
19+
class TTestHelper {
20+
public:
21+
class TCompression {
22+
YDB_ACCESSOR(TString, SerializerName, "ARROW_SERIALIZER");
23+
YDB_ACCESSOR(arrow::Compression::type, Type, arrow::Compression::type::UNCOMPRESSED);
24+
YDB_ACCESSOR(i32, CompressionLevel, Max<i32>());
25+
};
26+
27+
class TColumnSchema {
28+
YDB_ACCESSOR_DEF(TString, Name);
29+
YDB_ACCESSOR_DEF(NScheme::TTypeInfo, TypeInfo);
30+
YDB_FLAG_ACCESSOR(Nullable, true);
31+
1832
public:
19-
class TColumnSchema {
20-
YDB_ACCESSOR_DEF(TString, Name);
21-
YDB_ACCESSOR_DEF(NScheme::TTypeInfo, TypeInfo);
22-
YDB_FLAG_ACCESSOR(Nullable, true);
23-
public:
24-
TString BuildQuery() const;
25-
26-
TColumnSchema& SetType(NScheme::TTypeId typeId);
27-
};
28-
29-
using TUpdatesBuilder = NColumnShard::TTableUpdatesBuilder;
30-
31-
class TColumnTableBase {
32-
YDB_ACCESSOR_DEF(TString, Name);
33-
YDB_ACCESSOR_DEF(TVector<TColumnSchema>, Schema);
34-
YDB_ACCESSOR_DEF(TVector<TString>, PrimaryKey);
35-
YDB_ACCESSOR_DEF(TVector<TString>, Sharding);
36-
YDB_ACCESSOR(ui32, MinPartitionsCount, 1);
37-
38-
std::optional<std::pair<TString, TString>> TTLConf;
39-
public:
40-
TString BuildQuery() const;
41-
std::shared_ptr<arrow::Schema> GetArrowSchema(const TVector<TColumnSchema>& columns);
42-
43-
TColumnTableBase& SetTTL(const TString& columnName, const TString& ttlConf) {
44-
TTLConf = std::make_pair(columnName, ttlConf);
45-
return *this;
46-
}
47-
48-
private:
49-
virtual TString GetObjectType() const = 0;
50-
TString BuildColumnsStr(const TVector<TColumnSchema>& clumns) const;
51-
std::shared_ptr<arrow::Field> BuildField(const TString name, const NScheme::TTypeInfo& typeInfo, bool nullable) const;
52-
};
53-
54-
class TColumnTable : public TColumnTableBase {
55-
private:
56-
TString GetObjectType() const override;
57-
};
58-
59-
class TColumnTableStore : public TColumnTableBase {
60-
private:
61-
TString GetObjectType() const override;
62-
};
33+
TString BuildQuery() const;
6334

64-
private:
65-
std::unique_ptr<TKikimrRunner> Kikimr;
66-
std::unique_ptr<NYdb::NTable::TTableClient> TableClient;
67-
std::unique_ptr<NYdb::NTable::TSession> Session;
35+
TColumnSchema& SetType(NScheme::TTypeId typeId);
36+
};
37+
38+
using TUpdatesBuilder = NColumnShard::TTableUpdatesBuilder;
39+
40+
class TColumnTableBase {
41+
YDB_ACCESSOR_DEF(TString, Name);
42+
YDB_ACCESSOR_DEF(TVector<TColumnSchema>, Schema);
43+
YDB_ACCESSOR_DEF(TVector<TString>, PrimaryKey);
44+
YDB_ACCESSOR_DEF(TVector<TString>, Sharding);
45+
YDB_ACCESSOR(ui32, MinPartitionsCount, 1);
46+
47+
std::optional<std::pair<TString, TString>> TTLConf;
6848

6949
public:
70-
TTestHelper(const TKikimrSettings& settings);
71-
TKikimrRunner& GetKikimr();
72-
TTestActorRuntime& GetRuntime();
73-
NYdb::NTable::TSession& GetSession();
74-
void CreateTable(const TColumnTableBase& table, const NYdb::EStatus expectedStatus = NYdb::EStatus::SUCCESS);
75-
void DropTable(const TString& tableName);
76-
void CreateTier(const TString& tierName);
77-
TString CreateTieringRule(const TString& tierName, const TString& columnName);
78-
void SetTiering(const TString& tableName, const TString& ruleName);
79-
void ResetTiering(const TString& tableName);
80-
void BulkUpsert(const TColumnTable& table, TTestHelper::TUpdatesBuilder& updates, const Ydb::StatusIds_StatusCode& opStatus = Ydb::StatusIds::SUCCESS);
81-
void BulkUpsert(const TColumnTable& table, std::shared_ptr<arrow::RecordBatch> batch, const Ydb::StatusIds_StatusCode& opStatus = Ydb::StatusIds::SUCCESS);
82-
void ReadData(const TString& query, const TString& expected, const NYdb::EStatus opStatus = NYdb::EStatus::SUCCESS);
83-
void RebootTablets(const TString& tableName);
84-
void WaitTabletDeletionInHive(ui64 tabletId, TDuration duration);
50+
TString BuildQuery() const;
51+
TString BuildAlterCompressionQuery(const TString& columnName, const TCompression& compression) const;
52+
std::shared_ptr<arrow::Schema> GetArrowSchema(const TVector<TColumnSchema>& columns);
53+
54+
TColumnTableBase& SetTTL(const TString& columnName, const TString& ttlConf) {
55+
TTLConf = std::make_pair(columnName, ttlConf);
56+
return *this;
57+
}
58+
59+
private:
60+
virtual TString GetObjectType() const = 0;
61+
TString BuildColumnsStr(const TVector<TColumnSchema>& clumns) const;
62+
std::shared_ptr<arrow::Field> BuildField(const TString name, const NScheme::TTypeInfo& typeInfo, bool nullable) const;
63+
};
64+
65+
class TColumnTable: public TColumnTableBase {
66+
private:
67+
TString GetObjectType() const override;
68+
};
69+
70+
class TColumnTableStore: public TColumnTableBase {
71+
private:
72+
TString GetObjectType() const override;
8573
};
8674

75+
private:
76+
std::unique_ptr<TKikimrRunner> Kikimr;
77+
std::unique_ptr<NYdb::NTable::TTableClient> TableClient;
78+
std::unique_ptr<NYdb::NTable::TSession> Session;
79+
80+
public:
81+
TTestHelper(const TKikimrSettings& settings);
82+
TKikimrRunner& GetKikimr();
83+
TTestActorRuntime& GetRuntime();
84+
NYdb::NTable::TSession& GetSession();
85+
void CreateTable(const TColumnTableBase& table, const NYdb::EStatus expectedStatus = NYdb::EStatus::SUCCESS);
86+
void DropTable(const TString& tableName);
87+
void CreateTier(const TString& tierName);
88+
TString CreateTieringRule(const TString& tierName, const TString& columnName);
89+
void SetTiering(const TString& tableName, const TString& ruleName);
90+
void ResetTiering(const TString& tableName);
91+
void BulkUpsert(
92+
const TColumnTable& table, TTestHelper::TUpdatesBuilder& updates, const Ydb::StatusIds_StatusCode& opStatus = Ydb::StatusIds::SUCCESS);
93+
void BulkUpsert(const TColumnTable& table, std::shared_ptr<arrow::RecordBatch> batch,
94+
const Ydb::StatusIds_StatusCode& opStatus = Ydb::StatusIds::SUCCESS);
95+
void ReadData(const TString& query, const TString& expected, const NYdb::EStatus opStatus = NYdb::EStatus::SUCCESS);
96+
void RebootTablets(const TString& tableName);
97+
void WaitTabletDeletionInHive(ui64 tabletId, TDuration duration);
98+
void SetCompression(const TColumnTableBase& columnTable, const TString& columnName, const TCompression& compression,
99+
const NYdb::EStatus expectedStatus = NYdb::EStatus::SUCCESS);
100+
};
87101
}
88102
}

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

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
#include <ydb/core/kqp/ut/common/columnshard.h>
2+
3+
namespace NKikimr::NKqp {
4+
5+
Y_UNIT_TEST_SUITE(KqpOlapCompression) {
6+
Y_UNIT_TEST(DisabledAlterCompression) {
7+
TKikimrSettings settings = TKikimrSettings().SetWithSampleTables(false).SetEnableOlapCompression(false);
8+
TTestHelper testHelper(settings);
9+
TVector<TTestHelper::TColumnSchema> schema = {
10+
TTestHelper::TColumnSchema().SetName("pk_int").SetType(NScheme::NTypeIds::Uint64).SetNullable(false)
11+
};
12+
TTestHelper::TCompression compression = TTestHelper::TCompression().SetType(arrow::Compression::type::ZSTD);
13+
14+
TTestHelper::TColumnTable standaloneTable;
15+
standaloneTable.SetName("/Root/StandaloneTable").SetPrimaryKey({ "pk_int" }).SetSharding({ "pk_int" }).SetSchema(schema);
16+
testHelper.CreateTable(standaloneTable);
17+
testHelper.SetCompression(standaloneTable, "pk_int", compression, NYdb::EStatus::SCHEME_ERROR);
18+
19+
TTestHelper::TColumnTableStore testTableStore;
20+
testTableStore.SetName("/Root/TableStoreTest").SetPrimaryKey({ "pk_int" }).SetSchema(schema);
21+
testHelper.CreateTable(testTableStore);
22+
testHelper.SetCompression(testTableStore, "pk_int", compression, NYdb::EStatus::PRECONDITION_FAILED);
23+
24+
TTestHelper::TColumnTable testTable;
25+
testTable.SetName("/Root/TableStoreTest/ColumnTableTest").SetPrimaryKey({ "pk_int" }).SetSharding({ "pk_int" }).SetSchema(schema);
26+
testHelper.CreateTable(testTable);
27+
testHelper.SetCompression(testTable, "pk_int", compression, NYdb::EStatus::SCHEME_ERROR);
28+
}
29+
}
30+
}

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

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -219,17 +219,6 @@ Y_UNIT_TEST_SUITE(KqpOlapSysView) {
219219
}
220220
}
221221

222-
Y_UNIT_TEST(DisabledAlterCompression) {
223-
TKikimrSettings settings = TKikimrSettings().SetWithSampleTables(false).SetEnableOlapCompression(false);
224-
TKikimrRunner kikimr(settings);
225-
TTypedLocalHelper helper("", kikimr, "olapTable", "olapStore");
226-
helper.CreateTestOlapTable();
227-
helper.FillPKOnly(0, 1);
228-
helper.ExecuteSchemeQuery(
229-
"ALTER OBJECT `/Root/olapStore` (TYPE TABLESTORE) SET (ACTION=ALTER_COLUMN, NAME=pk_int, "
230-
"`SERIALIZER.CLASS_NAME`=`ARROW_SERIALIZER`, `COMPRESSION.TYPE`=`zstd`);", NYdb::EStatus::PRECONDITION_FAILED);
231-
}
232-
233222
Y_UNIT_TEST(StatsSysViewBytesColumnActualization) {
234223
ui64 rawBytes1;
235224
ui64 bytes1;

‎ydb/core/kqp/ut/olap/ya.make‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ SRCS(
2626
sparsed_ut.cpp
2727
tiering_ut.cpp
2828
decimal_ut.cpp
29+
compression_ut.cpp
2930
)
3031

3132
PEERDIR(

0 commit comments

Comments
 (0)