This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 7351dad0 feat(core): complete file index write lifecycle (#279)
7351dad0 is described below
commit 7351dad0aa7b479e462fb4309de9cbfc8a378d40
Author: Zhang Jiawei <[email protected]>
AuthorDate: Fri Sep 4 14:30:42 2026 +0800
feat(core): complete file index write lifecycle (#279)
---
src/paimon/CMakeLists.txt | 1 +
src/paimon/core/append/append_only_writer.cpp | 4 +-
src/paimon/core/append/append_only_writer_test.cpp | 15 +++-
src/paimon/core/io/data_file_index_writer_test.cpp | 17 ++--
src/paimon/core/io/data_file_writer_base.h | 1 +
src/paimon/core/io/data_file_writer_base_test.cpp | 97 ++++++++++++++++++++++
.../core/io/key_value_data_file_writer_factory.cpp | 1 +
...hredding_key_value_data_file_writer_factory.cpp | 1 +
src/paimon/core/io/single_file_writer.h | 12 ++-
src/paimon/core/mergetree/lookup_levels.h | 4 +-
src/paimon/core/mergetree/merge_tree_writer.cpp | 14 +++-
src/paimon/core/mergetree/merge_tree_writer.h | 1 +
.../core/mergetree/merge_tree_writer_test.cpp | 40 +++++++--
src/paimon/core/operation/expire_snapshots.cpp | 38 ++++++---
src/paimon/core/operation/expire_snapshots.h | 8 +-
.../core/operation/expire_snapshots_test.cpp | 47 ++++++++++-
.../core/operation/file_store_commit_impl.cpp | 5 +-
.../core/operation/file_store_commit_impl_test.cpp | 11 ++-
.../core/operation/orphan_files_cleaner_impl.cpp | 16 ++++
.../core/operation/orphan_files_cleaner_test.cpp | 33 +++++++-
test/inte/clean_inte_test.cpp | 10 ++-
21 files changed, 328 insertions(+), 48 deletions(-)
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index e2746649..4c9b9b34 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -779,6 +779,7 @@ if(PAIMON_BUILD_TESTS)
core/io/complete_row_tracking_fields_reader_test.cpp
core/io/vector_file_batch_reader_test.cpp
core/io/data_file_meta_test.cpp
+ core/io/data_file_writer_base_test.cpp
core/io/data_file_index_writer_test.cpp
core/io/file_index_options_test.cpp
core/io/file_index_evaluator_test.cpp
diff --git a/src/paimon/core/append/append_only_writer.cpp
b/src/paimon/core/append/append_only_writer.cpp
index d70611d5..778e898f 100644
--- a/src/paimon/core/append/append_only_writer.cpp
+++ b/src/paimon/core/append/append_only_writer.cpp
@@ -284,7 +284,9 @@ Status AppendOnlyWriter::Close() {
for (const auto& file : compact_after_) {
// AppendOnlyCompactManager will rewrite the file and no file upgrade
will occur, so we
// can directly delete the file in compact_after_.
- [[maybe_unused]] auto s = fs->Delete(path_factory_->ToPath(file));
+ for (const std::string& path : path_factory_->CollectFiles(file)) {
+ [[maybe_unused]] Status s = fs->Delete(path);
+ }
}
if (writer_) {
diff --git a/src/paimon/core/append/append_only_writer_test.cpp
b/src/paimon/core/append/append_only_writer_test.cpp
index 4a4d5dc6..835717d4 100644
--- a/src/paimon/core/append/append_only_writer_test.cpp
+++ b/src/paimon/core/append/append_only_writer_test.cpp
@@ -443,6 +443,9 @@ TEST_F(AppendOnlyWriterTest, TestWriteAndClose) {
std::map<std::string, std::string> raw_options;
raw_options[Options::FILE_FORMAT] = "orc";
raw_options[Options::FILE_SYSTEM] = "local";
+ raw_options[Options::TARGET_FILE_ROW_NUM] = "1";
+ raw_options["file-index.bitmap.columns"] = "f0";
+ raw_options[Options::FILE_INDEX_IN_MANIFEST_THRESHOLD] = "1B";
ASSERT_OK_AND_ASSIGN(CoreOptions options,
CoreOptions::FromMap(raw_options));
arrow::FieldVector fields = {arrow::field("f0", arrow::utf8())};
@@ -475,11 +478,15 @@ TEST_F(AppendOnlyWriterTest, TestWriteAndClose) {
ASSERT_OK_AND_ASSIGN(auto record_batch, batch_builder.Finish());
ASSERT_OK(writer->Write(std::move(record_batch)));
ASSERT_TRUE(ArrowArrayIsReleased(&arrow_array));
- ASSERT_OK(writer->Close());
auto file_system = std::make_shared<LocalFileSystem>();
std::vector<BasicFileStatus> file_status_list;
ASSERT_OK(file_system->ListDir(dir->Str(), &file_status_list));
+ ASSERT_EQ(2, file_status_list.size());
+
+ ASSERT_OK(writer->Close());
+ file_status_list.clear();
+ ASSERT_OK(file_system->ListDir(dir->Str(), &file_status_list));
ASSERT_TRUE(file_status_list.empty());
}
@@ -634,9 +641,13 @@ TEST_F(AppendOnlyWriterTest,
TestCloseDeletesCompactAfterFiles) {
auto compact_manager = std::make_shared<FakeCompactManager>();
auto compact_after = NewAppendFile("compact-after.orc", 1, 0, 0);
+ compact_after->extra_files = {"compact-after.orc.index"};
auto compact_after_path = path_factory->ToPath(compact_after->file_name);
+ auto compact_after_index_path =
path_factory->ToPath(compact_after->extra_files[0].value());
ASSERT_OK_AND_ASSIGN(auto output,
options.GetFileSystem()->Create(compact_after_path, true));
ASSERT_OK(output->Close());
+ ASSERT_OK_AND_ASSIGN(output,
options.GetFileSystem()->Create(compact_after_index_path, true));
+ ASSERT_OK(output->Close());
auto result =
std::make_shared<CompactResult>(std::vector<std::shared_ptr<DataFileMeta>>{},
@@ -653,8 +664,10 @@ TEST_F(AppendOnlyWriterTest,
TestCloseDeletesCompactAfterFiles) {
ASSERT_OK(writer->Sync());
ASSERT_TRUE(options.GetFileSystem()->Exists(compact_after_path).value());
+
ASSERT_TRUE(options.GetFileSystem()->Exists(compact_after_index_path).value());
ASSERT_OK(writer->Close());
ASSERT_FALSE(options.GetFileSystem()->Exists(compact_after_path).value());
+
ASSERT_FALSE(options.GetFileSystem()->Exists(compact_after_index_path).value());
ASSERT_TRUE(compact_manager->request_cancel_called);
ASSERT_TRUE(compact_manager->wait_called);
ASSERT_TRUE(compact_manager->close_called);
diff --git a/src/paimon/core/io/data_file_index_writer_test.cpp
b/src/paimon/core/io/data_file_index_writer_test.cpp
index 9837b556..524fc139 100644
--- a/src/paimon/core/io/data_file_index_writer_test.cpp
+++ b/src/paimon/core/io/data_file_index_writer_test.cpp
@@ -143,9 +143,10 @@ class DataFileIndexWriterTest : public ::testing::Test {
std::shared_ptr<arrow::Schema> schema_;
};
-TEST_F(DataFileIndexWriterTest, TestBitmapAndRangeBitmapEmbeddedRoundTrip) {
+TEST_F(DataFileIndexWriterTest,
TestMultipleIndexesOnSameColumnEmbeddedRoundTrip) {
ASSERT_OK_AND_ASSIGN(auto writer,
CreateWriter({{"file-index.bitmap.columns", "f0"},
+ {"file-index.bsi.columns", "f0"},
{"file-index.range-bitmap.columns",
"f1"},
{"file-index.range-bitmap.f1.chunk-size", "1KB"},
{Options::FILE_INDEX_IN_MANIFEST_THRESHOLD, "1MB"}}));
@@ -159,12 +160,14 @@ TEST_F(DataFileIndexWriterTest,
TestBitmapAndRangeBitmapEmbeddedRoundTrip) {
ASSERT_TRUE(result.extra_files.empty());
ASSERT_OK_AND_ASSIGN(auto reader, CreateReader(result.embedded_index));
- ASSERT_OK_AND_ASSIGN(auto bitmap_readers, ReadColumn(reader.get(), "f0"));
- ASSERT_EQ(1, bitmap_readers.size());
- ASSERT_OK_AND_ASSIGN(auto equal_result,
bitmap_readers[0]->VisitEqual(Literal(1)));
- ASSERT_EQ("{0,2}", equal_result->ToString());
- ASSERT_OK_AND_ASSIGN(auto null_result, bitmap_readers[0]->VisitIsNull());
- ASSERT_EQ("{3}", null_result->ToString());
+ ASSERT_OK_AND_ASSIGN(auto f0_readers, ReadColumn(reader.get(), "f0"));
+ ASSERT_EQ(2, f0_readers.size());
+ for (const std::shared_ptr<FileIndexReader>& f0_reader : f0_readers) {
+ ASSERT_OK_AND_ASSIGN(auto equal_result,
f0_reader->VisitEqual(Literal(1)));
+ ASSERT_EQ("{0,2}", equal_result->ToString());
+ ASSERT_OK_AND_ASSIGN(auto null_result, f0_reader->VisitIsNull());
+ ASSERT_EQ("{3}", null_result->ToString());
+ }
ASSERT_OK_AND_ASSIGN(auto range_readers, ReadColumn(reader.get(), "f1"));
ASSERT_EQ(1, range_readers.size());
diff --git a/src/paimon/core/io/data_file_writer_base.h
b/src/paimon/core/io/data_file_writer_base.h
index 9b3d6e71..a0d01845 100644
--- a/src/paimon/core/io/data_file_writer_base.h
+++ b/src/paimon/core/io/data_file_writer_base.h
@@ -88,6 +88,7 @@ class DataFileWriterBase : public SingleFileWriter<Record,
std::shared_ptr<DataF
/// Extracts the pre-conversion Arrow batch from record for file index
construction, then
/// passes record to the underlying data file writer, which may convert it
to a physical schema.
Status WriteRecordWithFileIndex(Record record) {
+ PAIMON_RETURN_NOT_OK(this->CheckNotClosed());
PAIMON_RETURN_NOT_OK(AddFileIndexBatch(GetFileIndexBatch(record)));
return Base::Write(std::move(record));
}
diff --git a/src/paimon/core/io/data_file_writer_base_test.cpp
b/src/paimon/core/io/data_file_writer_base_test.cpp
new file mode 100644
index 00000000..9a3df741
--- /dev/null
+++ b/src/paimon/core/io/data_file_writer_base_test.cpp
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/core/io/data_file_writer_base.h"
+
+#include <memory>
+#include <utility>
+
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
+#include "arrow/ipc/json_simple.h"
+#include "gtest/gtest.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/io/data_file_index_writer.h"
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/core/io/file_index_options.h"
+#include "paimon/defs.h"
+#include "paimon/format/file_format.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+class IndexedDataFileWriter : public DataFileWriterBase<::ArrowArray*> {
+ public:
+ IndexedDataFileWriter() : DataFileWriterBase(/*compression=*/"zstd",
/*converter=*/nullptr) {}
+
+ Status Write(::ArrowArray* batch) override {
+ return WriteRecordWithFileIndex(batch);
+ }
+
+ Result<std::shared_ptr<DataFileMeta>> GetResult() override {
+ return std::shared_ptr<DataFileMeta>();
+ }
+};
+
+TEST(DataFileWriterBaseTest, WriteAfterCloseDoesNotConsumeIndexBatch) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ std::shared_ptr<MemoryPool> pool = GetDefaultPool();
+ std::shared_ptr<arrow::Schema> schema = arrow::schema({arrow::field("col",
arrow::int32())});
+ std::shared_ptr<arrow::DataType> data_type =
arrow::struct_(schema->fields());
+ ASSERT_OK_AND_ASSIGN(
+ CoreOptions options,
+ CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"},
+ {"file-index.bitmap.columns", "col"},
+ {Options::FILE_INDEX_IN_MANIFEST_THRESHOLD,
"1MB"}}));
+ std::shared_ptr<FileSystem> file_system = options.GetFileSystem();
+ auto path_factory = std::make_shared<DataFilePathFactory>();
+ ASSERT_OK(path_factory->Init(dir->Str(), "orc", "data-", nullptr));
+ ASSERT_OK_AND_ASSIGN(FileIndexOptions file_index_options,
+ FileIndexOptions::FromCoreOptions(options));
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<DataFileIndexWriter> file_index_writer,
+ DataFileIndexWriter::Create(schema, file_index_options, file_system,
path_factory, pool));
+
+ ArrowSchema c_schema;
+ ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<WriterBuilder> writer_builder,
+
options.GetFileFormat()->CreateWriterBuilder(&c_schema,
+
/*batch_size=*/100));
+ IndexedDataFileWriter writer;
+ writer.SetFileIndexWriter(std::move(file_index_writer), schema);
+ ASSERT_OK(writer.Init(file_system, path_factory->NewPath(),
writer_builder));
+
+ std::shared_ptr<arrow::Array> first_array =
+ arrow::ipc::internal::json::ArrayFromJSON(data_type,
R"([[1]])").ValueOrDie();
+ ::ArrowArray first_batch;
+ ASSERT_TRUE(arrow::ExportArray(*first_array, &first_batch).ok());
+ ASSERT_OK(writer.Write(&first_batch));
+ ASSERT_OK(writer.Close());
+
+ std::shared_ptr<arrow::Array> second_array =
+ arrow::ipc::internal::json::ArrayFromJSON(data_type,
R"([[2]])").ValueOrDie();
+ ::ArrowArray second_batch;
+ ASSERT_TRUE(arrow::ExportArray(*second_array, &second_batch).ok());
+ ASSERT_NOK_WITH_MSG(writer.Write(&second_batch), "Writer has already
closed");
+ ASSERT_NE(nullptr, second_batch.release);
+ ArrowArrayRelease(&second_batch);
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/io/key_value_data_file_writer_factory.cpp
b/src/paimon/core/io/key_value_data_file_writer_factory.cpp
index e0640ab6..ae3287ce 100644
--- a/src/paimon/core/io/key_value_data_file_writer_factory.cpp
+++ b/src/paimon/core/io/key_value_data_file_writer_factory.cpp
@@ -62,6 +62,7 @@ KeyValueDataFileWriterFactory::CreateWriter() const {
auto writer = std::make_unique<KeyValueDataFileWriter>(
GetFileCompression(), std::move(converter), schema_id_, level_,
file_source_, primary_keys_,
resources.stats_extractor, write_schema_,
path_factory_->IsExternalPath(), pool_);
+ // Changelog files are consumed sequentially and intentionally do not
produce file indexes.
if (!is_changelog_) {
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<DataFileIndexWriter>
file_index_writer,
CreateFileIndexWriter(write_schema_,
path_factory_));
diff --git
a/src/paimon/core/io/shredding_key_value_data_file_writer_factory.cpp
b/src/paimon/core/io/shredding_key_value_data_file_writer_factory.cpp
index f56ba8d1..b8822324 100644
--- a/src/paimon/core/io/shredding_key_value_data_file_writer_factory.cpp
+++ b/src/paimon/core/io/shredding_key_value_data_file_writer_factory.cpp
@@ -90,6 +90,7 @@ ShreddingKeyValueDataFileWriterFactory::CreateShreddedWriter(
auto writer = std::make_unique<KeyValueDataFileWriter>(
compression, std::move(batch_converter), schema_id_, level_,
file_source_, primary_keys_,
resources.stats_extractor, file_schema,
path_factory_->IsExternalPath(), pool_);
+ // Changelog files are consumed sequentially and intentionally do not
produce file indexes.
if (!is_changelog_) {
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<DataFileIndexWriter>
file_index_writer,
CreateFileIndexWriter(write_schema_,
path_factory_));
diff --git a/src/paimon/core/io/single_file_writer.h
b/src/paimon/core/io/single_file_writer.h
index 6db3a699..6dd1db9b 100644
--- a/src/paimon/core/io/single_file_writer.h
+++ b/src/paimon/core/io/single_file_writer.h
@@ -133,6 +133,14 @@ class SingleFileWriter : public FileWriter<T, R> {
}
protected:
+ /// Returns an error if this writer has already been closed.
+ Status CheckNotClosed() const {
+ if (PAIMON_UNLIKELY(closed_)) {
+ return Status::Invalid("Writer has already closed!");
+ }
+ return Status::OK();
+ }
+
/// Hook called after Flush() and before Finish() during Close().
/// Subclasses can override to update per-field metadata before the file
is finalized.
virtual Status BeforeFinish() {
@@ -196,9 +204,7 @@ Status SingleFileWriter<T, R>::Init(const
std::shared_ptr<FileSystem>& fs, const
template <typename T, typename R>
Status SingleFileWriter<T, R>::Write(T record) {
- if (PAIMON_UNLIKELY(closed_)) {
- return Status::Invalid("Writer has already closed!");
- }
+ PAIMON_RETURN_NOT_OK(CheckNotClosed());
ScopeGuard guard([this]() -> void { this->Abort(); });
int64_t record_count = 0;
if (!converter_) {
diff --git a/src/paimon/core/mergetree/lookup_levels.h
b/src/paimon/core/mergetree/lookup_levels.h
index 95625956..4fbb26a4 100644
--- a/src/paimon/core/mergetree/lookup_levels.h
+++ b/src/paimon/core/mergetree/lookup_levels.h
@@ -45,6 +45,8 @@ struct RemoteSstFile {
template <typename T>
class LookupLevels : public Levels::DropFileCallback {
public:
+ static constexpr const char* REMOTE_LOOKUP_FILE_SUFFIX = ".lookup";
+
static Result<std::unique_ptr<LookupLevels<T>>> Create(
const std::shared_ptr<FileSystem>& fs, const BinaryRow& partition,
int32_t bucket,
const CoreOptions& options, const std::shared_ptr<SchemaManager>&
schema_manager,
@@ -119,8 +121,6 @@ class LookupLevels : public Levels::DropFileCallback {
Result<std::shared_ptr<PersistProcessor<T>>> GetOrCreateProcessor(
int64_t schema_id, const std::string& ser_version);
- static constexpr const char* REMOTE_LOOKUP_FILE_SUFFIX = ".lookup";
-
private:
std::shared_ptr<MemoryPool> pool_;
std::shared_ptr<FileSystem> fs_;
diff --git a/src/paimon/core/mergetree/merge_tree_writer.cpp
b/src/paimon/core/mergetree/merge_tree_writer.cpp
index 748740f8..88df1b16 100644
--- a/src/paimon/core/mergetree/merge_tree_writer.cpp
+++ b/src/paimon/core/mergetree/merge_tree_writer.cpp
@@ -124,8 +124,7 @@ Status MergeTreeWriter::DoClose() {
}
}
for (const auto& file : delete_files) {
- // Keep Java parity: temporary file cleanup is quiet.
- [[maybe_unused]] auto s =
options_.GetFileSystem()->Delete(path_factory_->ToPath(file));
+ DeleteFileQuietly(file);
}
write_buffer_->Clear();
@@ -266,8 +265,7 @@ Status MergeTreeWriter::UpdateCompactResult(const
std::shared_ptr<CompactResult>
// 2. This file is not the input of upgraded.
if (!in_compact_before(file->file_name) &&
after_files.find(file->file_name) == after_files.end()) {
- auto fs = options_.GetFileSystem();
- [[maybe_unused]] auto s =
fs->Delete(path_factory_->ToPath(file));
+ DeleteFileQuietly(file);
}
} else {
compact_before_.push_back(file);
@@ -282,6 +280,14 @@ Status MergeTreeWriter::UpdateCompactResult(const
std::shared_ptr<CompactResult>
return UpdateCompactDeletionFile(compact_result->DeletionFile());
}
+void MergeTreeWriter::DeleteFileQuietly(const std::shared_ptr<DataFileMeta>&
file) const {
+ std::shared_ptr<FileSystem> fs = options_.GetFileSystem();
+ for (const std::string& path : path_factory_->CollectFiles(file)) {
+ // Keep Java parity: temporary and intermediate file cleanup is quiet.
+ [[maybe_unused]] Status status = fs->Delete(path);
+ }
+}
+
Status MergeTreeWriter::UpdateCompactDeletionFile(
const std::shared_ptr<CompactDeletionFile>& new_deletion_file) {
if (new_deletion_file) {
diff --git a/src/paimon/core/mergetree/merge_tree_writer.h
b/src/paimon/core/mergetree/merge_tree_writer.h
index 575bd76f..9f73e43b 100644
--- a/src/paimon/core/mergetree/merge_tree_writer.h
+++ b/src/paimon/core/mergetree/merge_tree_writer.h
@@ -111,6 +111,7 @@ class MergeTreeWriter : public BatchWriter {
Status TrySyncLatestCompaction(bool blocking);
Status UpdateCompactResult(const std::shared_ptr<CompactResult>&
compact_result);
Status UpdateCompactDeletionFile(const
std::shared_ptr<CompactDeletionFile>& new_deletion_file);
+ void DeleteFileQuietly(const std::shared_ptr<DataFileMeta>& file) const;
private:
MergeTreeWriter(const std::vector<std::string>& trimmed_primary_keys,
diff --git a/src/paimon/core/mergetree/merge_tree_writer_test.cpp
b/src/paimon/core/mergetree/merge_tree_writer_test.cpp
index 1885a976..f4cbaeaf 100644
--- a/src/paimon/core/mergetree/merge_tree_writer_test.cpp
+++ b/src/paimon/core/mergetree/merge_tree_writer_test.cpp
@@ -1266,7 +1266,9 @@ TEST_P(MergeTreeWriterTest, TestCloseBeforePrepareCommit)
{
TEST_P(MergeTreeWriterTest, TestCloseDeletesUncommittedFiles) {
ASSERT_OK_AND_ASSIGN(CoreOptions options,
- CoreOptions::FromMap({{Options::FILE_FORMAT,
"orc"}}));
+ CoreOptions::FromMap({{Options::FILE_FORMAT, "orc"},
+ {"file-index.bitmap.columns",
"f1"},
+
{Options::FILE_INDEX_IN_MANIFEST_THRESHOLD, "1B"}}));
auto dir = UniqueTestDirectory::Create();
ASSERT_TRUE(dir);
@@ -1291,10 +1293,13 @@ TEST_P(MergeTreeWriterTest,
TestCloseDeletesUncommittedFiles) {
ASSERT_OK(merge_writer->Compact(/*full_compaction=*/false));
std::string expected_data_file_path = dir->Str() + "/data-" + uuid +
"-0.orc";
+ std::string expected_index_file_path = expected_data_file_path + ".index";
ASSERT_TRUE(options.GetFileSystem()->Exists(expected_data_file_path).value());
+
ASSERT_TRUE(options.GetFileSystem()->Exists(expected_index_file_path).value());
ASSERT_OK(merge_writer->Close());
ASSERT_FALSE(options.GetFileSystem()->Exists(expected_data_file_path).value());
+
ASSERT_FALSE(options.GetFileSystem()->Exists(expected_index_file_path).value());
}
TEST_P(MergeTreeWriterTest, TestAutoFlush) {
@@ -1580,6 +1585,15 @@ TEST_P(MergeTreeWriterTest,
TestUpdateCompactResultDeleteIntermediateFile) {
auto file_a = CreateMeta("file_a", /*level=*/0);
auto file_x = CreateMeta("file_x", /*level=*/0);
auto file_y = CreateMeta("file_y", /*level=*/1);
+ file_x->extra_files = {"file_x.index"};
+
+ std::vector<std::string> file_x_paths = path_factory->CollectFiles(file_x);
+ ASSERT_EQ(2, file_x_paths.size());
+ for (const std::string& path : file_x_paths) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> output,
+ options.GetFileSystem()->Create(path,
/*overwrite=*/true));
+ ASSERT_OK(output->Close());
+ }
merge_writer->compact_before_ = {file_a};
merge_writer->compact_after_ = {file_x};
@@ -1590,6 +1604,9 @@ TEST_P(MergeTreeWriterTest,
TestUpdateCompactResultDeleteIntermediateFile) {
ASSERT_OK(merge_writer->UpdateCompactResult(compact_result));
ASSERT_EQ(merge_writer->compact_before_,
std::vector<std::shared_ptr<DataFileMeta>>({file_a}));
ASSERT_EQ(merge_writer->compact_after_,
std::vector<std::shared_ptr<DataFileMeta>>({file_y}));
+ for (const std::string& path : file_x_paths) {
+ ASSERT_FALSE(options.GetFileSystem()->Exists(path).value()) << path;
+ }
}
TEST_P(MergeTreeWriterTest, TestUpdateCompactResultPropagatesChangelog) {
@@ -1706,20 +1723,25 @@ TEST_P(MergeTreeWriterTest,
TestCloseSkipsDeleteForUpgradedFilesInCompactAfter)
std::string intermediate_file_name = "data-intermediate-0.orc";
std::string upgraded_file_path = dir->Str() + "/" + upgraded_file_name;
std::string intermediate_file_path = dir->Str() + "/" +
intermediate_file_name;
+ std::string upgraded_index_path = upgraded_file_path + ".index";
+ std::string intermediate_index_path = intermediate_file_path + ".index";
// Create placeholder files on disk
- ASSERT_OK_AND_ASSIGN(auto out1,
- options.GetFileSystem()->Create(upgraded_file_path,
/*overwrite=*/true));
- ASSERT_OK(out1->Close());
- ASSERT_OK_AND_ASSIGN(auto out2,
options.GetFileSystem()->Create(intermediate_file_path,
-
/*overwrite=*/true));
- ASSERT_OK(out2->Close());
+ const std::vector<std::string> placeholder_paths = {
+ upgraded_file_path, intermediate_file_path, upgraded_index_path,
intermediate_index_path};
+ for (const std::string& path : placeholder_paths) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> output,
+ options.GetFileSystem()->Create(path,
/*overwrite=*/true));
+ ASSERT_OK(output->Close());
+ }
ASSERT_TRUE(options.GetFileSystem()->Exists(upgraded_file_path).value());
ASSERT_TRUE(options.GetFileSystem()->Exists(intermediate_file_path).value());
auto upgraded_file = CreateMeta(upgraded_file_name, /*level=*/1);
auto intermediate_file = CreateMeta(intermediate_file_name, /*level=*/1);
+ upgraded_file->extra_files = {upgraded_file_name + ".index"};
+ intermediate_file->extra_files = {intermediate_file_name + ".index"};
// Setup: upgraded_file appears in both compact_before_ and compact_after_
// (simulating an upgrade operation where the file is promoted to a higher
level).
@@ -1732,10 +1754,14 @@ TEST_P(MergeTreeWriterTest,
TestCloseSkipsDeleteForUpgradedFilesInCompactAfter)
// upgraded_file should NOT be deleted (it's in compact_before_)
ASSERT_TRUE(options.GetFileSystem()->Exists(upgraded_file_path).value())
<< "Upgraded file should be preserved because it exists in
compact_before_";
+ ASSERT_TRUE(options.GetFileSystem()->Exists(upgraded_index_path).value())
+ << "Upgraded file index should be preserved with its data file";
// intermediate_file SHOULD be deleted (it's only in compact_after_)
ASSERT_FALSE(options.GetFileSystem()->Exists(intermediate_file_path).value())
<< "Intermediate file should be deleted because it's not in
compact_before_";
+
ASSERT_FALSE(options.GetFileSystem()->Exists(intermediate_index_path).value())
+ << "Intermediate file index should be deleted with its data file";
}
TEST_F(MergeTreeWriterTest, TestSpillWithSameKeyDeduplicate) {
diff --git a/src/paimon/core/operation/expire_snapshots.cpp
b/src/paimon/core/operation/expire_snapshots.cpp
index a5bd2505..9dcf1b5d 100644
--- a/src/paimon/core/operation/expire_snapshots.cpp
+++ b/src/paimon/core/operation/expire_snapshots.cpp
@@ -33,6 +33,7 @@
#include "paimon/common/utils/date_time_utils.h"
#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/io/data_file_path_factory.h"
#include "paimon/core/manifest/file_kind.h"
#include "paimon/core/manifest/manifest_entry.h"
#include "paimon/core/manifest/manifest_file.h"
@@ -137,6 +138,7 @@ Result<int32_t> ExpireSnapshots::ExpireUntil(int64_t
earliest_snapshot_id,
// Since the data file deletion information for each snapshot is recorded
in the delta part of
// the next snapshot, it is necessary to check the next snapshot.
Otherwise, its data files will
// not be deleted in this round.
+ DataFilePathFactoryCache data_file_path_factory_cache;
for (int64_t id = begin_inclusive_id + 1; id <= end_exclusive_id; id++) {
PAIMON_ASSIGN_OR_RAISE(bool exist,
snapshot_manager_->SnapshotExists(id));
if (!exist) {
@@ -144,7 +146,8 @@ Result<int32_t> ExpireSnapshots::ExpireUntil(int64_t
earliest_snapshot_id,
continue;
}
PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot,
snapshot_manager_->LoadSnapshot(id));
-
PAIMON_RETURN_NOT_OK(CleanUnusedDataFiles(snapshot.DeltaManifestList()));
+ PAIMON_RETURN_NOT_OK(
+ CleanUnusedDataFiles(snapshot.DeltaManifestList(),
&data_file_path_factory_cache));
}
// TODO(jinli.zjw): support delete changelog files
@@ -290,7 +293,8 @@ Status ExpireSnapshots::CleanUnusedManifests(const
std::string& manifest_list_na
return Status::OK();
}
-Status ExpireSnapshots::CleanUnusedDataFiles(const std::string&
manifest_list_name) {
+Status ExpireSnapshots::CleanUnusedDataFiles(
+ const std::string& manifest_list_name, DataFilePathFactoryCache*
data_file_path_factory_cache) {
std::vector<ManifestFileMeta> manifest_file_metas;
auto status = manifest_list_->Read(manifest_list_name, nullptr,
&manifest_file_metas);
if (status.ok()) {
@@ -311,13 +315,28 @@ Status ExpireSnapshots::CleanUnusedDataFiles(const
std::string& manifest_list_na
std::vector<std::future<void>> futures;
ScopeGuard guard([&futures]() { Wait(futures); });
- for (const auto& [data_file_to_delete, entry] : data_files_to_delete) {
- auto delete_file_path = data_file_to_delete;
- futures.push_back(Via(executor_.get(), [this, delete_file_path]() {
- auto status = fs_->Delete(delete_file_path);
- // delete quietly will ignore any status error
- (void)status;
- }));
+ for (const auto& [_, entry] : data_files_to_delete) {
+ std::unordered_map<int32_t, std::shared_ptr<DataFilePathFactory>>&
bucket_factories =
+ (*data_file_path_factory_cache)[entry.Partition()];
+ auto factory_iter = bucket_factories.find(entry.Bucket());
+ if (factory_iter == bucket_factories.end()) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<DataFilePathFactory>
data_file_path_factory,
+
path_factory_->CreateDataFilePathFactory(entry.Partition(), entry.Bucket()));
+ factory_iter =
+ bucket_factories.emplace(entry.Bucket(),
std::move(data_file_path_factory))
+ .first;
+ }
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory
=
+ factory_iter->second;
+ for (const std::string& delete_file_path :
+ data_file_path_factory->CollectFiles(entry.File())) {
+ futures.push_back(Via(executor_.get(), [this,
delete_file_path]() {
+ auto status = fs_->Delete(delete_file_path);
+ // delete quietly will ignore any status error
+ (void)status;
+ }));
+ }
deletion_buckets_[entry.Partition()].insert(entry.Bucket());
}
}
@@ -334,7 +353,6 @@ Status ExpireSnapshots::GetDataFilesToDelete(
if (entry.Kind() == FileKind::Add()) {
data_files_to_delete->erase(data_file_path);
} else if (entry.Kind() == FileKind::Delete()) {
- // TODO(jinli.zjw): do not support extra files
data_files_to_delete->insert({data_file_path, entry});
} else {
return Status::Invalid(
diff --git a/src/paimon/core/operation/expire_snapshots.h
b/src/paimon/core/operation/expire_snapshots.h
index 238f599e..64af99a8 100644
--- a/src/paimon/core/operation/expire_snapshots.h
+++ b/src/paimon/core/operation/expire_snapshots.h
@@ -37,6 +37,7 @@ namespace paimon {
class Snapshot;
class SnapshotManager;
class FileStorePathFactory;
+class DataFilePathFactory;
class FileSystem;
class ManifestEntry;
class ManifestList;
@@ -56,9 +57,14 @@ class ExpireSnapshots {
Result<int32_t> Expire();
private:
+ using DataFilePathFactoryCache =
+ std::unordered_map<BinaryRow,
+ std::unordered_map<int32_t,
std::shared_ptr<DataFilePathFactory>>>;
+
Result<int32_t> ExpireUntil(int64_t earliest_snapshot_id, int64_t
end_exclusive_id);
- Status CleanUnusedDataFiles(const std::string& manifest_list_name);
+ Status CleanUnusedDataFiles(const std::string& manifest_list_name,
+ DataFilePathFactoryCache*
data_file_path_factory_cache);
Status CleanUnusedManifests(const std::string& manifest_list_name,
const std::set<std::string>& skipping_sets);
Status CleanEmptyDirectories();
diff --git a/src/paimon/core/operation/expire_snapshots_test.cpp
b/src/paimon/core/operation/expire_snapshots_test.cpp
index f3472d5b..f256476d 100644
--- a/src/paimon/core/operation/expire_snapshots_test.cpp
+++ b/src/paimon/core/operation/expire_snapshots_test.cpp
@@ -28,6 +28,7 @@
#include "paimon/common/data/binary_row_writer.h"
#include "paimon/core/core_options.h"
#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/data_file_path_factory.h"
#include "paimon/core/manifest/file_kind.h"
#include "paimon/core/manifest/manifest_entry.h"
#include "paimon/core/manifest/manifest_file.h"
@@ -72,7 +73,9 @@ class ExpireSnapshotsTest : public testing::Test {
FieldMapping::GetPartitionSchema(schema_,
partition_keys_));
fs_ = std::make_shared<LocalFileSystem>();
- test_data_path_ = "tmp";
+ directory_ = UniqueTestDirectory::Create();
+ ASSERT_TRUE(directory_);
+ test_data_path_ = directory_->Str();
path_factory_ = CreateFactory(test_data_path_);
std::shared_ptr<FileFormat> manifest_format =
options.GetManifestFormat();
@@ -132,6 +135,13 @@ class ExpireSnapshotsTest : public testing::Test {
ManifestEntry CreateManifestEntry(const std::string& file_name, int32_t
bucket,
const FileKind& kind) const {
+ return CreateManifestEntry(file_name, bucket, kind,
+ std::vector<std::optional<std::string>>());
+ }
+
+ ManifestEntry CreateManifestEntry(
+ const std::string& file_name, int32_t bucket, const FileKind& kind,
+ const std::vector<std::optional<std::string>>& extra_files) const {
int32_t arity = 2;
BinaryRow row(arity);
BinaryRowWriter writer(&row, 20, mem_pool_.get());
@@ -143,7 +153,7 @@ class ExpireSnapshotsTest : public testing::Test {
file_name, 1024, 8, DataFileMeta::EmptyMinKey(),
DataFileMeta::EmptyMaxKey(),
SimpleStats::EmptyStats(), SimpleStats::EmptyStats(),
/*min_seq_no=*/16,
/*max_seq_no=*/32,
- /*schema_id=*/1, /*level=*/2,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*schema_id=*/1, /*level=*/2, extra_files,
/*creation_time=*/Timestamp(0, 0), /*delete_row_count=*/3,
/*embedded_index=*/nullptr, /*file_source=*/std::nullopt,
/*external_path=*/std::nullopt,
@@ -154,6 +164,7 @@ class ExpireSnapshotsTest : public testing::Test {
private:
std::string test_data_path_;
+ std::unique_ptr<UniqueTestDirectory> directory_;
std::vector<std::string> partition_keys_;
std::shared_ptr<arrow::Schema> schema_;
std::shared_ptr<arrow::Schema> partition_schema_;
@@ -253,4 +264,36 @@ TEST_F(ExpireSnapshotsTest, TestGetDataFileToDelete) {
}
}
+TEST_F(ExpireSnapshotsTest, TestCleanUnusedDataFileDeletesExtraFiles) {
+ auto mgr = std::make_shared<SnapshotManager>(fs_, test_data_path_);
+ ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
+ ExpireSnapshots expire(mgr, path_factory_, manifest_list_, manifest_file_,
fs_,
+ options.GetExpireConfig(),
options.RealtimeEnabled(), executor_);
+
+ ManifestEntry entry = CreateManifestEntry("file1", /*bucket=*/0,
FileKind::Delete(),
+ {"file1.index", "file1.lookup"});
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<DataFilePathFactory> data_file_path_factory,
+ path_factory_->CreateDataFilePathFactory(entry.Partition(),
entry.Bucket()));
+ std::vector<std::string> files =
data_file_path_factory->CollectFiles(entry.File());
+ ASSERT_EQ(3, files.size());
+ for (const std::string& file : files) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> output,
+ fs_->Create(file, /*overwrite=*/true));
+ ASSERT_OK(output->Close());
+ }
+
+ ASSERT_OK_AND_ASSIGN(std::vector<ManifestFileMeta> manifest_files,
+ manifest_file_->Write({entry}));
+ std::pair<std::string, int64_t> manifest_list;
+ ASSERT_OK_AND_ASSIGN(manifest_list, manifest_list_->Write(manifest_files));
+ ExpireSnapshots::DataFilePathFactoryCache data_file_path_factory_cache;
+ ASSERT_OK(expire.CleanUnusedDataFiles(manifest_list.first,
&data_file_path_factory_cache));
+
+ for (const std::string& file : files) {
+ ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(file));
+ ASSERT_FALSE(exists) << file;
+ }
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/operation/file_store_commit_impl.cpp
b/src/paimon/core/operation/file_store_commit_impl.cpp
index 8268e811..736825f7 100644
--- a/src/paimon/core/operation/file_store_commit_impl.cpp
+++ b/src/paimon/core/operation/file_store_commit_impl.cpp
@@ -230,8 +230,9 @@ Status FileStoreCommitImpl::Abort(
append_data_files(compact_increment.ChangelogFiles());
for (const auto& file : data_files_to_delete) {
// Best-effort cleanup: ignore delete failures, aligning with Java
deleteQuietly.
- [[maybe_unused]] Status status =
- fs_->Delete(data_file_path_factory->ToPath(file),
/*recursive=*/false);
+ for (const std::string& path :
data_file_path_factory->CollectFiles(file)) {
+ [[maybe_unused]] Status status = fs_->Delete(path,
/*recursive=*/false);
+ }
}
std::vector<std::shared_ptr<IndexFileMeta>> index_files_to_delete;
diff --git a/src/paimon/core/operation/file_store_commit_impl_test.cpp
b/src/paimon/core/operation/file_store_commit_impl_test.cpp
index a5079d7d..d55f21cb 100644
--- a/src/paimon/core/operation/file_store_commit_impl_test.cpp
+++ b/src/paimon/core/operation/file_store_commit_impl_test.cpp
@@ -1522,6 +1522,8 @@ TEST_F(FileStoreCommitImplTest,
TestAbortDeletesDataAndIndexFiles) {
const int32_t bucket = 0;
auto new_data_file = CreateAppendDataFileMeta("abort-new-data", 1);
auto compact_data_file = CreateAppendDataFileMeta("abort-compact-data", 1);
+ new_data_file->extra_files = {"abort-new-data.index"};
+ compact_data_file->extra_files = {"abort-compact-data.index"};
auto new_index_file = CreateIndexFileMeta("abort-new-index");
auto compact_index_file = CreateIndexFileMeta("abort-compact-index");
@@ -1540,9 +1542,12 @@ TEST_F(FileStoreCommitImplTest,
TestAbortDeletesDataAndIndexFiles) {
commit_impl->path_factory_->CreateDataFilePathFactory(partition, bucket));
ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexPathFactory> index_pf,
commit_impl->path_factory_->CreateIndexFileFactory(partition, bucket));
- std::vector<std::string> paths = {
- data_pf->ToPath(new_data_file), data_pf->ToPath(compact_data_file),
- index_pf->ToPath(new_index_file),
index_pf->ToPath(compact_index_file)};
+ std::vector<std::string> paths = {data_pf->ToPath(new_data_file),
+
data_pf->ToPath(new_data_file->extra_files[0].value()),
+ data_pf->ToPath(compact_data_file),
+
data_pf->ToPath(compact_data_file->extra_files[0].value()),
+ index_pf->ToPath(new_index_file),
+ index_pf->ToPath(compact_index_file)};
for (const auto& path : paths) {
ASSERT_OK(file_system_->WriteFile(path, /*content=*/"",
/*overwrite=*/false));
ASSERT_OK_AND_ASSIGN(bool exist, file_system_->Exists(path));
diff --git a/src/paimon/core/operation/orphan_files_cleaner_impl.cpp
b/src/paimon/core/operation/orphan_files_cleaner_impl.cpp
index 015d1340..f42708c8 100644
--- a/src/paimon/core/operation/orphan_files_cleaner_impl.cpp
+++ b/src/paimon/core/operation/orphan_files_cleaner_impl.cpp
@@ -31,10 +31,12 @@
#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/scope_guard.h"
#include "paimon/common/utils/string_utils.h"
+#include "paimon/core/io/data_file_path_factory.h"
#include "paimon/core/manifest/manifest_entry.h"
#include "paimon/core/manifest/manifest_file.h"
#include "paimon/core/manifest/manifest_file_meta.h"
#include "paimon/core/manifest/manifest_list.h"
+#include "paimon/core/mergetree/lookup_levels.h"
#include "paimon/core/operation/commit/realtime_commit_properties.h"
#include "paimon/core/operation/metrics/clean_metrics.h"
#include "paimon/core/snapshot.h"
@@ -86,6 +88,14 @@ bool OrphanFilesCleanerImpl::SupportToClean(const
std::string& file_name) {
return true;
}
}
+ static std::vector<std::string> supported_extra_file_suffixes = {
+ DataFilePathFactory::INDEX_PATH_SUFFIX,
LookupLevels<bool>::REMOTE_LOOKUP_FILE_SUFFIX};
+ for (const std::string& suffix : supported_extra_file_suffixes) {
+ if (StringUtils::StartsWith(file_name, "data-") &&
+ StringUtils::EndsWith(file_name, suffix)) {
+ return true;
+ }
+ }
return StringUtils::EndsWith(file_name, ".offsets");
}
@@ -319,6 +329,12 @@ Result<std::set<std::string>>
OrphanFilesCleanerImpl::GetUsedFilesBySnapshot(
manifest.FileName(), /*filter=*/nullptr, &manifest_entries));
for (const auto& manifest_entry : manifest_entries) {
used_files.insert(manifest_entry.FileName());
+ for (const std::optional<std::string>& extra_file :
+ manifest_entry.File()->extra_files) {
+ if (extra_file) {
+ used_files.insert(extra_file.value());
+ }
+ }
}
}
diff --git a/src/paimon/core/operation/orphan_files_cleaner_test.cpp
b/src/paimon/core/operation/orphan_files_cleaner_test.cpp
index 7749b692..b2774b20 100644
--- a/src/paimon/core/operation/orphan_files_cleaner_test.cpp
+++ b/src/paimon/core/operation/orphan_files_cleaner_test.cpp
@@ -53,8 +53,10 @@ TEST(OrphanFilesCleanerTest, TestSupportToClean) {
ASSERT_FALSE(OrphanFilesCleanerImpl::SupportToClean("bucket-0"));
ASSERT_FALSE(OrphanFilesCleanerImpl::SupportToClean(
"changelog-ce64d06d-c4cd-456b-a1b3-ae570042620f-0.parquet"));
- ASSERT_FALSE(OrphanFilesCleanerImpl::SupportToClean(
+ ASSERT_TRUE(OrphanFilesCleanerImpl::SupportToClean(
"data-5515726b-0f0f-4556-a942-e795e9f94c4a-0.orc.index"));
+ ASSERT_TRUE(OrphanFilesCleanerImpl::SupportToClean(
+
"data-5515726b-0f0f-4556-a942-e795e9f94c4a-0.orc.128.processor.v1.lookup"));
ASSERT_FALSE(
OrphanFilesCleanerImpl::SupportToClean("index-aa60193d-d7cd-434f-bc1a-c1adb210e1f7-0"));
ASSERT_FALSE(
@@ -123,6 +125,35 @@ TEST(OrphanFilesCleanerTest, TestTableWithIndex) {
ASSERT_TRUE(cleaned_paths.empty());
}
+TEST(OrphanFilesCleanerTest, TestCleanOrphanExtraFiles) {
+ std::string test_data_path =
+ paimon::test::GetDataDir() +
"/orc/append_with_bsi.db/append_with_bsi/";
+ auto dir = UniqueTestDirectory::Create();
+ std::string table_path = dir->Str();
+ ASSERT_TRUE(TestUtil::CopyDirectory(test_data_path, table_path));
+ auto file_system = std::make_shared<LocalFileSystem>();
+ std::string orphan_index_path =
+ PathUtil::JoinPath(table_path, "bucket-0/data-orphan-0.orc.index");
+ std::string orphan_lookup_path =
+ PathUtil::JoinPath(table_path,
"bucket-0/data-orphan-0.orc.128.processor.v1.lookup");
+ ASSERT_OK(file_system->WriteFile(orphan_index_path, "orphan",
/*overwrite=*/true));
+ ASSERT_OK(file_system->WriteFile(orphan_lookup_path, "orphan",
/*overwrite=*/true));
+
+ CleanContextBuilder clean_context_builder(table_path);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<CleanContext> clean_context,
+ clean_context_builder.AddOption(Options::FILE_SYSTEM,
"local")
+
.WithOlderThanMs(std::numeric_limits<int64_t>::max())
+ .Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OrphanFilesCleaner> cleaner,
+ OrphanFilesCleaner::Create(std::move(clean_context)));
+ ASSERT_OK_AND_ASSIGN(std::set<std::string> cleaned_paths,
cleaner->Clean());
+ ASSERT_EQ(cleaned_paths, std::set<std::string>({orphan_index_path,
orphan_lookup_path}));
+ for (const std::string& path : cleaned_paths) {
+ ASSERT_OK_AND_ASSIGN(bool exists, file_system->Exists(path));
+ ASSERT_FALSE(exists) << path;
+ }
+}
+
TEST(OrphanFilesCleanerTest, TestTableWithBrokenSnapshot) {
std::string test_data_path = paimon::test::GetDataDir() +
"/orc/append_09.db/append_09/";
auto file_system = std::make_shared<LocalFileSystem>();
diff --git a/test/inte/clean_inte_test.cpp b/test/inte/clean_inte_test.cpp
index 3f7f353e..ec418d2c 100644
--- a/test/inte/clean_inte_test.cpp
+++ b/test/inte/clean_inte_test.cpp
@@ -671,8 +671,8 @@ TEST_F(CleanInteTest, TestOrphanFilesClean) {
.Finish());
ASSERT_OK_AND_ASSIGN(auto cleaner,
OrphanFilesCleaner::Create(std::move(clean_context)));
ASSERT_OK_AND_ASSIGN(std::set<std::string> cleaned_paths,
cleaner->Clean());
- ASSERT_TRUE(CheckEqual(
- cleaned_paths, {"data-orphan2.orc", "manifest-orphan",
".manifest-orphan.uuid.tmp"}));
+ ASSERT_TRUE(CheckEqual(cleaned_paths, {"data-orphan2.orc",
"data-orphan2.orc.index",
+ "manifest-orphan",
".manifest-orphan.uuid.tmp"}));
}
}
@@ -782,7 +782,8 @@ TEST_F(CleanInteTest,
TestOrphanFilesCleanWithFileRetainCondition) {
.Finish());
ASSERT_OK_AND_ASSIGN(auto cleaner,
OrphanFilesCleaner::Create(std::move(clean_context)));
ASSERT_OK_AND_ASSIGN(std::set<std::string> cleaned_paths,
cleaner->Clean());
- ASSERT_TRUE(CheckEqual(cleaned_paths, {"manifest-orphan",
".manifest-orphan.uuid.tmp"}));
+ ASSERT_TRUE(CheckEqual(
+ cleaned_paths, {"data-orphan2.orc.index", "manifest-orphan",
".manifest-orphan.uuid.tmp"}));
}
TEST_F(CleanInteTest, TestOrphanFilesCleanWithIOException) {
@@ -921,7 +922,8 @@ TEST_F(CleanInteTest, TestOrphanFilesCleanWithIOException) {
}
ASSERT_OK(clean_result);
- std::set<std::string> expected_orphans = {"data-orphan2.orc",
"manifest-orphan",
+ std::set<std::string> expected_orphans = {"data-orphan2.orc",
"data-orphan2.orc.index",
+ "manifest-orphan",
".manifest-orphan.uuid.tmp"};
// In the first clean, IO errors may already have been triggered in
// TryBestListingDirs or MinimalTryBestListingDirs. Those errors
are handled quietly,
