Skip to content

Commit 2de05bd

Browse files
lucasfanglucasfang
committed
fix
1 parent 3eef591 commit 2de05bd

7 files changed

Lines changed: 23 additions & 27 deletions

‎src/paimon/core/io/append_data_file_writer_factory.cpp‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,12 +42,14 @@ AppendDataFileWriterFactory::AppendDataFileWriterFactory(
4242

4343
Result<std::unique_ptr<SingleFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>>
4444
AppendDataFileWriterFactory::CreateWriter() const {
45+
std::shared_ptr<LongCounter> seq_num_counter =
46+
options_.DataEvolutionEnabled() ? std::make_shared<LongCounter>(0) : seq_num_counter_;
4547
PAIMON_ASSIGN_OR_RAISE(WriterResources resources,
4648
CreateWriterResources(*options_.GetFileFormat(), write_schema_,
4749
/*create_stats_extractor=*/true));
4850
auto writer = std::make_unique<DataFileWriter>(
4951
options_.GetFileCompression(), std::function<Status(::ArrowArray*, ::ArrowArray*)>(),
50-
schema_id_, seq_num_counter_, file_source_, resources.stats_extractor,
52+
schema_id_, seq_num_counter, file_source_, resources.stats_extractor,
5153
path_factory_->IsExternalPath(), write_cols_, pool_);
5254
PAIMON_RETURN_NOT_OK(
5355
writer->Init(options_.GetFileSystem(), path_factory_->NewPath(), resources.writer_builder));

‎src/paimon/core/io/blob_data_file_writer_factory.cpp‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,14 +43,16 @@ BlobDataFileWriterFactory::BlobDataFileWriterFactory(
4343

4444
Result<std::unique_ptr<SingleFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>>>>
4545
BlobDataFileWriterFactory::CreateWriter() const {
46+
std::shared_ptr<LongCounter> seq_num_counter =
47+
options_.DataEvolutionEnabled() ? std::make_shared<LongCounter>(0) : seq_num_counter_;
4648
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileFormat> format,
4749
FileFormatFactory::Get("blob", options_.ToMap()));
4850
PAIMON_ASSIGN_OR_RAISE(WriterResources resources,
4951
CreateWriterResources(*format, file_schema_,
5052
/*create_stats_extractor=*/true));
5153
auto writer = std::make_unique<DataFileWriter>(
5254
/*compression=*/"none", std::function<Status(::ArrowArray*, ::ArrowArray*)>(), schema_id_,
53-
seq_num_counter_, FileSource::Append(), resources.stats_extractor,
55+
seq_num_counter, FileSource::Append(), resources.stats_extractor,
5456
path_factory_->IsExternalPath(), write_cols_, pool_);
5557
PAIMON_RETURN_NOT_OK(writer->Init(options_.GetFileSystem(), path_factory_->NewBlobPath(),
5658
resources.writer_builder));

‎src/paimon/core/io/shredding_append_data_file_writer_factory.cpp‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,8 @@ ShreddingAppendDataFileWriterFactory::CreateWriter() const {
4747
if (!shredding_context_) {
4848
return Status::Invalid("Shared-shredding append writer requires a shredding context.");
4949
}
50+
std::shared_ptr<LongCounter> seq_num_counter =
51+
options_.DataEvolutionEnabled() ? std::make_shared<LongCounter>(0) : seq_num_counter_;
5052
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<MapSharedShreddingBatchConverter> converter,
5153
MapSharedShreddingBatchConverter::Create(
5254
write_schema_, shredding_context_, options_, pool_));
@@ -61,7 +63,7 @@ ShreddingAppendDataFileWriterFactory::CreateWriter() const {
6163
CreateWriterResources(*options_.GetFileFormat(), file_schema,
6264
/*create_stats_extractor=*/true));
6365
auto writer = std::make_unique<DataFileWriter>(
64-
options_.GetFileCompression(), std::move(batch_converter), schema_id_, seq_num_counter_,
66+
options_.GetFileCompression(), std::move(batch_converter), schema_id_, seq_num_counter,
6567
file_source_, resources.stats_extractor, path_factory_->IsExternalPath(), write_cols_,
6668
pool_);
6769
PAIMON_RETURN_NOT_OK(

‎src/paimon/core/operation/commit/row_tracking_commit_utils.cpp‎

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -62,19 +62,13 @@ void RowTrackingCommitUtils::AssignSnapshotId(int64_t snapshot_id,
6262
for (const auto& entry : delta_files) {
6363
ManifestEntry assigned_entry = CloneEntryWithClonedFileMeta(entry);
6464
int64_t min_seq_number = assigned_entry.File()->min_sequence_number;
65-
int64_t max_seq_number = assigned_entry.File()->max_sequence_number;
6665
if (min_seq_number == 0L) {
6766
// Case 1: New file (e.g., from INSERT)
6867
// All records in this file get the current snapshot ID as sequence number
6968
assigned_entry.AssignSequenceNumber(snapshot_id, snapshot_id);
70-
} else if (max_seq_number == 0L) {
71-
// Case 2: File with some modified records
72-
// - min: Preserve original sequence number (from unmodified records)
73-
// - max: Assign current snapshot ID
74-
assigned_entry.AssignSequenceNumber(min_seq_number, snapshot_id);
7569
} else {
76-
// Case 3: Pure compact file (no modified records)
77-
// Preserve original min/max sequence numbers from source files
70+
// Preserve original min/max sequence numbers from source files.
71+
// This follows Java release-1.4 row-tracking commit semantics.
7872
}
7973
snapshot_assigned->emplace_back(std::move(assigned_entry));
8074
}

‎src/paimon/core/operation/commit/row_tracking_commit_utils_test.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@ TEST_F(RowTrackingCommitUtilsTest, TestAssignRowTrackingStampsSequence) {
8181
EXPECT_EQ(100, assigned.assigned_entries[0].File()->min_sequence_number);
8282
EXPECT_EQ(100, assigned.assigned_entries[0].File()->max_sequence_number);
8383
EXPECT_EQ(7, assigned.assigned_entries[1].File()->min_sequence_number);
84-
EXPECT_EQ(100, assigned.assigned_entries[1].File()->max_sequence_number);
84+
EXPECT_EQ(0, assigned.assigned_entries[1].File()->max_sequence_number);
8585
EXPECT_EQ(3, assigned.assigned_entries[2].File()->min_sequence_number);
8686
EXPECT_EQ(5, assigned.assigned_entries[2].File()->max_sequence_number);
8787
}

‎test/inte/clean_inte_test.cpp‎

Lines changed: 5 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -500,11 +500,9 @@ TEST_F(CleanInteTest, TestDropPartitionAndExpireSnapshotWithIOException) {
500500
ASSERT_TRUE(commit_impl);
501501
ASSERT_OK_AND_ASSIGN(bool snapshot_exist,
502502
commit_impl->snapshot_manager_->SnapshotExists(1));
503-
if (snapshot_exist) {
504-
ASSERT_OK_AND_ASSIGN(bool snapshot2_exist,
505-
commit_impl->snapshot_manager_->SnapshotExists(2));
506-
ASSERT_TRUE(snapshot2_exist);
507-
}
503+
ASSERT_TRUE(snapshot_exist);
504+
ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(2));
505+
ASSERT_TRUE(snapshot_exist);
508506
ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(3));
509507
ASSERT_TRUE(snapshot_exist);
510508
ASSERT_OK_AND_ASSIGN(Snapshot snapshot_3, commit_impl->snapshot_manager_->LoadSnapshot(3));
@@ -528,10 +526,8 @@ TEST_F(CleanInteTest, TestDropPartitionAndExpireSnapshotWithIOException) {
528526
ASSERT_FALSE(snapshot_exist);
529527
ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(2));
530528
ASSERT_FALSE(snapshot_exist);
531-
ASSERT_OK_AND_ASSIGN(std::optional<Snapshot> latest_snapshot,
532-
commit_impl->snapshot_manager_->LatestSnapshot());
533-
ASSERT_TRUE(latest_snapshot.has_value());
534-
ASSERT_GE(latest_snapshot->Id(), 3);
529+
ASSERT_OK_AND_ASSIGN(snapshot_exist, commit_impl->snapshot_manager_->SnapshotExists(3));
530+
ASSERT_TRUE(snapshot_exist);
535531
ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch_5,
536532
MakeRecordBatch({{"Lucas", 20, 2, 22.2}}, {{"f1", "20"}}, 0));
537533
ASSERT_OK(file_store_write->Write(std::move(batch_5)));

‎test/inte/write_inte_test.cpp‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3854,7 +3854,7 @@ TEST_P(WriteInteTest, TestAppendTableWriteWithBlobType) {
38543854
/*key_stats=*/SimpleStats::EmptyStats(),
38553855
BinaryRowGenerator::GenerateStats({std::string("str_0"), 1}, {std::string("str_3"), 2},
38563856
std::vector<int64_t>({0, 2}), pool_.get()),
3857-
/*min_sequence_number=*/4, /*max_sequence_number=*/7, /*schema_id=*/0,
3857+
/*min_sequence_number=*/1, /*max_sequence_number=*/1, /*schema_id=*/0,
38583858
/*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
38593859
/*creation_time=*/Timestamp(1724090888706ll, 0),
38603860
/*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(),
@@ -3865,7 +3865,7 @@ TEST_P(WriteInteTest, TestAppendTableWriteWithBlobType) {
38653865
"data-xxx.blob", /*file_size=*/764, /*row_count=*/3,
38663866
/*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/BinaryRow::EmptyRow(),
38673867
/*key_stats=*/SimpleStats::EmptyStats(), GenerateBlobValueStats(),
3868-
/*min_sequence_number=*/4, /*max_sequence_number=*/6, /*schema_id=*/0,
3868+
/*min_sequence_number=*/1, /*max_sequence_number=*/1, /*schema_id=*/0,
38693869
/*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
38703870
/*creation_time=*/Timestamp(1724090888706ll, 0),
38713871
/*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(),
@@ -3875,7 +3875,7 @@ TEST_P(WriteInteTest, TestAppendTableWriteWithBlobType) {
38753875
"data-xxx.blob", /*file_size=*/3023, /*row_count=*/1,
38763876
/*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/BinaryRow::EmptyRow(),
38773877
/*key_stats=*/SimpleStats::EmptyStats(), GenerateBlobValueStats(),
3878-
/*min_sequence_number=*/7, /*max_sequence_number=*/7, /*schema_id=*/0,
3878+
/*min_sequence_number=*/1, /*max_sequence_number=*/1, /*schema_id=*/0,
38793879
/*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
38803880
/*creation_time=*/Timestamp(1724090888706ll, 0),
38813881
/*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(),
@@ -4687,7 +4687,7 @@ TEST_P(WriteInteTest, TestAppendTableWriteWithMultipleBlobFields) {
46874687
/*key_stats=*/SimpleStats::EmptyStats(),
46884688
BinaryRowGenerator::GenerateStats({std::string("str_0"), 1}, {std::string("str_2"), 2},
46894689
std::vector<int64_t>({0, 1}), pool_.get()),
4690-
/*min_sequence_number=*/6, /*max_sequence_number=*/8, /*schema_id=*/0,
4690+
/*min_sequence_number=*/1, /*max_sequence_number=*/1, /*schema_id=*/0,
46914691
/*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
46924692
/*creation_time=*/Timestamp(0, 0),
46934693
/*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(),
@@ -4700,7 +4700,7 @@ TEST_P(WriteInteTest, TestAppendTableWriteWithMultipleBlobFields) {
47004700
"data-xxx.blob", /*file_size=*/0, /*row_count=*/3,
47014701
/*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/BinaryRow::EmptyRow(),
47024702
/*key_stats=*/SimpleStats::EmptyStats(), GenerateBlobValueStats(),
4703-
/*min_sequence_number=*/6, /*max_sequence_number=*/8, /*schema_id=*/0,
4703+
/*min_sequence_number=*/1, /*max_sequence_number=*/1, /*schema_id=*/0,
47044704
/*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
47054705
/*creation_time=*/Timestamp(0, 0),
47064706
/*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(),
@@ -4712,7 +4712,7 @@ TEST_P(WriteInteTest, TestAppendTableWriteWithMultipleBlobFields) {
47124712
"data-xxx.blob", /*file_size=*/0, /*row_count=*/3,
47134713
/*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/BinaryRow::EmptyRow(),
47144714
/*key_stats=*/SimpleStats::EmptyStats(), GenerateBlobValueStats(),
4715-
/*min_sequence_number=*/6, /*max_sequence_number=*/8, /*schema_id=*/0,
4715+
/*min_sequence_number=*/1, /*max_sequence_number=*/1, /*schema_id=*/0,
47164716
/*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
47174717
/*creation_time=*/Timestamp(0, 0),
47184718
/*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(),

0 commit comments

Comments
 (0)