Skip to content

Commit e0a15dd

Browse files
kungakunga
authored
Forward cache simplify (#2514)
1 parent 3a6ff80 commit e0a15dd

File tree

6 files changed

+117
-188
lines changed

6 files changed

+117
-188
lines changed

‎ydb/core/tablet_flat/flat_fwd_cache.h

Lines changed: 70 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,13 @@
11
#pragma once
22

33
#include "flat_part_iface.h"
4-
#include "flat_part_forward.h"
54
#include "flat_fwd_iface.h"
65
#include "flat_fwd_misc.h"
76
#include "flat_fwd_page.h"
7+
#include "flat_part_index_iter.h"
8+
#include "flat_part_index_iter_iface.h"
9+
#include "flat_table_part.h"
10+
#include "flat_part_slice.h"
811

912
namespace NKikimr {
1013
namespace NTable {
@@ -58,8 +61,16 @@ namespace NFwd {
5861
TCache() = delete;
5962

6063
TCache(const TPart* part, IPages* env, TGroupId groupId, const TIntrusiveConstPtr<TSlices>& bounds = nullptr)
61-
: Index(part, env, groupId, 1, bounds)
62-
{ }
64+
: Index(MakeHolder<TPartIndexIt>(part, env, groupId)) // TODO: use CreateIndexIter(part, env, groupId)
65+
{
66+
if (bounds && !bounds->empty()) {
67+
BeginRowId = bounds->front().BeginRowId();
68+
EndRowId = bounds->back().EndRowId();
69+
} else {
70+
BeginRowId = 0;
71+
EndRowId = Index->GetEndRowId();
72+
}
73+
}
6374

6475
~TCache()
6576
{
@@ -70,23 +81,35 @@ namespace NFwd {
7081

7182
TResult Handle(IPageLoadingQueue *head, TPageId pageId, ui64 lower) noexcept override
7283
{
73-
Y_ABORT_UNLESS(pageId != Max<TPageId>(), "Invalid requested pageId");
84+
Y_ABORT_UNLESS(pageId != Max<TPageId>(), "Requested page is invalid");
7485

7586
if (auto *page = Trace.Get(pageId)) {
7687
return { page, false, true };
7788
}
7889

79-
Rewind(pageId); /* points Offset to pageId */
90+
DropPagesBefore(pageId);
8091
Shrink();
8192

82-
bool more = Grow && (OnHold + OnFetch <= lower);
93+
bool grow = OnHold + OnFetch <= lower;
94+
95+
if (Offset == Pages.size()) { // isn't processed yet
96+
SyncIndex(pageId);
97+
AddToQueue(head, pageId);
98+
}
99+
100+
grow &= Index->IsValid() && Index->GetRowId() < EndRowId;
83101

84-
return { Preload(head, 0).Touch(pageId, Stat), more, true };
102+
return {Pages.at(Offset).Touch(pageId, Stat), grow, true};
85103
}
86104

87105
void Forward(IPageLoadingQueue *head, ui64 upper) noexcept override
88106
{
89-
Preload(head, upper);
107+
Y_ABORT_UNLESS(Started, "Couldn't be called before Handle returns grow");
108+
109+
while (OnHold + OnFetch < upper && Index->IsValid() && Index->GetRowId() < EndRowId) {
110+
AddToQueue(head, Index->GetPageId());
111+
Y_ABORT_UNLESS(Index->Next() != EReady::Page);
112+
}
90113
}
91114

92115
void Apply(TArrayRef<NPageCollection::TLoadedPage> loaded) noexcept override
@@ -117,35 +140,16 @@ namespace NFwd {
117140
}
118141

119142
private:
120-
TPage& Preload(IPageLoadingQueue *head, ui64 upper) noexcept
121-
{
122-
auto until = [this, upper]() {
123-
return OnHold + OnFetch < upper ? Max<TPageId>() : 0;
124-
};
125-
126-
while (auto more = Index.More(until())) {
127-
auto size = head->AddToQueue(more, EPage::DataPage);
128-
129-
Stat.Fetch += size;
130-
OnFetch += size;
131-
132-
Pages.emplace_back(more, size, 0, Max<TPageId>());
133-
Pages.back().Fetch = EFetch::Wait;
134-
}
135-
136-
Grow = Grow && Index.On(true) < Max<TPageId>();
137-
138-
return Pages.at(Offset);
139-
}
140-
141-
void Rewind(TPageId pageId) noexcept
143+
void DropPagesBefore(TPageId pageId) noexcept
142144
{
143-
while (auto drop = Index.Clean(pageId)) {
145+
while (Offset < Pages.size()) {
144146
auto &page = Pages.at(Offset);
145147

146-
if (!Pages || page.PageId != drop.PageId) {
147-
Y_ABORT("Dropping page that is not exist in cache");
148-
} else if (page.Size == 0) {
148+
if (page.PageId >= pageId) {
149+
break;
150+
}
151+
152+
if (page.Size == 0) {
149153
Y_ABORT("Dropping page that has not been touched");
150154
} else if (page.Usage == EUsage::Keep && page) {
151155
OnHold -= Trace.Emplace(page);
@@ -166,13 +170,42 @@ namespace NFwd {
166170
}
167171
}
168172

173+
void SyncIndex(TPageId pageId) noexcept
174+
{
175+
if (!Started) {
176+
Y_ABORT_UNLESS(Index->Seek(BeginRowId) == EReady::Data);
177+
Y_ABORT_UNLESS(Index->GetPageId() <= pageId, "Requested page is out of slice bounds");
178+
Started = true;
179+
}
180+
181+
while (Index->IsValid() && Index->GetPageId() < pageId) {
182+
Y_ABORT_UNLESS(Index->Next() == EReady::Data);
183+
Y_ABORT_UNLESS(Index->GetRowId() < EndRowId, "Requested page is out of slice bounds");
184+
}
185+
186+
Y_ABORT_UNLESS(Index->GetPageId() == pageId, "Requested page doesn't belong to the part");
187+
Y_ABORT_UNLESS(Index->Next() != EReady::Page);
188+
}
189+
190+
void AddToQueue(IPageLoadingQueue *head, TPageId pageId) noexcept
191+
{
192+
auto size = head->AddToQueue(pageId, EPage::DataPage);
193+
194+
Stat.Fetch += size;
195+
OnFetch += size;
196+
197+
Y_ABORT_UNLESS(!Pages || Pages.back().PageId < pageId);
198+
Pages.emplace_back(pageId, size, 0, Max<TPageId>());
199+
Pages.back().Fetch = EFetch::Wait;
200+
}
201+
169202
private:
170-
bool Grow = true; /* Have some pages for Forward(...) */
171-
TForward Index;
203+
THolder<IIndexIter> Index; /* Points on next to load page */
204+
bool Started = false;
205+
TRowId BeginRowId, EndRowId;
172206
TLoadedPagesCircularBuffer<TPart::Trace> Trace;
173207

174208
/*_ Forward cache line state */
175-
176209
ui64 OnHold = 0;
177210
ui64 OnFetch = 0;
178211
ui32 Offset = 0;

‎ydb/core/tablet_flat/flat_fwd_env.h

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,10 +10,8 @@
1010
#include "flat_table_subset.h"
1111
#include "flat_part_iface.h"
1212
#include "flat_part_store.h"
13-
#include "flat_sausage_packet.h"
1413
#include "flat_sausage_fetch.h"
1514
#include "util_fmt_abort.h"
16-
#include "util_basics.h"
1715
#include <util/generic/deque.h>
1816
#include <util/random/random.h>
1917
#include <util/generic/intrlist.h>

‎ydb/core/tablet_flat/flat_part_forward.h

Lines changed: 0 additions & 135 deletions
This file was deleted.

‎ydb/core/tablet_flat/test/libs/table/test_part.h

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,15 @@
11
#pragma once
22

33
#include "test_store.h"
4+
#include <ydb/core/tablet_flat/flat_fwd_blobs.h>
5+
#include <ydb/core/tablet_flat/flat_fwd_cache.h>
6+
#include <ydb/core/tablet_flat/flat_part_iface.h>
7+
#include <ydb/core/tablet_flat/flat_part_index_iter.h>
8+
#include <ydb/core/tablet_flat/flat_part_laid.h>
9+
#include <ydb/core/tablet_flat/flat_row_scheme.h>
410
#include <ydb/core/tablet_flat/flat_table_misc.h>
511
#include <ydb/core/tablet_flat/flat_table_part.h>
612
#include <ydb/core/tablet_flat/flat_table_subset.h>
7-
#include <ydb/core/tablet_flat/flat_part_laid.h>
8-
#include <ydb/core/tablet_flat/flat_part_iface.h>
9-
#include <ydb/core/tablet_flat/flat_fwd_cache.h>
10-
#include <ydb/core/tablet_flat/flat_fwd_blobs.h>
11-
#include <ydb/core/tablet_flat/flat_row_scheme.h>
1213
#include <ydb/core/tablet_flat/util_fmt_abort.h>
1314

1415
#include <util/generic/cast.h>

‎ydb/core/tablet_flat/test/libs/table/test_steps.h

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
#pragma once
22

3+
#include "util/string/builder.h"
34
#include <util/system/yassert.h>
45
#include <util/stream/output.h>
56

@@ -21,13 +22,15 @@ namespace NTest {
2122

2223
IOutputStream& Log() const noexcept
2324
{
24-
Cerr << "On " << Seq << ": ";
25+
Cerr << CurrentStepStr() << ": ";
2526

2627
return Cerr;
2728
}
2829

2930
size_t CurrentStep() const noexcept { return Seq; }
3031

32+
TString CurrentStepStr() const noexcept { return TStringBuilder() << "On " << CurrentStep(); }
33+
3134
private:
3235
size_t Seq = 0;
3336
};

0 commit comments

Comments
 (0)