Skip to content

Commit e15dd5a

Browse files
gripleafgripleaf
committed
fix(manifest): align bucket pruning and address reader review
1 parent fa27538 commit e15dd5a

9 files changed

Lines changed: 237 additions & 249 deletions

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

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -113,7 +113,7 @@ Status ManifestFile::ReadBucketEntries(const std::string& file_name, int32_t buc
113113
row.SetRowId(i);
114114
PAIMON_RETURN_NOT_OK(
115115
ManifestEntrySerializer::ValidateVersion(row.GetInt(kVersionFieldIndex)));
116-
// Different or unknown bucket counts must reach the compatibility checks.
116+
// Different or unknown bucket counts must reach per-entry bucket filtering.
117117
const bool historical_layout =
118118
expected_total_buckets &&
119119
(row.IsNullAt(kBucketFieldIndex) || row.IsNullAt(kTotalBucketsFieldIndex) ||
@@ -142,28 +142,27 @@ Status ManifestFile::PrepareBucketRead(int32_t bucket,
142142
arrow::ImportSchema(c_schema.get()));
143143
const auto& target_type = serializer_->GetDataType();
144144
const std::string& bucket_name = target_type->field(kBucketFieldIndex)->name();
145-
std::shared_ptr<Predicate> selector = PredicateBuilder::Equal(
146-
file_schema->GetFieldIndex(bucket_name), bucket_name, FieldType::INT, Literal(bucket));
145+
std::shared_ptr<Predicate> selector =
146+
PredicateBuilder::Equal(kBucketFieldIndex, bucket_name, FieldType::INT, Literal(bucket));
147147
if (expected_total_buckets) {
148148
const std::string& total_name = target_type->field(kTotalBucketsFieldIndex)->name();
149-
const int32_t total_index = file_schema->GetFieldIndex(total_name);
150149
PAIMON_ASSIGN_OR_RAISE(
151150
selector,
152151
PredicateBuilder::Or(
153-
{selector, PredicateBuilder::IsNull(total_index, total_name, FieldType::INT),
154-
PredicateBuilder::NotEqual(total_index, total_name, FieldType::INT,
152+
{selector,
153+
PredicateBuilder::IsNull(kTotalBucketsFieldIndex, total_name, FieldType::INT),
154+
PredicateBuilder::NotEqual(kTotalBucketsFieldIndex, total_name, FieldType::INT,
155155
Literal(expected_total_buckets.value()))}));
156156
}
157157
// Retain unsupported versions regardless of bucket so the consumer validates every version
158158
// before bucket filtering, including when the probe would otherwise select no entries.
159159
const std::string& version_name = target_type->field(kVersionFieldIndex)->name();
160-
const int32_t version_index = file_schema->GetFieldIndex(version_name);
161160
PAIMON_ASSIGN_OR_RAISE(
162161
selector,
163162
PredicateBuilder::Or(
164-
{selector, PredicateBuilder::IsNull(version_index, version_name, FieldType::INT),
163+
{selector, PredicateBuilder::IsNull(kVersionFieldIndex, version_name, FieldType::INT),
165164
PredicateBuilder::NotEqual(
166-
version_index, version_name, FieldType::INT,
165+
kVersionFieldIndex, version_name, FieldType::INT,
167166
Literal(
168167
checked_cast<ManifestEntrySerializer*>(serializer_.get())->GetVersion()))}));
169168
if (!PredicateValidator::ValidatePredicateWithSchema(*file_schema, selector,

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

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -64,9 +64,8 @@ class ManifestFile : public ObjectsFile<ManifestEntry> {
6464
Result<std::vector<ManifestFileMeta>> Write(const std::vector<ManifestEntry>& entries);
6565

6666
/// Read entries for a bucket. An inferred total bucket count also retains entries with
67-
/// different or unknown bucket counts for the existing compatibility checks.
68-
/// For inferred buckets, the caller must first verify that all historical bucket schemas
69-
/// covered by the manifest are compatible with the scan's bucket selector.
67+
/// different or unknown bucket counts for per-entry bucket filtering.
68+
/// Bucket pruning assumes stable bucket-key hashing across schema versions.
7069
/// Serialization versions are validated for all entries before bucket filtering.
7170
Status ReadBucketEntries(const std::string& file_name, int32_t bucket,
7271
const std::optional<int32_t>& expected_total_buckets,

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

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,9 @@
3030
#include "gtest/gtest.h"
3131
#include "paimon/common/data/binary_row.h"
3232
#include "paimon/common/data/data_define.h"
33+
#include "paimon/common/utils/checked_cast.h"
3334
#include "paimon/core/io/data_file_meta.h"
35+
#include "paimon/core/io/meta_to_arrow_array_converter.h"
3436
#include "paimon/core/manifest/file_kind.h"
3537
#include "paimon/core/manifest/file_source.h"
3638
#include "paimon/core/manifest/manifest_entry.h"
@@ -530,6 +532,73 @@ TEST_F(ManifestFileTest, TestReadBucketEntriesSkipsDeserializingOtherBuckets) {
530532
}
531533
}
532534

535+
TEST_F(ManifestFileTest, TestReadBucketEntriesWithReorderedFields) {
536+
auto pool = GetDefaultPool();
537+
auto source =
538+
ReadManifestEntry("orc", paimon::test::GetDataDir() + "/orc/append_09.db/append_09",
539+
"manifest-3a44a0da-1008-463c-914e-28d271375e24-0", pool);
540+
ASSERT_EQ(source.size(), 2);
541+
std::vector<ManifestEntry> entries;
542+
std::vector<BinaryRow> rows;
543+
ManifestEntrySerializer serializer(pool);
544+
for (int32_t bucket = 0; bucket < 2; ++bucket) {
545+
entries.emplace_back(FileKind::Add(), source[bucket].Partition(), bucket, 2,
546+
source[bucket].File());
547+
ASSERT_OK_AND_ASSIGN(BinaryRow row, serializer.ToRow(entries.back()));
548+
rows.push_back(std::move(row));
549+
}
550+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<MetaToArrowArrayConverter> converter,
551+
MetaToArrowArrayConverter::Create(serializer.GetDataType(), pool));
552+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> array, converter->NextBatch(rows));
553+
auto struct_array = checked_pointer_cast<arrow::StructArray>(array);
554+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> format, FileFormatFactory::Get("avro", {}));
555+
// Move each selector field away from its standard position. Ordinary schema alignment
556+
// must still restore the fields before version validation and bucket filtering.
557+
for (int32_t field_index : {0, 3, 4}) {
558+
SCOPED_TRACE(field_index);
559+
auto fields = serializer.GetDataType()->fields();
560+
auto columns = struct_array->fields();
561+
std::swap(fields[1], fields[field_index]);
562+
std::swap(columns[1], columns[field_index]);
563+
auto reordered_result = arrow::StructArray::Make(columns, fields);
564+
ASSERT_TRUE(reordered_result.ok()) << reordered_result.status().ToString();
565+
std::shared_ptr<arrow::StructArray> reordered = reordered_result.ValueOrDie();
566+
auto dir = UniqueTestDirectory::Create();
567+
ASSERT_TRUE(dir);
568+
auto fs = dir->GetFileSystem();
569+
const std::string manifest_dir = FileStorePathFactory::ManifestPath(dir->Str());
570+
ASSERT_OK(fs->Mkdirs(manifest_dir));
571+
const std::string file_name = "manifest-reordered";
572+
ArrowSchema c_schema;
573+
ASSERT_TRUE(arrow::ExportSchema(*arrow::schema(fields), &c_schema).ok());
574+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<WriterBuilder> builder,
575+
format->CreateWriterBuilder(&c_schema, 2));
576+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> output,
577+
fs->Create(manifest_dir + "/" + file_name, false));
578+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<FormatWriter> writer, builder->Build(output, "null"));
579+
ArrowArray c_array;
580+
ASSERT_TRUE(arrow::ExportArray(*reordered, &c_array).ok());
581+
ASSERT_OK(writer->AddBatch(&c_array));
582+
ASSERT_OK(writer->Flush());
583+
ASSERT_OK(writer->Finish());
584+
ASSERT_OK(output->Close());
585+
for (bool cache_enabled : {false, true}) {
586+
for (bool inferred_bucket : {false, true}) {
587+
for (int32_t bucket : {0, 1, 2}) {
588+
ASSERT_OK_AND_ASSIGN(
589+
std::vector<ManifestEntry> actual,
590+
TryReadManifestEntry("avro", dir->Str(), file_name, pool, bucket,
591+
cache_enabled, inferred_bucket));
592+
const std::vector<ManifestEntry> expected =
593+
bucket < 2 ? std::vector<ManifestEntry>{entries[bucket]}
594+
: std::vector<ManifestEntry>{};
595+
ASSERT_EQ(actual, expected);
596+
}
597+
}
598+
}
599+
}
600+
}
601+
533602
TEST_F(ManifestFileTest, TestReadBucketEntriesValidatesAllVersions) {
534603
auto pool = GetDefaultPool();
535604
ManifestEntrySerializer serializer(pool);

0 commit comments

Comments
 (0)