Skip to content

Commit 5bcc9c4

Browse files
ivanmorozov333ivanmorozov333
authored
Merge a86f731 into 2b752b4 
2 parents 2b752b4 + a86f731 commit 5bcc9c4

File tree

26 files changed

+320
-152
lines changed

26 files changed

+320
-152
lines changed

‎ydb/core/formats/arrow/permutations.cpp‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@
1111
#include <ydb/library/actors/core/log.h>
1212

1313
#include <contrib/libs/apache/arrow/cpp/src/arrow/array/builder_primitive.h>
14-
#include <contrib/libs/xxhash/xxhash.h>
1514

1615
namespace NKikimr::NArrow {
1716

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

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -482,7 +482,7 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
482482
}
483483
{
484484
ResetZeroLevel(csController);
485-
ui32 requestsCount = 100;
485+
ui32 requestsCount = 300;
486486
for (ui32 i = 0; i < requestsCount; ++i) {
487487
const ui32 idx = RandomNumber<ui32>(uids.size());
488488
const auto query = [](const TString& res, const TString& uid, const ui32 level) {
@@ -494,12 +494,12 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
494494
};
495495
ExecuteSQL(query(resourceIds[idx], uids[idx], levels[idx]), "[[1u;]]");
496496
}
497-
AFL_VERIFY(csController->GetIndexesSkippingOnSelect().Val() - SkipStart > 1)("approved", csController->GetIndexesApprovedOnSelect().Val() - ApproveStart)(
497+
AFL_VERIFY(csController->GetIndexesSkippingOnSelect().Val() - SkipStart)("approved", csController->GetIndexesApprovedOnSelect().Val() - ApproveStart)(
498498
"skipped", csController->GetIndexesSkippingOnSelect().Val() - SkipStart);
499499
}
500500
{
501501
ResetZeroLevel(csController);
502-
ui32 requestsCount = 100;
502+
ui32 requestsCount = 300;
503503
for (ui32 i = 0; i < requestsCount; ++i) {
504504
const ui32 idx = RandomNumber<ui32>(uids.size());
505505
const auto query = [](const TString& res, const TString& uid, const ui32 level) {
@@ -511,13 +511,13 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
511511
};
512512
ExecuteSQL(query(resourceIds[idx], uids[idx], levels[idx]), "[[1u;]]");
513513
}
514-
AFL_VERIFY(csController->GetIndexesSkippingOnSelect().Val() - SkipStart > 1)(
514+
AFL_VERIFY(csController->GetIndexesSkippingOnSelect().Val() - SkipStart)(
515515
"approved", csController->GetIndexesApprovedOnSelect().Val() - ApproveStart)(
516516
"skipped", csController->GetIndexesSkippingOnSelect().Val() - SkipStart);
517517
}
518518
{
519519
ResetZeroLevel(csController);
520-
ui32 requestsCount = 100;
520+
ui32 requestsCount = 300;
521521
for (ui32 i = 0; i < requestsCount; ++i) {
522522
const ui32 idx = RandomNumber<ui32>(uids.size());
523523
const auto query = [](const TString& res, const TString& uid, const ui32 level) {
@@ -529,7 +529,7 @@ Y_UNIT_TEST_SUITE(KqpOlapIndexes) {
529529
};
530530
ExecuteSQL(query(resourceIds[idx], uids[idx], levels[idx]), "[[1u;]]");
531531
}
532-
AFL_VERIFY(csController->GetIndexesSkippingOnSelect().Val() - SkipStart > 1)(
532+
AFL_VERIFY(csController->GetIndexesSkippingOnSelect().Val() - SkipStart)(
533533
"approved", csController->GetIndexesApprovedOnSelect().Val() - ApproveStart)(
534534
"skipped", csController->GetIndexesSkippingOnSelect().Val() - SkipStart);
535535
}

‎ydb/core/tx/columnshard/engines/changes/compaction/merger.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -147,7 +147,7 @@ std::vector<TWritePortionInfoWithBlobsResult> TMerger::Execute(const std::shared
147147
for (auto&& i : packs) {
148148
TGeneralSerializedSlice slicePrimary(std::move(i));
149149
auto dataWithSecondary = resultFiltered->GetIndexInfo()
150-
.AppendIndexes(slicePrimary.GetPortionChunksToHash(), SaverContext.GetStoragesManager())
150+
.AppendIndexes(slicePrimary.GetPortionChunksToHash(), SaverContext.GetStoragesManager(), slicePrimary.GetRecordsCount())
151151
.DetachResult();
152152
TGeneralSerializedSlice slice(dataWithSecondary.GetExternalData(), schemaDetails, Context.Counters.SplitterCounters);
153153

‎ydb/core/tx/columnshard/engines/portions/read_with_blobs.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -124,7 +124,7 @@ std::optional<TWritePortionInfoWithBlobsResult> TReadPortionInfoWithBlobs::SyncP
124124
TIndexInfo::TSecondaryData secondaryData;
125125
secondaryData.MutableExternalData() = entityChunksNew;
126126
for (auto&& i : to->GetIndexInfo().GetIndexes()) {
127-
to->GetIndexInfo().AppendIndex(entityChunksNew, i.first, storages, secondaryData).Validate();
127+
to->GetIndexInfo().AppendIndex(entityChunksNew, i.first, storages, source.PortionInfo.GetPortionInfo().GetRecordsCount(), secondaryData).Validate();
128128
}
129129

130130
const NSplitter::TEntityGroups groups = source.PortionInfo.GetPortionInfo().GetEntityGroupsByStorageId(targetTier, *storages, to->GetIndexInfo());

‎ydb/core/tx/columnshard/engines/reader/common_reader/iterator/fetching.h‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,12 +27,12 @@ class TFetchingStepSignals: public NColumnShard::TCommonCountersOwner {
2727
public:
2828
TFetchingStepSignals(NColumnShard::TCommonCountersOwner&& owner)
2929
: TBase(std::move(owner))
30-
, DurationCounter(TBase::GetDeriviative("duration_ms"))
31-
, BytesCounter(TBase::GetDeriviative("bytes_ms")) {
30+
, DurationCounter(TBase::GetDeriviative("Duration/Us"))
31+
, BytesCounter(TBase::GetDeriviative("Bytes/Count")) {
3232
}
3333

3434
void AddDuration(const TDuration d) const {
35-
DurationCounter->Add(d.MilliSeconds());
35+
DurationCounter->Add(d.MicroSeconds());
3636
}
3737

3838
void AddBytes(const ui32 v) const {
@@ -56,7 +56,7 @@ class TFetchingStepsSignalsCollection: public NColumnShard::TCommonCountersOwner
5656

5757
public:
5858
TFetchingStepsSignalsCollection()
59-
: TBase("scan_steps") {
59+
: TBase("ScanSteps") {
6060
}
6161

6262
static TFetchingStepSignals GetSignals(const TString& name) {

‎ydb/core/tx/columnshard/engines/scheme/index_info.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -412,11 +412,11 @@ std::shared_ptr<arrow::Scalar> TIndexInfo::GetColumnExternalDefaultValueVerified
412412
}
413413

414414
NKikimr::TConclusionStatus TIndexInfo::AppendIndex(const THashMap<ui32, std::vector<std::shared_ptr<IPortionDataChunk>>>& originalData,
415-
const ui32 indexId, const std::shared_ptr<IStoragesManager>& operators, TSecondaryData& result) const {
415+
const ui32 indexId, const std::shared_ptr<IStoragesManager>& operators, const ui32 recordsCount, TSecondaryData& result) const {
416416
auto it = Indexes.find(indexId);
417417
AFL_VERIFY(it != Indexes.end());
418418
auto& index = it->second;
419-
std::shared_ptr<IPortionDataChunk> chunk = index->BuildIndex(originalData, *this);
419+
std::shared_ptr<IPortionDataChunk> chunk = index->BuildIndex(originalData, recordsCount, *this);
420420
auto opStorage = operators->GetOperatorVerified(index->GetStorageId());
421421
if ((i64)chunk->GetPackedSize() > opStorage->GetBlobSplitSettings().GetMaxBlobSize()) {
422422
return TConclusionStatus::Fail("blob size for secondary data (" + ::ToString(indexId) + ") bigger than limit (" +

‎ydb/core/tx/columnshard/engines/scheme/index_info.h‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -313,11 +313,11 @@ struct TIndexInfo: public IIndexInfo {
313313
};
314314

315315
[[nodiscard]] TConclusion<TSecondaryData> AppendIndexes(const THashMap<ui32, std::vector<std::shared_ptr<IPortionDataChunk>>>& primaryData,
316-
const std::shared_ptr<IStoragesManager>& operators) const {
316+
const std::shared_ptr<IStoragesManager>& operators, const ui32 recordsCount) const {
317317
TSecondaryData result;
318318
result.MutableExternalData() = primaryData;
319319
for (auto&& i : Indexes) {
320-
auto conclusion = AppendIndex(primaryData, i.first, operators, result);
320+
auto conclusion = AppendIndex(primaryData, i.first, operators, recordsCount, result);
321321
if (conclusion.IsFail()) {
322322
return conclusion;
323323
}
@@ -329,7 +329,7 @@ struct TIndexInfo: public IIndexInfo {
329329
std::shared_ptr<NIndexes::NCountMinSketch::TIndexMeta> GetIndexMetaCountMinSketch(const std::set<ui32>& columnIds) const;
330330

331331
[[nodiscard]] TConclusionStatus AppendIndex(const THashMap<ui32, std::vector<std::shared_ptr<IPortionDataChunk>>>& originalData,
332-
const ui32 indexId, const std::shared_ptr<IStoragesManager>& operators, TSecondaryData& result) const;
332+
const ui32 indexId, const std::shared_ptr<IStoragesManager>& operators, const ui32 recordsCount, TSecondaryData& result) const;
333333

334334
/// Returns an id of the column located by name. The name should exists in the schema.
335335
ui32 GetColumnIdVerified(const std::string& name) const;

‎ydb/core/tx/columnshard/engines/scheme/indexes/abstract/meta.h‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,8 @@ class IIndexMeta {
3131
YDB_READONLY(ui32, IndexId, 0);
3232
YDB_READONLY(TString, StorageId, IStoragesManager::DefaultStorageId);
3333
protected:
34-
virtual std::shared_ptr<IPortionDataChunk> DoBuildIndex(const THashMap<ui32, std::vector<std::shared_ptr<IPortionDataChunk>>>& data, const TIndexInfo& indexInfo) const = 0;
34+
virtual std::shared_ptr<IPortionDataChunk> DoBuildIndex(const THashMap<ui32, std::vector<std::shared_ptr<IPortionDataChunk>>>& data,
35+
const ui32 recordsCount, const TIndexInfo& indexInfo) const = 0;
3536
virtual void DoFillIndexCheckers(const std::shared_ptr<NRequest::TDataForIndexesCheckers>& info, const NSchemeShard::TOlapSchema& schema) const = 0;
3637
virtual bool DoDeserializeFromProto(const NKikimrSchemeOp::TOlapIndexDescription& proto) = 0;
3738
virtual void DoSerializeToProto(NKikimrSchemeOp::TOlapIndexDescription& proto) const = 0;
@@ -67,8 +68,8 @@ class IIndexMeta {
6768

6869
virtual ~IIndexMeta() = default;
6970

70-
std::shared_ptr<IPortionDataChunk> BuildIndex(const THashMap<ui32, std::vector<std::shared_ptr<IPortionDataChunk>>>& data, const TIndexInfo& indexInfo) const {
71-
return DoBuildIndex(data, indexInfo);
71+
std::shared_ptr<IPortionDataChunk> BuildIndex(const THashMap<ui32, std::vector<std::shared_ptr<IPortionDataChunk>>>& data, const ui32 recordsCount, const TIndexInfo& indexInfo) const {
72+
return DoBuildIndex(data, recordsCount, indexInfo);
7273
}
7374

7475
void FillIndexCheckers(const std::shared_ptr<NRequest::TDataForIndexesCheckers>& info, const NSchemeShard::TOlapSchema& schema) const {

‎ydb/core/tx/columnshard/engines/storage/indexes/bloom/checker.h‎

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,39 @@ class TFixStringBitsStorage {
1111
: Data(data)
1212
{}
1313

14+
static ui32 GrowBitsCountToByte(const ui32 bitsCount) {
15+
const ui32 bytesCount = bitsCount / 8;
16+
return (bytesCount + ((bitsCount % 8) ? 1 : 0)) * 8;
17+
}
18+
19+
TFixStringBitsStorage(const std::vector<bool>& bitsVector)
20+
: TFixStringBitsStorage(bitsVector.size()) {
21+
ui32 byteIdx = 0;
22+
ui8 byteCurrent = 0;
23+
ui8 shiftCurrent = 0;
24+
for (ui32 i = 0; i < bitsVector.size(); ++i) {
25+
if (i && i % 8 == 0) {
26+
Data[byteIdx] = (char)byteCurrent;
27+
byteCurrent = 0;
28+
shiftCurrent = 1;
29+
++byteIdx;
30+
}
31+
if (bitsVector[i]) {
32+
byteCurrent += shiftCurrent;
33+
}
34+
shiftCurrent = (shiftCurrent << 1);
35+
}
36+
if (byteCurrent) {
37+
Data[byteIdx] = (char)byteCurrent;
38+
}
39+
}
40+
1441
ui32 GetSizeBits() const {
1542
return Data.size() * 8;
1643
}
1744

1845
TFixStringBitsStorage(const ui32 sizeBits)
19-
: Data(sizeBits / 8 + ((sizeBits % 8) ? 1 : 0), '\0') {
46+
: Data(GrowBitsCountToByte(sizeBits) / 8, '\0') {
2047
}
2148

2249
void Set(const bool val, const ui32 idx) {

‎ydb/core/tx/columnshard/engines/storage/indexes/bloom/meta.cpp‎

Lines changed: 14 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -10,26 +10,21 @@
1010

1111
namespace NKikimr::NOlap::NIndexes {
1212

13-
TString TBloomIndexMeta::DoBuildIndexImpl(TChunkedBatchReader& reader) const {
14-
std::set<ui64> hashes;
15-
{
16-
NArrow::NHash::NXX64::TStreamStringHashCalcer hashCalcer(0);
13+
TString TBloomIndexMeta::DoBuildIndexImpl(TChunkedBatchReader& reader, const ui32 recordsCount) const {
14+
const ui32 bitsCount = TFixStringBitsStorage::GrowBitsCountToByte(HashesCount * recordsCount / std::log(2));
15+
std::vector<bool> filterBits(bitsCount, false);
16+
for (ui32 i = 0; i < HashesCount; ++i) {
17+
NArrow::NHash::NXX64::TStreamStringHashCalcer_H3 hashCalcer(i);
1718
for (reader.Start(); reader.IsCorrect(); reader.ReadNext()) {
1819
hashCalcer.Start();
1920
for (auto&& i : reader) {
2021
NArrow::NHash::TXX64::AppendField(i.GetCurrentChunk(), i.GetCurrentRecordIndex(), hashCalcer);
2122
}
22-
hashes.emplace(hashCalcer.Finish());
23+
filterBits[hashCalcer.Finish() % bitsCount] = true;
2324
}
2425
}
2526

26-
const ui32 bitsCount = HashesCount * hashes.size() / std::log(2);
27-
TFixStringBitsStorage bits(bitsCount);
28-
const auto pred = [&bits](const ui64 hash) {
29-
bits.Set(true, hash % bits.GetSizeBits());
30-
};
31-
BuildHashesSet(hashes, pred);
32-
return bits.GetData();
27+
return TFixStringBitsStorage(filterBits).GetData();
3328
}
3429

3530
void TBloomIndexMeta::DoFillIndexCheckers(const std::shared_ptr<NRequest::TDataForIndexesCheckers>& info, const NSchemeShard::TOlapSchema& schema) const {
@@ -51,15 +46,14 @@ void TBloomIndexMeta::DoFillIndexCheckers(const std::shared_ptr<NRequest::TDataF
5146
continue;
5247
}
5348
std::set<ui64> hashes;
54-
const auto pred = [&hashes](const ui64 hash) {
55-
hashes.emplace(hash);
56-
};
57-
NArrow::NHash::NXX64::TStreamStringHashCalcer calcer(0);
58-
calcer.Start();
59-
for (auto&& i : foundColumns) {
60-
NArrow::NHash::TXX64::AppendField(i.second, calcer);
49+
for (ui32 i = 0; i < HashesCount; ++i) {
50+
NArrow::NHash::NXX64::TStreamStringHashCalcer_H3 calcer(i);
51+
calcer.Start();
52+
for (auto&& i : foundColumns) {
53+
NArrow::NHash::TXX64::AppendField(i.second, calcer);
54+
}
55+
hashes.emplace(calcer.Finish());
6156
}
62-
BuildHashesSet(calcer.Finish(), pred);
6357
branch->MutableIndexes().emplace_back(std::make_shared<TBloomFilterChecker>(GetIndexId(), std::move(hashes)));
6458
}
6559
}

0 commit comments

Comments
 (0)