Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

ignore delmark when add minmax for pk column (#4746) #4757

Merged
merged 4 commits into from
Jun 21, 2022
Merged

ignore delmark when add minmax for pk column (#4746) #4757

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
147e537
This is an automated cherry-pick of #4746
lidezhu Apr 25, 2022
c26dc7f
Update dm_basic_include.h
lidezhu Apr 25, 2022
a6370f8
Update gtest_dm_segment.cpp
lidezhu Apr 25, 2022
26f63a9
Merge branch 'release-5.1' into cherry-pick-4746-to-release-5.1
lidezhu Jun 21, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions dbms/src/Storages/DeltaMerge/File/DMFileWriter.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,9 @@ void DMFileWriter::writeColumn(ColId col_id, const IDataType & type, const IColu
auto & minmax_indexs = single_file_stream->minmax_indexs;
if (auto iter = minmax_indexs.find(stream_name); iter != minmax_indexs.end())
{
iter->second->addPack(column, del_mark);
// For EXTRA_HANDLE_COLUMN_ID, we ignore del_mark when add minmax index.
// Because we need all rows which satisfy a certain range when place delta index no matter whether the row is a delete row.
iter->second->addPack(column, col_id == EXTRA_HANDLE_COLUMN_ID ? nullptr : del_mark);
}

auto offset_in_compressed_block = single_file_stream->original_hashing.offset();
Expand Down Expand Up @@ -217,7 +219,11 @@ void DMFileWriter::writeColumn(ColId col_id, const IDataType & type, const IColu
const auto name = DMFile::getFileNameBase(col_id, substream);
auto & stream = column_streams.at(name);
if (stream->minmaxes)
stream->minmaxes->addPack(column, del_mark);
{
// For EXTRA_HANDLE_COLUMN_ID, we ignore del_mark when add minmax index.
// Because we need all rows which satisfy a certain range when place delta index no matter whether the row is a delete row.
stream->minmaxes->addPack(column, col_id == EXTRA_HANDLE_COLUMN_ID ? nullptr : del_mark);
}

/// There could already be enough data to compress into the new block.
if (stream->original_hashing.offset() >= options.min_compress_block_size)
Expand Down
5 changes: 3 additions & 2 deletions dbms/src/Storages/DeltaMerge/tests/dm_basic_include.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,9 +201,10 @@ class DMTestEnv
* @param ts_beg `timestamp`'s value begin
* @param ts_end `timestamp`'s value end (not included)
* @param reversed increasing/decreasing insert `timestamp`'s value
* @param deleted if deleted is false, set `tag` to 0; otherwise set `tag` to 1
* @return
*/
static Block prepareBlockWithTso(Int64 pk, size_t ts_beg, size_t ts_end, bool reversed = false)
static Block prepareBlockWithTso(Int64 pk, size_t ts_beg, size_t ts_end, bool reversed = false, bool deleted = false)
{
Block block;
const size_t num_rows = (ts_end - ts_beg);
Expand Down Expand Up @@ -240,7 +241,7 @@ class DMTestEnv
column_data.resize(num_rows);
for (size_t i = 0; i < num_rows; ++i)
{
column_data[i] = 0;
column_data[i] = deleted ? 1 : 0;
}
tag_col.column = std::move(m_col);
}
Expand Down
36 changes: 36 additions & 0 deletions dbms/src/Storages/DeltaMerge/tests/gtest_dm_segment.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,6 +212,42 @@ try
}
CATCH

TEST_F(Segment_test, WriteRead2)
try
{
const size_t num_rows_write = dmContext().stable_pack_rows;
{
// write a block with rows all deleted
Block block = DMTestEnv::prepareBlockWithTso(2, 100, 100 + num_rows_write, false, true);
segment->write(dmContext(), block);
// write not deleted rows with larger pk
Block block2 = DMTestEnv::prepareBlockWithTso(3, 100, 100 + num_rows_write, false, false);
segment->write(dmContext(), block2);

// flush segment and make sure there is two packs in stable
segment = segment->mergeDelta(dmContext(), tableColumns());
ASSERT_EQ(segment->getStable()->getPacks(), 2);
}

{
Block block = DMTestEnv::prepareBlockWithTso(1, 100, 100 + num_rows_write, false, false);
segment->write(dmContext(), block);
}

{
auto in = segment->getInputStream(dmContext(), *tableColumns(), {RowKeyRange::newAll(false, 1)});
size_t num_rows_read = 0;
in->readPrefix();
while (Block block = in->read())
{
num_rows_read += block.rows();
}
in->readSuffix();
// only write two visible pks
ASSERT_EQ(num_rows_read, 2);
}
}
CATCH

TEST_F(Segment_test, ReadWithMoreAdvacedDeltaIndex)
try
Expand Down