Skip to content

test: improve coverage collection and cover storage edge cases #505

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

Merged
lxy-9602 merged 3 commits into from
Nov 4, 2026
Merged

test: improve coverage collection and cover storage edge cases #505

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5eac60f
test: cover reader and commit edge cases
lucasfang Nov 4, 2026
19adffd
test(coverage): use atomic counters and enable optional components
lucasfang Nov 4, 2026
7ccb2e7
test(coverage): cover credential refresh and cleanup edge cases
lucasfang Nov 4, 2026
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
6 changes: 4 additions & 2 deletions CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -306,8 +306,10 @@ include(san-config)

# Code coverage
if("${PAIMON_GENERATE_COVERAGE}")
set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} --coverage -DCOVERAGE_BUILD")
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} --coverage -DCOVERAGE_BUILD")
set(CMAKE_C_FLAGS
"${CMAKE_C_FLAGS} --coverage -fprofile-update=atomic -DCOVERAGE_BUILD")
set(CMAKE_CXX_FLAGS
"${CMAKE_CXX_FLAGS} --coverage -fprofile-update=atomic -DCOVERAGE_BUILD")
endif()

# CMAKE_CXX_FLAGS now fully assembled
Expand Down
2 changes: 2 additions & 0 deletions cmake_modules/ThirdpartyToolchain.cmake
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -430,6 +430,8 @@ string(REPLACE "-Wglobal-constructors" "" EP_C_FLAGS ${EP_C_FLAGS})
# Remove coverage flags from third-party dependencies to avoid gcov dependency
string(REPLACE "--coverage" "" EP_CXX_FLAGS ${EP_CXX_FLAGS})
string(REPLACE "--coverage" "" EP_C_FLAGS ${EP_C_FLAGS})
string(REPLACE " -fprofile-update=atomic" "" EP_CXX_FLAGS ${EP_CXX_FLAGS})
string(REPLACE " -fprofile-update=atomic" "" EP_C_FLAGS ${EP_C_FLAGS})
string(REPLACE "-DCOVERAGE_BUILD" "" EP_CXX_FLAGS ${EP_CXX_FLAGS})
string(REPLACE "-DCOVERAGE_BUILD" "" EP_C_FLAGS ${EP_C_FLAGS})
if(NOT MSVC_TOOLCHAIN)
Expand Down
31 changes: 27 additions & 4 deletions src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@
#include "paimon/core/global_index/global_index_file_manager.h"
#include "paimon/data/timestamp.h"
#include "paimon/global_index/bitmap_global_index_result.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/predicate/predicate_utils.h"
#include "paimon/testing/mock/mock_index_path_factory.h"
#include "paimon/testing/utils/testharness.h"

