Skip to content

Commit 4c2277c

Browse files
authored
feat(parquet): support configurable bitmap row-range refining strategies (coalesce and trim) (#424)
1 parent 4faf0e9 commit 4c2277c

5 files changed

Lines changed: 356 additions & 39 deletions

File tree

src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp

Lines changed: 201 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -136,15 +136,12 @@ class PageFilteredRowGroupReaderTest : public ::testing::Test {
136136
const std::shared_ptr<Predicate>& predicate,
137137
const RoaringBitmap32& bitmap,
138138
std::shared_ptr<arrow::ChunkedArray>* out,
139-
int32_t batch_size = 1024,
140-
bool enable_page_level_filter = true) {
139+
const std::map<std::string, std::string> options = {},
140+
int32_t batch_size = 1024) {
141141
ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_name));
142142
ASSERT_OK_AND_ASSIGN(int64_t length, in->Length());
143143
auto in_stream = std::make_shared<ArrowInputStreamAdapter>(in, arrow_pool_, length);
144144

145-
std::map<std::string, std::string> options;
146-
options[PARQUET_READ_ENABLE_PAGE_INDEX_FILTER] =
147-
enable_page_level_filter ? "true" : "false";
148145
ASSERT_OK_AND_ASSIGN(auto batch_reader,
149146
ParquetFileBatchReader::Create(std::move(in_stream), options,
150147
batch_size, nullptr, arrow_pool_));
@@ -1290,15 +1287,16 @@ TEST_F(PageFilteredRowGroupReaderTest, BitmapAllPagesSomeRowGroups) {
12901287
/// Test: bitmap hits partial pages of a row group (no predicate).
12911288
///
12921289
/// 200 rows, 10 rows per page, 100 rows per row group → 2 row groups.
1293-
/// Bitmap: {30..59} hits pages 3-5 of RG0 (rows 30-59), RG1 excluded.
1294-
/// Expected: 30 rows (30-59).
1295-
TEST_F(PageFilteredRowGroupReaderTest, BitmapPartialPagesSingleRowGroup) {
1290+
/// Bitmap: {90..109} hits page 9 of RG0 (rows 90-99), and page 0 of RG1 (rows 100-109).
1291+
/// Expected: 20 rows (90-109).
1292+
TEST_F(PageFilteredRowGroupReaderTest, BitmapPartialPagesAcrossRowGroups) {
12961293
std::string file_name = dir_->Str() + "/bitmap_partial_pages_rg.parquet";
12971294
auto data = MakeSequentialIntData(200);
12981295
WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/100);
12991296

13001297
RoaringBitmap32 bitmap;
1301-
bitmap.AddRange(90, 110); // hits pages 3-5 of RG0
1298+
// {90..109} hits page 9 of RG0 (rows 90-99), and page 0 of RG1 (rows 100-109).
1299+
bitmap.AddRange(90, 110);
13021300

13031301
auto read_schema = arrow::schema({arrow::field("val", arrow::int32())});
13041302
std::shared_ptr<arrow::ChunkedArray> result;
@@ -1364,8 +1362,10 @@ TEST_F(PageFilteredRowGroupReaderTest, BitmapWithPageFilteredOptionDisabled) {
13641362

13651363
auto read_schema = arrow::schema({arrow::field("val", arrow::int32())});
13661364
std::shared_ptr<arrow::ChunkedArray> result;
1365+
std::map<std::string, std::string> options;
1366+
options[PARQUET_READ_ENABLE_PAGE_INDEX_FILTER] = "false";
13671367
ReadWithPredicateAndBitmapImpl(file_name, read_schema, /*predicate=*/nullptr, bitmap, &result,
1368-
1024, false);
1368+
options);
13691369
ASSERT_TRUE(result);
13701370
ASSERT_EQ(100, result->length());
13711371

@@ -1487,42 +1487,217 @@ TEST_F(PageFilteredRowGroupReaderTest, BitmapMixedWithPredicate) {
14871487
}
14881488
}
14891489

1490-
/// Test: read parquet with scattered bitmap
1490+
/// Test: coalesce strategy with default hole_size_limit (32).
14911491
///
14921492
/// 200 rows, 50 rows per page, 100 rows per row group → 2 row groups.
1493-
/// Bitmap: [20,30), [35, 40), [125, 126), [130, 131), [150, 200)
1494-
/// To test if unneeded row at the start and end of pages are filtered out.
1495-
/// Expected: 76 rows ([20, 40) + [125, 131) + [150, 200).
1496-
TEST_F(PageFilteredRowGroupReaderTest, ScatteredBitmapTest) {
1493+
/// Bitmap: [0,10), [45, 50), [60, 70).
1494+
/// - [0,10) and [45,50) are both in RG0 page 0 ([0,49]); gap = 35 > 32, NOT merged.
1495+
/// - [45,50) and [60,70) straddle RG0 page 0 ([0,49]) and page 1 ([50,99]); gap = 10 <= 32,
1496+
/// merged across the page boundary, so rows 50-59 are read even though not in the bitmap.
1497+
/// Expected: 35 rows ([0,10) + [45, 70)).
1498+
TEST_F(PageFilteredRowGroupReaderTest, BitmapCoalesceTest) {
14971499
std::string file_name = dir_->Str() + "/scattered_bitmap.parquet";
14981500
auto data = MakeSequentialIntData(200);
14991501
WriteTestFile(file_name, data, /*write_batch_size=*/50, /*max_row_group_length=*/100);
15001502

15011503
RoaringBitmap32 bitmap;
1502-
bitmap.AddRange(20, 30);
1503-
bitmap.AddRange(35, 40);
1504-
bitmap.Add(125);
1505-
bitmap.Add(130);
1506-
bitmap.AddRange(150, 200);
1504+
bitmap.AddRange(0, 10);
1505+
bitmap.AddRange(45, 50);
1506+
bitmap.AddRange(60, 70);
15071507

15081508
auto read_schema = arrow::schema({arrow::field("val", arrow::int32())});
15091509
std::shared_ptr<arrow::ChunkedArray> result;
15101510
ReadWithPredicateAndBitmapImpl(file_name, read_schema, /*predicate=*/nullptr, bitmap, &result);
15111511
ASSERT_TRUE(result);
1512-
ASSERT_EQ(76, result->length());
1512+
ASSERT_EQ(35, result->length());
15131513

15141514
auto flat = arrow::Concatenate(result->chunks()).ValueOrDie();
15151515
auto struct_arr = std::dynamic_pointer_cast<arrow::StructArray>(flat);
15161516
ASSERT_TRUE(struct_arr);
15171517
auto val_arr = std::dynamic_pointer_cast<arrow::Int32Array>(struct_arr->field(0));
1518-
for (int32_t i = 0; i < 20; ++i) {
1519-
ASSERT_EQ(20 + i, val_arr->Value(i));
1518+
// [0, 9] — not merged (gap to next range = 35 > 32)
1519+
for (int32_t i = 0; i < 10; ++i) {
1520+
ASSERT_EQ(0 + i, val_arr->Value(i));
1521+
}
1522+
// [45, 69] — merged across page boundary (gap = 10 <= 32)
1523+
for (int32_t i = 0; i < 25; ++i) {
1524+
ASSERT_EQ(45 + i, val_arr->Value(10 + i));
1525+
}
1526+
}
1527+
1528+
/// Test: coalesce strategy with hole_size_limit=5 (instead of default 32).
1529+
///
1530+
/// Same bitmap as BitmapCoalesceTest: [0,10), [45, 50), [60, 70).
1531+
/// Both gaps (35 and 10) exceed 5, so no ranges are merged.
1532+
/// Expected: 25 rows (exact bitmap selection, no holes filled).
1533+
/// Compare with BitmapCoalesceTest which gets 35 rows (gap of 10 is filled with default limit=32).
1534+
TEST_F(PageFilteredRowGroupReaderTest, BitmapCoalesceSmallHoleSizeTest) {
1535+
std::string file_name = dir_->Str() + "/coalesce_small_hole.parquet";
1536+
auto data = MakeSequentialIntData(200);
1537+
WriteTestFile(file_name, data, /*write_batch_size=*/50, /*max_row_group_length=*/100);
1538+
1539+
RoaringBitmap32 bitmap;
1540+
bitmap.AddRange(0, 10);
1541+
bitmap.AddRange(45, 50);
1542+
bitmap.AddRange(60, 70);
1543+
1544+
std::map<std::string, std::string> options;
1545+
options[PARQUET_READ_ROW_RANGES_COALESCE_HOLE_SIZE_LIMIT] = "5";
1546+
1547+
auto read_schema = arrow::schema({arrow::field("val", arrow::int32())});
1548+
std::shared_ptr<arrow::ChunkedArray> result;
1549+
ReadWithPredicateAndBitmapImpl(file_name, read_schema, /*predicate=*/nullptr, bitmap, &result,
1550+
options);
1551+
ASSERT_TRUE(result);
1552+
ASSERT_EQ(25, result->length());
1553+
1554+
auto flat = arrow::Concatenate(result->chunks()).ValueOrDie();
1555+
auto struct_arr = std::dynamic_pointer_cast<arrow::StructArray>(flat);
1556+
ASSERT_TRUE(struct_arr);
1557+
auto val_arr = std::dynamic_pointer_cast<arrow::Int32Array>(struct_arr->field(0));
1558+
ASSERT_TRUE(val_arr);
1559+
// [0, 9] — not merged (gap = 35 > 5)
1560+
for (int32_t i = 0; i < 10; ++i) {
1561+
ASSERT_EQ(0 + i, val_arr->Value(i));
1562+
}
1563+
// [45, 49] — not merged (gap = 10 > 5)
1564+
for (int32_t i = 0; i < 5; ++i) {
1565+
ASSERT_EQ(45 + i, val_arr->Value(10 + i));
1566+
}
1567+
// [60, 69] — not merged
1568+
for (int32_t i = 0; i < 10; ++i) {
1569+
ASSERT_EQ(60 + i, val_arr->Value(15 + i));
1570+
}
1571+
}
1572+
1573+
/// Test: trim strategy for bitmap row filtering.
1574+
///
1575+
/// Same bitmap as BitmapCoalesceTest: [0,10), [45, 50), [60, 70).
1576+
/// Trim produces one range per page (trimmed to first/last selected row in that page):
1577+
/// - RG0 page 0 ([0,49]): first selected = 0, last selected = 49 → [0, 49] (50 rows, includes
1578+
/// the 35-row gap 10-44 that coalesce keeps as a hole because gap = 35 > 32).
1579+
/// - RG0 page 1 ([50,99]): first selected = 60, last selected = 69 → [60, 69] (10 rows; rows
1580+
/// 50-59 are NOT read, unlike coalesce which merges across the page boundary).
1581+
/// Expected: 60 rows ([0, 50) + [60, 70)).
1582+
/// Compare with BitmapCoalesceTest which gets 35 rows.
1583+
TEST_F(PageFilteredRowGroupReaderTest, BitmapTrimStrategyTest) {
1584+
std::string file_name = dir_->Str() + "/trim_strategy.parquet";
1585+
auto data = MakeSequentialIntData(200);
1586+
WriteTestFile(file_name, data, /*write_batch_size=*/50, /*max_row_group_length=*/100);
1587+
1588+
RoaringBitmap32 bitmap;
1589+
bitmap.AddRange(0, 10);
1590+
bitmap.AddRange(45, 50);
1591+
bitmap.AddRange(60, 70);
1592+
1593+
std::map<std::string, std::string> options;
1594+
options[PARQUET_READ_BITMAP_ROW_RANGE_REFINING_STRATEGY] = "trim";
1595+
1596+
auto read_schema = arrow::schema({arrow::field("val", arrow::int32())});
1597+
std::shared_ptr<arrow::ChunkedArray> result;
1598+
ReadWithPredicateAndBitmapImpl(file_name, read_schema, /*predicate=*/nullptr, bitmap, &result,
1599+
options);
1600+
ASSERT_TRUE(result);
1601+
ASSERT_EQ(60, result->length());
1602+
1603+
auto flat = arrow::Concatenate(result->chunks()).ValueOrDie();
1604+
auto struct_arr = std::dynamic_pointer_cast<arrow::StructArray>(flat);
1605+
ASSERT_TRUE(struct_arr);
1606+
auto val_arr = std::dynamic_pointer_cast<arrow::Int32Array>(struct_arr->field(0));
1607+
ASSERT_TRUE(val_arr);
1608+
// RG0 page 0 trimmed to [0, 49] — includes gap 10-44 that coalesce skips
1609+
for (int32_t i = 0; i < 50; ++i) {
1610+
ASSERT_EQ(0 + i, val_arr->Value(i));
15201611
}
1521-
for (int32_t i = 0; i < 6; ++i) {
1522-
ASSERT_EQ(125 + i, val_arr->Value(20 + i));
1612+
// RG0 page 1 trimmed to [60, 69] — rows 50-59 not read (unlike coalesce)
1613+
for (int32_t i = 0; i < 10; ++i) {
1614+
ASSERT_EQ(60 + i, val_arr->Value(50 + i));
15231615
}
1616+
}
1617+
1618+
/// Test: invalid strategy value returns Status::Invalid.
1619+
///
1620+
/// 200 rows, 50 rows per page, 100 rows per row group → 2 row groups.
1621+
/// Bitmap: [0,10), [45, 50), [60, 70) — same as BitmapCoalesceTest.
1622+
/// Strategy: "invalid" (not one of "coalesce", "trim", "none").
1623+
/// Expected: SetReadSchema returns Status::Invalid.
1624+
TEST_F(PageFilteredRowGroupReaderTest, BitmapInvalidStrategyTest) {
1625+
std::string file_name = dir_->Str() + "/invalid_strategy.parquet";
1626+
auto data = MakeSequentialIntData(200);
1627+
WriteTestFile(file_name, data, /*write_batch_size=*/50, /*max_row_group_length=*/100);
1628+
1629+
RoaringBitmap32 bitmap;
1630+
bitmap.AddRange(0, 10);
1631+
bitmap.AddRange(45, 50);
1632+
bitmap.AddRange(60, 70);
1633+
1634+
std::map<std::string, std::string> options;
1635+
options[PARQUET_READ_BITMAP_ROW_RANGE_REFINING_STRATEGY] = "invalid";
1636+
1637+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_name));
1638+
ASSERT_OK_AND_ASSIGN(int64_t length, in->Length());
1639+
auto in_stream = std::make_shared<ArrowInputStreamAdapter>(in, arrow_pool_, length);
1640+
1641+
ASSERT_OK_AND_ASSIGN(
1642+
auto batch_reader,
1643+
ParquetFileBatchReader::Create(std::move(in_stream), options, 1024, nullptr, arrow_pool_));
1644+
1645+
auto read_schema = arrow::schema({arrow::field("val", arrow::int32())});
1646+
auto c_schema = std::make_unique<ArrowSchema>();
1647+
ASSERT_TRUE(arrow::ExportSchema(*read_schema, c_schema.get()).ok());
1648+
1649+
auto status = batch_reader->SetReadSchema(c_schema.get(), /*predicate=*/nullptr, bitmap);
1650+
ASSERT_FALSE(status.ok());
1651+
ASSERT_TRUE(status.IsInvalid());
1652+
}
1653+
1654+
/// Test: trim strategy with multiple columns — intersection of per-column trimmed ranges.
1655+
///
1656+
/// 200 rows, 50 rows per page, 100 rows per row group → 2 row groups.
1657+
/// Two columns: a[i] = i, b[i] = i * 10.
1658+
/// Bitmap: [0,10), [45, 50), [60, 70) — same as BitmapTrimStrategyTest.
1659+
/// Both columns share the same page boundaries, so their trimmed ranges are identical.
1660+
/// The intersection is the same as either column alone:
1661+
/// - RG0 page 0 ([0,49]): trimmed to [0, 49] (50 rows)
1662+
/// - RG0 page 1 ([50,99]): trimmed to [60, 69] (10 rows)
1663+
/// Expected: 60 rows. Both columns must remain aligned after trimming.
1664+
TEST_F(PageFilteredRowGroupReaderTest, BitmapTrimMultiColumnTest) {
1665+
std::string file_name = dir_->Str() + "/trim_multi_col.parquet";
1666+
auto data = MakeTwoColumnData(200);
1667+
WriteTestFile(file_name, data, /*write_batch_size=*/50, /*max_row_group_length=*/100);
1668+
1669+
RoaringBitmap32 bitmap;
1670+
bitmap.AddRange(0, 10);
1671+
bitmap.AddRange(45, 50);
1672+
bitmap.AddRange(60, 70);
1673+
1674+
std::map<std::string, std::string> options;
1675+
options[PARQUET_READ_BITMAP_ROW_RANGE_REFINING_STRATEGY] = "trim";
1676+
1677+
auto read_schema =
1678+
arrow::schema({arrow::field("a", arrow::int32()), arrow::field("b", arrow::int32())});
1679+
std::shared_ptr<arrow::ChunkedArray> result;
1680+
ReadWithPredicateAndBitmapImpl(file_name, read_schema, /*predicate=*/nullptr, bitmap, &result,
1681+
options);
1682+
ASSERT_TRUE(result);
1683+
ASSERT_EQ(60, result->length());
1684+
1685+
auto flat = arrow::Concatenate(result->chunks()).ValueOrDie();
1686+
auto struct_arr = std::dynamic_pointer_cast<arrow::StructArray>(flat);
1687+
ASSERT_TRUE(struct_arr);
1688+
auto a_arr = std::dynamic_pointer_cast<arrow::Int32Array>(struct_arr->field(0));
1689+
auto b_arr = std::dynamic_pointer_cast<arrow::Int32Array>(struct_arr->field(1));
1690+
ASSERT_TRUE(a_arr);
1691+
ASSERT_TRUE(b_arr);
1692+
// RG0 page 0 trimmed to [0, 49]
15241693
for (int32_t i = 0; i < 50; ++i) {
1525-
ASSERT_EQ(150 + i, val_arr->Value(26 + i));
1694+
ASSERT_EQ(i, a_arr->Value(i));
1695+
ASSERT_EQ(i * 10, b_arr->Value(i));
1696+
}
1697+
// RG0 page 1 trimmed to [60, 69]
1698+
for (int32_t i = 0; i < 10; ++i) {
1699+
ASSERT_EQ(60 + i, a_arr->Value(50 + i));
1700+
ASSERT_EQ((60 + i) * 10, b_arr->Value(50 + i));
15261701
}
15271702
}
15281703

0 commit comments

Comments
 (0)