Skip to content

Commit e24479a

Browse files
author
zhongzheyun
committed
perf(manifest): selectively decode bucket entries with bounded block indexes
1 parent ded7cfb commit e24479a

10 files changed

Lines changed: 529 additions & 67 deletions

‎src/paimon/core/manifest/manifest_file.cpp‎

Lines changed: 77 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,10 +22,12 @@
2222
#include <limits>
2323
#include <utility>
2424

25+
#include "arrow/array/array_primitive.h"
2526
#include "arrow/c/abi.h"
2627
#include "arrow/c/bridge.h"
2728
#include "paimon/common/data/columnar/columnar_row.h"
2829
#include "paimon/common/utils/arrow/status_utils.h"
30+
#include "paimon/common/utils/checked_cast.h"
2931
#include "paimon/core/io/rolling_file_writer.h"
3032
#include "paimon/core/manifest/manifest_entry.h"
3133
#include "paimon/core/manifest/manifest_entry_serializer.h"
@@ -106,7 +108,81 @@ Status ManifestFile::ReadBucketEntries(const std::string& file_name, int32_t buc
106108
entries->push_back(std::move(entry));
107109
}
108110
return Status::OK();
109-
});
111+
},
112+
[this, bucket](FileBatchReader* reader) { return PrepareBucketRead(reader, bucket); });
113+
}
114+
115+
Status ManifestFile::PrepareBucketRead(FileBatchReader* reader, int32_t bucket) const {
116+
if (!reader->SupportPreciseBitmapSelection()) {
117+
return Status::OK();
118+
}
119+
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ArrowSchema> c_schema, reader->GetFileSchema());
120+
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> file_schema,
121+
arrow::ImportSchema(c_schema.get()));
122+
const auto& target_type = serializer_->GetDataType();
123+
const std::string& version_name = target_type->field(0)->name();
124+
const std::string& bucket_name = target_type->field(3)->name();
125+
auto version_field = file_schema->GetFieldByName(version_name);
126+
auto bucket_field = file_schema->GetFieldByName(bucket_name);
127+
if (!version_field || !bucket_field || version_field->type()->id() != arrow::Type::INT32 ||
128+
bucket_field->type()->id() != arrow::Type::INT32) {
129+
return Status::OK();
130+
}
131+
arrow::FieldVector probe_fields;
132+
for (const auto& field : file_schema->fields()) {
133+
if (field->name() == version_name || field->name() == bucket_name) {
134+
probe_fields.push_back(field);
135+
}
136+
}
137+
auto probe_schema = arrow::schema(probe_fields);
138+
ArrowSchema probe_c_schema;
139+
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*probe_schema, &probe_c_schema));
140+
PAIMON_RETURN_NOT_OK(
141+
reader->SetReadSchema(&probe_c_schema, /*predicate=*/nullptr, std::nullopt));
142+
RoaringBitmap32 selected;
143+
bool row_ids_fit = true;
144+
while (true) {
145+
PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatch batch, reader->NextBatch());
146+
if (BatchReader::IsEofBatch(batch)) {
147+
break;
148+
}
149+
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
150+
std::shared_ptr<arrow::Array> array,
151+
arrow::ImportArray(batch.first.get(), batch.second.get()));
152+
if (!array || array->type_id() != arrow::Type::STRUCT) {
153+
return Status::Invalid("Manifest bucket probe must return a struct array");
154+
}
155+
auto rows = checked_pointer_cast<arrow::StructArray>(array);
156+
auto version_array = rows->GetFieldByName(version_name);
157+
auto bucket_array = rows->GetFieldByName(bucket_name);
158+
if (!version_array || !bucket_array || version_array->type_id() != arrow::Type::INT32 ||
159+
bucket_array->type_id() != arrow::Type::INT32) {
160+
return Status::Invalid("Manifest bucket probe must return int32 version and bucket");
161+
}
162+
auto versions = checked_pointer_cast<arrow::Int32Array>(version_array);
163+
auto buckets = checked_pointer_cast<arrow::Int32Array>(bucket_array);
164+
for (int64_t i = 0; i < rows->length(); ++i) {
165+
if (rows->IsNull(i) || versions->IsNull(i) || buckets->IsNull(i)) {
166+
return Status::Invalid("Manifest version and bucket must not be null");
167+
}
168+
// Validate every entry, including buckets not selected by this scan.
169+
PAIMON_RETURN_NOT_OK(ManifestEntrySerializer::ValidateVersion(versions->Value(i)));
170+
if (buckets->Value(i) == bucket) {
171+
PAIMON_ASSIGN_OR_RAISE(uint64_t file_row, reader->GetPreviousBatchFileRowId(i));
172+
if (file_row > std::numeric_limits<uint32_t>::max()) {
173+
row_ids_fit = false;
174+
} else {
175+
selected.Add(static_cast<uint32_t>(file_row));
176+
}
177+
}
178+
}
179+
}
180+
// Keep the on-disk schema; ManifestMetaReader still performs schema evolution afterwards.
181+
ArrowSchema full_c_schema;
182+
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*file_schema, &full_c_schema));
183+
return reader->SetReadSchema(
184+
&full_c_schema, /*predicate=*/nullptr,
185+
row_ids_fit ? std::optional<RoaringBitmap32>(std::move(selected)) : std::nullopt);
110186
}
111187