Expand Down Expand Up @@ -125,10 +127,31 @@ TEST_F(BitmapGlobalIndexTest, WriteAndReadStringPredicates) {
CheckResult(reader->VisitGreaterThan(StringLiteral("banana")), {4, 5});
CheckResult(reader->VisitGreaterOrEqual(StringLiteral("banana")), {3, 4, 5});

CheckResult(reader->VisitStartsWith(StringLiteral("ban")), {3, 4});
CheckResult(reader->VisitEndsWith(StringLiteral("le")), {0, 1});
CheckResult(reader->VisitContains(StringLiteral("an")), {3, 4});
CheckResult(reader->VisitLike(StringLiteral("ban_n_")), {3});
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<Predicate> starts,
PredicateBuilder::StartsWith(0, "f0", FieldType::STRING, StringLiteral("ban")));
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<Predicate> ends,
PredicateBuilder::EndsWith(0, "f0", FieldType::STRING, StringLiteral("le")));
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<Predicate> contains,
PredicateBuilder::Contains(0, "f0", FieldType::STRING, StringLiteral("an")));
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<Predicate> like,
PredicateBuilder::Like(0, "f0", FieldType::STRING, StringLiteral("ban_n_")));
using IndexResult = std::shared_ptr<GlobalIndexResult>;
CheckResult(PredicateUtils::VisitPredicate<IndexResult>(
std::dynamic_pointer_cast<LeafPredicate>(starts), reader),
{3, 4});
CheckResult(PredicateUtils::VisitPredicate<IndexResult>(
std::dynamic_pointer_cast<LeafPredicate>(ends), reader),
{0, 1});
CheckResult(PredicateUtils::VisitPredicate<IndexResult>(
std::dynamic_pointer_cast<LeafPredicate>(contains), reader),
{3, 4});
CheckResult(PredicateUtils::VisitPredicate<IndexResult>(
std::dynamic_pointer_cast<LeafPredicate>(like), reader),
{3});
ASSERT_NOK_WITH_MSG(reader->VisitVectorSearch(nullptr),
"Vector search is not supported in BitmapIndexReader");
ASSERT_NOK_WITH_MSG(reader->VisitFullTextSearch(nullptr),
Expand Down
179 changes: 179 additions & 0 deletions src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,12 @@

namespace paimon::test {

TEST(PrefetchFileBatchReaderTest, DefaultPreBufferRangeIsEmpty) {
MockFileBatchReader reader(nullptr, arrow::struct_({}), 1);
ASSERT_OK_AND_ASSIGN(auto ranges, reader.PrefetchFileBatchReader::PreBufferRange());
ASSERT_TRUE(ranges.empty());
}

class ControlledMockFileBatchReader : public MockFileBatchReader {
public:
ControlledMockFileBatchReader(const std::shared_ptr<arrow::Array>& data,
Expand Down Expand Up @@ -150,6 +156,21 @@ class IoReadingMockFileBatchReader : public MockFileBatchReader {
std::shared_ptr<InputStream> input_stream_;
};

class StreamCapturingMockReaderBuilder : public MockFormatReaderBuilder {
public:
using MockFormatReaderBuilder::MockFormatReaderBuilder;

Result<std::unique_ptr<FileBatchReader>> Build(
const std::shared_ptr<InputStream>& input_stream) const override {
std::lock_guard<std::mutex> lock(mutex_);
input_streams_.push_back(input_stream);
return MockFormatReaderBuilder::Build(input_stream);
}

mutable std::mutex mutex_;
mutable std::vector<std::shared_ptr<InputStream>> input_streams_;
};

class IoReadingMockFormatReaderBuilder : public ReaderBuilder {
public:
IoReadingMockFormatReaderBuilder(const std::shared_ptr<arrow::Array>& data,
Expand Down Expand Up @@ -493,6 +514,164 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestFailedIoMetrics) {
ASSERT_EQ(latency_count, read_requests);
}

TEST_F(PrefetchFileBatchReaderImplTest, TestSequentialIoMetricsAndPositionForwarding) {
const std::string path = PathUtil::JoinPath(dir_->Str(), "stream-data");
ASSERT_OK(local_fs_->WriteFile(path, "abcdef", /*overwrite=*/false));
StreamCapturingMockReaderBuilder builder(PrepareArray(1), data_type_, 1);
ASSERT_OK_AND_ASSIGN(
auto reader,
PrefetchFileBatchReaderImpl::Create(
path, 6, &builder, local_fs_, /*prefetch_max_parallel_num=*/1, /*batch_size=*/1,
/*prefetch_batch_count=*/1, /*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/false, /*read_ahead_cache_enabled=*/false, CacheConfig(),
/*enable_io_metrics=*/true, WarmupLevel::NONE, pool_, GetArrowPool(pool_)));
ASSERT_EQ(builder.input_streams_.size(), 1u);
const auto& stream = builder.input_streams_.front();
ASSERT_OK(stream->Seek(2, FS_SEEK_SET));
ASSERT_OK_AND_ASSIGN(int64_t position, stream->GetPos());
ASSERT_EQ(position, 2);
char buffer[8] = {};
ASSERT_OK_AND_ASSIGN(int64_t bytes_read, stream->Read(buffer, 4));
ASSERT_EQ(bytes_read, 4);
ASSERT_EQ(std::string(buffer, bytes_read), "cdef");
ASSERT_OK_AND_ASSIGN(position, stream->GetPos());
ASSERT_EQ(position, 6);
ASSERT_OK(stream->Seek(-2, FS_SEEK_END));
ASSERT_OK_AND_ASSIGN(bytes_read, stream->Read(buffer, 2));
ASSERT_EQ(std::string(buffer, bytes_read), "ef");
ASSERT_OK(stream->Seek(0, FS_SEEK_SET));
ASSERT_NOK(stream->Read(buffer, 8));
const auto metrics = reader->GetReaderMetrics();
ASSERT_OK_AND_ASSIGN(uint64_t requests, metrics->GetCounter(PrefetchIoMetrics::READ_REQUESTS));
ASSERT_EQ(requests, 3);
ASSERT_OK_AND_ASSIGN(uint64_t requested,
metrics->GetCounter(PrefetchIoMetrics::READ_REQUESTED_BYTES));
ASSERT_EQ(requested, 14);
ASSERT_OK_AND_ASSIGN(uint64_t physical,
metrics->GetCounter(PrefetchIoMetrics::READ_PHYSICAL_BYTES));
ASSERT_EQ(physical, 6);
ASSERT_OK_AND_ASSIGN(uint64_t failures, metrics->GetCounter(PrefetchIoMetrics::READ_FAILED));
ASSERT_EQ(failures, 1);
}

TEST_F(PrefetchFileBatchReaderImplTest, TestMinimumQueueCapacityAndFileSchema) {
auto data = PrepareArray(10);
MockFormatReaderBuilder builder(data, data_type_, 2);
ASSERT_OK_AND_ASSIGN(
auto reader,
PrefetchFileBatchReaderImpl::Create(
"", 0, &builder, mock_fs_, /*prefetch_max_parallel_num=*/3, /*batch_size=*/2,
/*prefetch_batch_count=*/1, /*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/false, CacheConfig(),
/*enable_io_metrics=*/false, WarmupLevel::NONE, pool_, GetArrowPool(pool_)));
ASSERT_EQ(reader->prefetch_queue_capacity_, 1u);
ASSERT_OK_AND_ASSIGN(std::unique_ptr<ArrowSchema> c_schema, reader->GetFileSchema());
auto imported_schema = arrow::ImportSchema(c_schema.get());
ASSERT_TRUE(imported_schema.ok()) << imported_schema.status().ToString();
std::shared_ptr<arrow::Schema> schema = std::move(imported_schema).ValueOrDie();
ASSERT_TRUE(schema->Equals(*arrow::schema(fields_)));
ASSERT_OK_AND_ASSIGN(auto result, CollectResultAndRowIds(reader.get()));
ASSERT_TRUE(result.first);
ASSERT_TRUE(result.first->Equals(*std::make_shared<arrow::ChunkedArray>(data)));
}

TEST_F(PrefetchFileBatchReaderImplTest, TestDiscardedBatchesAreCountedAndReleased) {
for (int32_t scenario : {0, 1, 2}) {
SCOPED_TRACE(scenario);
MockFormatReaderBuilder builder(PrepareArray(20), data_type_, 5);
ASSERT_OK_AND_ASSIGN(
auto reader,
PrefetchFileBatchReaderImpl::Create(
"", 0, &builder, mock_fs_, /*prefetch_max_parallel_num=*/1, /*batch_size=*/5,
/*prefetch_batch_count=*/2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/false, CacheConfig(),
/*enable_io_metrics=*/false, WarmupLevel::NONE, pool_, GetArrowPool(pool_)));
auto* source = dynamic_cast<MockFileBatchReader*>(reader->readers_.front().get());
ASSERT_TRUE(source);
if (scenario == 2) {
ASSERT_OK(reader->SetReadRanges({{0, 5}}));
ASSERT_OK(source->SetReadRanges({{10, 15}}));
}
ASSERT_OK_AND_ASSIGN(auto batch, source->NextBatchWithBitmap());
ASSERT_FALSE(BatchReader::IsEofBatch(batch));
auto imported_array = arrow::ImportArray(batch.first.first.get(), batch.first.second.get());
ASSERT_TRUE(imported_array.ok()) << imported_array.status().ToString();
std::shared_ptr<arrow::Array> array = std::move(imported_array).ValueOrDie();
if (scenario == 0) {
array = PrepareArray(0);
}
std::weak_ptr<arrow::ArrayData> lifetime = array->data();
ASSERT_TRUE(
arrow::ExportArray(*array, batch.first.first.get(), batch.first.second.get()).ok());
array.reset();
if (scenario == 1) {
batch.second = RoaringBitmap32();
}
ASSERT_FALSE(lifetime.expired());
ASSERT_OK(reader->HandleReadResult(/*reader_idx=*/0, {0, 5}, std::move(batch)));
ASSERT_TRUE(lifetime.expired());
ASSERT_TRUE(reader->prefetch_queues_.front()->empty());
ASSERT_OK_AND_ASSIGN(uint64_t discarded, reader->GetReaderMetrics()->GetCounter(
PrefetchMetrics::DISCARDED_BATCHES));
ASSERT_EQ(discarded, 1);
ASSERT_OK_AND_ASSIGN(uint64_t produced, reader->GetReaderMetrics()->GetCounter(
PrefetchMetrics::PRODUCED_BATCHES));
ASSERT_EQ(produced, 0);
}
}

TEST_F(PrefetchFileBatchReaderImplTest, TestBatchFilteredIntoLaterReadRange) {
auto data = PrepareArray(20);
RoaringBitmap32 bitmap;
bitmap.Add(12);
bitmap.Add(13);
MockFormatReaderBuilder builder(data, data_type_, bitmap, 5);
ASSERT_OK_AND_ASSIGN(
auto reader,
PrefetchFileBatchReaderImpl::Create(
"", 0, &builder, mock_fs_, /*prefetch_max_parallel_num=*/1, /*batch_size=*/5,
/*prefetch_batch_count=*/2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/false, CacheConfig(),
/*enable_io_metrics=*/false, WarmupLevel::NONE, pool_, GetArrowPool(pool_)));
auto* mock_reader = dynamic_cast<MockFileBatchReader*>(reader->readers_[0].get());
ASSERT_TRUE(mock_reader);
mock_reader->EnableRandomizeBatchSize(false);
ASSERT_OK(reader->SetReadRanges({{0, 5}, {10, 15}}));
ASSERT_OK_AND_ASSIGN(auto batch_with_bitmap, reader->NextBatchWithBitmap());
ASSERT_FALSE(BatchReader::IsEofBatch(batch_with_bitmap));
ASSERT_EQ(batch_with_bitmap.second, RoaringBitmap32::From({2, 3}));
// Row IDs address the original batch, before the selection bitmap is applied.
ASSERT_OK_AND_ASSIGN(uint64_t first_row_id, reader->GetPreviousBatchFileRowId(2));
ASSERT_OK_AND_ASSIGN(uint64_t last_row_id, reader->GetPreviousBatchFileRowId(3));
ASSERT_EQ(first_row_id, 12);
ASSERT_EQ(last_row_id, 13);
ASSERT_OK_AND_ASSIGN(auto batch, ReaderUtils::ApplyBitmapToReadBatch(
std::move(batch_with_bitmap), GetArrowPool(pool_)));
ASSERT_OK_AND_ASSIGN(auto result, ReadResultCollector::GetArray(std::move(batch)));
ASSERT_TRUE(result);
ASSERT_TRUE(result->Equals(data->Slice(12, 2)));
ASSERT_OK_AND_ASSIGN(auto eof_batch, reader->NextBatchWithBitmap());
ASSERT_TRUE(BatchReader::IsEofBatch(eof_batch));
ASSERT_OK_AND_ASSIGN(uint64_t produced,
reader->GetReaderMetrics()->GetCounter(PrefetchMetrics::PRODUCED_BATCHES));
ASSERT_EQ(produced, 1);
}

TEST_F(PrefetchFileBatchReaderImplTest, TestWorkloopStopsAfterReadError) {
MockFormatReaderBuilder builder(PrepareArray(10), data_type_, 5);
ASSERT_OK_AND_ASSIGN(
auto reader,
PrefetchFileBatchReaderImpl::Create(
"", 0, &builder, mock_fs_, /*prefetch_max_parallel_num=*/1, /*batch_size=*/5,
/*prefetch_batch_count=*/2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/false, CacheConfig(),
/*enable_io_metrics=*/false, WarmupLevel::NONE, pool_, GetArrowPool(pool_)));
reader->SetReadStatus(Status::IOError("reader stopped"));
reader->Workloop();
ASSERT_NOK_WITH_MSG(reader->GetReadStatus(), "reader stopped");
ASSERT_TRUE(reader->prefetch_queues_.front()->empty());
}

TEST_F(PrefetchFileBatchReaderImplTest, FilterReadRangesWithoutBitmap) {
std::vector<std::pair<uint64_t, uint64_t>> read_ranges = {
{0, 1000}, {1000, 2000}, {2000, 3000}, {3000, 4000}, {4000, 5000},
Expand Down
64 changes: 64 additions & 0 deletions src/paimon/core/catalog/file_system_catalog_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,70 @@ TEST(FileSystemCatalogTest, TestDatabaseExists) {
ASSERT_EQ(db_location, PathUtil::JoinPath(dir->Str(), "db1.db"));
}

TEST(FileSystemCatalogTest, TestGlobalSystemTableDiscoveryAndSchema) {
std::map<std::string, std::string> options = {{Options::FILE_SYSTEM, "local"}};
ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap(options));
auto dir = UniqueTestDirectory::Create();
ASSERT_TRUE(dir);
FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options);
ASSERT_EQ(options, catalog.GetOptions());
ASSERT_OK_AND_ASSIGN(bool exists, catalog.DatabaseExists("sys"));
ASSERT_TRUE(exists);
ASSERT_OK_AND_ASSIGN(bool system_table, catalog.IsSystemTable(Identifier("sys", "tables")));
ASSERT_TRUE(system_table);
ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("sys", "tables")));
ASSERT_TRUE(exists);
ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("sys", "missing")));
ASSERT_FALSE(exists);
ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("sys", "catalog_options")));
ASSERT_FALSE(exists);

options[CatalogOptionsSystemTable::kEnabledOption] = "true";
FileSystemCatalog enabled_catalog(core_options.GetFileSystem(), dir->Str(), options);
ASSERT_OK_AND_ASSIGN(exists, enabled_catalog.TableExists(Identifier("sys", "catalog_options")));
ASSERT_TRUE(exists);
ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> schema,
enabled_catalog.LoadTableSchema(Identifier("sys", "catalog_options")));
ASSERT_NE(nullptr, std::dynamic_pointer_cast<SystemTableSchema>(schema));
ASSERT_OK_AND_ASSIGN(std::unique_ptr<ArrowSchema> c_schema, schema->GetArrowSchema());
auto arrow_schema = arrow::ImportSchema(c_schema.get());
ASSERT_TRUE(arrow_schema.ok()) << arrow_schema.status().ToString();
ASSERT_TRUE(arrow_schema.ValueOrDie()->Equals(*arrow::schema(
{arrow::field("key", arrow::utf8(), false), arrow::field("value", arrow::utf8(), false)})));
}

TEST(FileSystemCatalogTest, TestTableDiscoveryWithoutSchemaZero) {
std::map<std::string, std::string> options = {{Options::FILE_SYSTEM, "local"}};
ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap(options));
auto dir = UniqueTestDirectory::Create();
ASSERT_TRUE(dir);
std::shared_ptr<FileSystem> fs = core_options.GetFileSystem();
FileSystemCatalog catalog(fs, dir->Str(), options);
ASSERT_OK(catalog.CreateDatabase("db1", {}, false));
Identifier identifier("db1", "evolved");
ASSERT_OK_AND_ASSIGN(std::string table_path, catalog.GetTableLocation(identifier));
SchemaManager schema_manager(fs, table_path);
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
TableSchema::Create(3, arrow::schema({arrow::field("id", arrow::int32())}),
{}, {}, options));
ASSERT_OK_AND_ASSIGN(std::string json, schema->ToJsonString());
ASSERT_OK(fs->AtomicStore(schema_manager.ToSchemaPath(3), json));
// An incomplete table directory must not appear in listings alongside the evolved table.
std::string incomplete_path = PathUtil::JoinPath(dir->Str(), "db1.db/incomplete");
ASSERT_OK(fs->Mkdirs(incomplete_path));
ASSERT_OK_AND_ASSIGN(std::vector<std::string> tables, catalog.ListTables("db1"));
ASSERT_EQ((std::vector<std::string>{"evolved"}), tables);
ASSERT_OK_AND_ASSIGN(bool exists, catalog.TableExists(identifier));
ASSERT_TRUE(exists);
ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> loaded, catalog.LoadTableSchema(identifier));
std::shared_ptr<DataSchema> data_schema = std::dynamic_pointer_cast<DataSchema>(loaded);
ASSERT_NE(nullptr, data_schema);
ASSERT_EQ(3, data_schema->Id());
ASSERT_OK(catalog.DropTable(Identifier("db1", "incomplete"), true));
ASSERT_OK_AND_ASSIGN(exists, fs->Exists(incomplete_path));
ASSERT_TRUE(exists);
}

TEST(FileSystemCatalogTest, TestInvalidCreateDatabase) {
{
std::map<std::string, std::string> options;
Expand Down
Loading
Loading