112188
Result<std::vector<ManifestFileMeta>> ManifestFile::Write(

‎src/paimon/core/manifest/manifest_file.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,8 @@ class ManifestFile : public ObjectsFile<ManifestEntry> {
6767
std::vector<ManifestEntry>* entries) const;
6868

6969
private:
70+
Status PrepareBucketRead(FileBatchReader* reader, int32_t bucket) const;
71+
7072
ManifestFile(const std::shared_ptr<FileSystem>& file_system,
7173
const std::shared_ptr<ReaderBuilder>& reader_builder,
7274
const std::shared_ptr<WriterBuilder>& writer_builder,

‎src/paimon/core/manifest/manifest_file_test.cpp‎

Lines changed: 124 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,15 @@
2525
#include <utility>
2626

2727
#include "arrow/api.h"
28+
#include "arrow/ipc/json_simple.h"
2829
#include "gtest/gtest.h"
2930
#include "paimon/common/data/binary_row.h"
3031
#include "paimon/common/data/data_define.h"
3132
#include "paimon/core/io/data_file_meta.h"
3233
#include "paimon/core/manifest/file_kind.h"
3334
#include "paimon/core/manifest/file_source.h"
3435
#include "paimon/core/manifest/manifest_entry.h"
36+
#include "paimon/core/manifest/manifest_entry_serializer.h"
3537
#include "paimon/core/manifest/manifest_file_meta.h"
3638
#include "paimon/core/stats/simple_stats.h"
3739
#include "paimon/core/utils/file_store_path_factory.h"
@@ -40,6 +42,8 @@
4042
#include "paimon/defs.h"
4143
#include "paimon/format/file_format.h"
4244
#include "paimon/format/file_format_factory.h"
45+
#include "paimon/format/format_writer.h"
46+
#include "paimon/format/writer_builder.h"
4347
#include "paimon/fs/local/local_file_system.h"
4448
#include "paimon/memory/memory_pool.h"
4549
#include "paimon/testing/utils/binary_row_generator.h"
@@ -100,38 +104,52 @@ class CountingFileSystem : public FileSystem {
100104

101105
class ManifestFileTest : public testing::Test {
102106
public:
103-
std::vector<ManifestEntry> ReadManifestEntry(
107+
Result<std::vector<ManifestEntry>> TryReadManifestEntry(
104108
const std::string& file_format_str, const std::string& root_path,
105109
const std::string& file_name, const std::shared_ptr<MemoryPool>& pool,
106-
const std::optional<int32_t>& bucket = std::nullopt) const {
110+
const std::optional<int32_t>& bucket = std::nullopt, bool cache_enabled = false) const {
107111
std::shared_ptr<FileSystem> file_system = std::make_shared<LocalFileSystem>();
108-
EXPECT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> file_format,
109-
FileFormatFactory::Get(file_format_str, {}));
112+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileFormat> file_format,
113+
FileFormatFactory::Get(file_format_str, {}));
110114
auto unused_schema = arrow::schema(arrow::FieldVector({arrow::field("f0", arrow::utf8())}));
111-
EXPECT_OK_AND_ASSIGN(std::shared_ptr<FileStorePathFactory> path_factory,
112-
FileStorePathFactory::Create(
113-
root_path, unused_schema, /*partition_keys=*/{},
114-
/*default_part_value=*/"", file_format->Identifier(),
115-
/*data_file_prefix=*/"data-",
116-
/*legacy_partition_name_enabled=*/true, /*external_paths=*/{},
117-
/*global_index_external_path=*/std::nullopt,
118-
/*index_file_in_data_file_dir=*/false, pool));
119-
EXPECT_OK_AND_ASSIGN(CoreOptions options,
120-
CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"}}));
121-
EXPECT_OK_AND_ASSIGN(
115+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStorePathFactory> path_factory,
116+
FileStorePathFactory::Create(
117+
root_path, unused_schema, /*partition_keys=*/{},
118+
/*default_part_value=*/"", file_format->Identifier(),
119+
/*data_file_prefix=*/"data-",
120+
/*legacy_partition_name_enabled=*/true, /*external_paths=*/{},
121+
/*global_index_external_path=*/std::nullopt,
122+
/*index_file_in_data_file_dir=*/false, pool));
123+
PAIMON_ASSIGN_OR_RAISE(CoreOptions options,
124+
CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"}}));
125+
if (cache_enabled) {
126+
options.WithCache(
127+
std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST, 64 * 1024 * 1024));
128+
}
129+
PAIMON_ASSIGN_OR_RAISE(
122130
std::unique_ptr<ManifestFile> manifest_file,
123131
ManifestFile::Create(file_system, file_format, "zstd", path_factory,
124132
/*target_file_size=*/1024, pool, options, unused_schema));
125133
std::vector<ManifestEntry> manifest_entries;
126134
if (bucket) {
127-
EXPECT_OK(
135+
PAIMON_RETURN_NOT_OK(
128136
manifest_file->ReadBucketEntries(file_name, bucket.value(), &manifest_entries));
129137
} else {
130-
EXPECT_OK(manifest_file->Read(file_name, /*filter=*/nullptr, &manifest_entries));
138+
PAIMON_RETURN_NOT_OK(
139+
manifest_file->Read(file_name, /*filter=*/nullptr, &manifest_entries));
131140
}
132141

133142
return manifest_entries;
134143
}
144+
145+
std::vector<ManifestEntry> ReadManifestEntry(
146+
const std::string& file_format_str, const std::string& root_path,
147+
const std::string& file_name, const std::shared_ptr<MemoryPool>& pool,
148+
const std::optional<int32_t>& bucket = std::nullopt) const {
149+
EXPECT_OK_AND_ASSIGN(auto manifest_entries, TryReadManifestEntry(file_format_str, root_path,
150+
file_name, pool, bucket));
151+
return manifest_entries;
152+
}
135153
};
136154

137155
TEST_F(ManifestFileTest, TestSimple) {
@@ -373,45 +391,97 @@ TEST_F(ManifestFileTest, TestReadBucketEntriesSkipsDeserializingOtherBuckets) {
373391
"manifest-3a44a0da-1008-463c-914e-28d271375e24-0", pool);
374392
ASSERT_EQ(2, source_entries.size());
375393

376-
auto test_dir = UniqueTestDirectory::Create();
377-
ASSERT_TRUE(test_dir);
378-
std::shared_ptr<FileSystem> file_system = test_dir->GetFileSystem();
379-
ASSERT_OK(file_system->Mkdirs(FileStorePathFactory::ManifestPath(test_dir->Str())));
380-
ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> file_format,
381-
FileFormatFactory::Get("avro", {}));
382-
auto unused_schema = arrow::schema(arrow::FieldVector({arrow::field("f0", arrow::utf8())}));
383-
ASSERT_OK_AND_ASSIGN(
384-
std::shared_ptr<FileStorePathFactory> path_factory,
385-
FileStorePathFactory::Create(test_dir->Str(), unused_schema, /*partition_keys=*/{},
386-
/*default_part_value=*/"", file_format->Identifier(),
387-
/*data_file_prefix=*/"data-",
388-
/*legacy_partition_name_enabled=*/true, /*external_paths=*/{},
389-
/*global_index_external_path=*/std::nullopt,
390-
/*index_file_in_data_file_dir=*/false, pool));
391-
ASSERT_OK_AND_ASSIGN(CoreOptions options,
392-
CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"}}));
393-
ASSERT_OK_AND_ASSIGN(
394-
std::unique_ptr<ManifestFile> manifest_file,
395-
ManifestFile::Create(file_system, file_format, "zstd", path_factory,
396-
/*target_file_size=*/1024, pool, options, unused_schema));
397-
398-
ManifestEntry invalid_other_bucket(FileKind(static_cast<int8_t>(2)),
399-
source_entries[0].Partition(), /*bucket=*/1,
400-
/*total_buckets=*/2, source_entries[0].File());
401-
ManifestEntry valid_target_bucket(FileKind::Add(), source_entries[1].Partition(), /*bucket=*/0,
402-
/*total_buckets=*/2, source_entries[1].File());
403-
using WrittenFile = std::pair<std::string, int64_t>;
404-
ASSERT_OK_AND_ASSIGN(
405-
WrittenFile written_file,
406-
manifest_file->WriteWithoutRolling({invalid_other_bucket, valid_target_bucket}));
394+
for (bool cache_enabled : {false, true}) {
395+
SCOPED_TRACE(cache_enabled);
396+
const std::string format = "avro";
397+
auto test_dir = UniqueTestDirectory::Create();
398+
ASSERT_TRUE(test_dir);
399+
std::shared_ptr<FileSystem> file_system = test_dir->GetFileSystem();
400+
ASSERT_OK(file_system->Mkdirs(FileStorePathFactory::ManifestPath(test_dir->Str())));
401+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> file_format,
402+
FileFormatFactory::Get(format, {}));
403+
auto unused_schema = arrow::schema(arrow::FieldVector({arrow::field("f0", arrow::utf8())}));
404+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileStorePathFactory> path_factory,
405+
FileStorePathFactory::Create(
406+
test_dir->Str(), unused_schema, /*partition_keys=*/{},
407+
/*default_part_value=*/"", file_format->Identifier(),
408+
/*data_file_prefix=*/"data-",
409+
/*legacy_partition_name_enabled=*/true, /*external_paths=*/{},
410+
/*global_index_external_path=*/std::nullopt,
411+
/*index_file_in_data_file_dir=*/false, pool));
412+
ASSERT_OK_AND_ASSIGN(CoreOptions options,
413+
CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"}}));
414+
if (cache_enabled) {
415+
options.WithCache(
416+
std::make_shared<CountingRoutingCache>(CacheKind::MANIFEST, 64 * 1024 * 1024));
417+
}
418+
ASSERT_OK_AND_ASSIGN(
419+
std::unique_ptr<ManifestFile> manifest_file,
420+
ManifestFile::Create(file_system, file_format, "null", path_factory,
421+
/*target_file_size=*/1024, pool, options, unused_schema));
407422

408-
std::vector<ManifestEntry> all_entries;
409-
ASSERT_NOK_WITH_MSG(manifest_file->Read(written_file.first, /*filter=*/nullptr, &all_entries),
410-
"Unsupported byte value 2 for file kind.");
423+
ManifestEntry invalid_other_bucket(FileKind(static_cast<int8_t>(2)),
424+
source_entries[0].Partition(), /*bucket=*/1,
425+
/*total_buckets=*/2, source_entries[0].File());
426+
ManifestEntry valid_target_bucket(FileKind::Add(), source_entries[1].Partition(),
427+
/*bucket=*/0,
428+
/*total_buckets=*/2, source_entries[1].File());
429+
using WrittenFile = std::pair<std::string, int64_t>;
430+
ASSERT_OK_AND_ASSIGN(
431+
WrittenFile written_file,
432+
manifest_file->WriteWithoutRolling({invalid_other_bucket, valid_target_bucket}));
433+
434+
std::vector<ManifestEntry> all_entries;
435+
ASSERT_NOK_WITH_MSG(
436+
manifest_file->Read(written_file.first, /*filter=*/nullptr, &all_entries),
437+
"Unsupported byte value 2 for file kind.");
438+
439+
std::vector<ManifestEntry> bucket_entries;
440+
ASSERT_OK(
441+
manifest_file->ReadBucketEntries(written_file.first, /*bucket=*/0, &bucket_entries));
442+
ASSERT_EQ(std::vector<ManifestEntry>({valid_target_bucket}), bucket_entries);
443+
}
444+
}
411445

412-
std::vector<ManifestEntry> bucket_entries;
413-
ASSERT_OK(manifest_file->ReadBucketEntries(written_file.first, /*bucket=*/0, &bucket_entries));
414-
ASSERT_EQ(std::vector<ManifestEntry>({valid_target_bucket}), bucket_entries);
446+
TEST_F(ManifestFileTest, TestBucketProbeValidatesVersionsOutsideSelectedBucket) {
447+
auto pool = GetDefaultPool();
448+
ManifestEntrySerializer serializer(pool);
449+
auto type =
450+
arrow::struct_({arrow::field(serializer.GetDataType()->field(0)->name(), arrow::int32()),
451+
arrow::field(serializer.GetDataType()->field(3)->name(), arrow::int32())});
452+
const std::vector<std::pair<std::string, std::string>> cases = {
453+
{R"([[999,1],[2,0]])", "Unsupported version: 999"},
454+
{R"([[1,1],[2,0]])", "not compatible"},
455+
{R"([[null,1],[2,0]])", "must not be null"},
456+
{R"([[2,null],[2,0]])", "must not be null"}};
457+
for (const auto& item : cases) {
458+
SCOPED_TRACE(item.first);
459+
auto dir = UniqueTestDirectory::Create();
460+
ASSERT_TRUE(dir);
461+
auto fs = dir->GetFileSystem();
462+
const std::string manifest_dir = FileStorePathFactory::ManifestPath(dir->Str());
463+
ASSERT_OK(fs->Mkdirs(manifest_dir));
464+
const std::string file_name = "manifest-invalid-version";
465+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> format,
466+
FileFormatFactory::Get("avro", {}));
467+
ArrowSchema c_schema;
468+
ASSERT_TRUE(arrow::ExportSchema(*arrow::schema(type->fields()), &c_schema).ok());
469+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<WriterBuilder> builder,
470+
format->CreateWriterBuilder(&c_schema, 2));
471+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> output,
472+
fs->Create(manifest_dir + "/" + file_name, false));
473+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<FormatWriter> writer, builder->Build(output, "null"));
474+
auto array = arrow::ipc::internal::json::ArrayFromJSON(type, item.first).ValueOrDie();
475+
ArrowArray c_array;
476+
ASSERT_TRUE(arrow::ExportArray(*array, &c_array).ok());
477+
ASSERT_OK(writer->AddBatch(&c_array));
478+
ASSERT_OK(writer->Flush());
479+
ASSERT_OK(writer->Finish());
480+
ASSERT_OK(output->Close());
481+
ASSERT_NOK_WITH_MSG(TryReadManifestEntry("avro", dir->Str(), file_name, pool, /*bucket=*/0,
482+
/*cache_enabled=*/true),
483+
item.second);
484+
}
415485
}
416486

417487
TEST_F(ManifestFileTest, TestLegacyManifestFormatIsReadOnly) {

‎src/paimon/core/operation/file_store_scan.cpp‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -415,8 +415,15 @@ Status FileStoreScan::ReadAndMergeBucketFileEntries(
415415
for (const auto& meta : manifest_metas) {
416416
auto read_meta_task = [this, meta, bucket]() -> Result<std::vector<ManifestEntry>> {
417417
std::vector<ManifestEntry> bucket_entries;
418-
PAIMON_RETURN_NOT_OK(
419-
manifest_file_->ReadBucketEntries(meta.FileName(), bucket, &bucket_entries));
418+
if (meta.MinBucket() && meta.MaxBucket() && meta.MinBucket().value() == bucket &&
419+
meta.MaxBucket().value() == bucket) {
420+
// Every entry belongs to this bucket; a projection pass cannot prune rows.
421+
PAIMON_RETURN_NOT_OK(
422+
manifest_file_->Read(meta.FileName(), /*filter=*/nullptr, &bucket_entries));
423+
} else {
424+
PAIMON_RETURN_NOT_OK(manifest_file_->ReadBucketEntries(meta.FileName(), bucket,
425+
&bucket_entries));
426+
}
420427
return bucket_entries;
421428
};
422429
futures.push_back(Via(executor_.get(), read_meta_task));

0 commit comments

Comments
 (0)