Skip to content

Commit 57677ea

Browse files
lxy-9602lxy-9602
authored
feat: Use IOManager channel for lookup file creation and make Lucene tmp dir configurable (alibaba#251)
1 parent e2a73de commit 57677ea

9 files changed

Lines changed: 79 additions & 61 deletions

File tree

‎include/paimon/predicate/literal.h‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,9 @@ class PAIMON_EXPORT Literal {
3737
explicit Literal(FieldType type);
3838

3939
/// Creates a literal from a typed value.
40-
/// The template parameter T must be compatible with one of the supported field types.
41-
/// @tparam T The C++ type of the value (must match a supported FieldType).
40+
/// The template parameter T must be compatible with one of the supported field types
41+
/// (must match a supported FieldType).
42+
/// T can be bool, int8_t, int16_t, int32_t, int64_t, float, double, Timestamp and Decimal.
4243
/// @param val The value to store in the literal.
4344
template <typename T>
4445
explicit Literal(const T& val);

‎src/paimon/core/disk/io_manager.h‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -35,14 +35,6 @@ class IOManager {
3535
return tmp_dir_;
3636
}
3737

38-
Result<std::string> GenerateTempFilePath(const std::string& prefix) const {
39-
std::string uuid;
40-
if (!UUID::Generate(&uuid)) {
41-
return Status::Invalid("generate uuid for io manager tmp path failed.");
42-
}
43-
return PathUtil::JoinPath(tmp_dir_, prefix + "-" + uuid + std::string(kSuffix));
44-
}
45-
4638
Result<FileIOChannel::ID> CreateChannel() {
4739
PAIMON_ASSIGN_OR_RAISE(auto* manager, GetFileChannelManager());
4840
return manager->CreateChannel();
@@ -74,7 +66,6 @@ class IOManager {
7466
return file_channel_manager_.get();
7567
}
7668

77-
static constexpr char kSuffix[] = ".channel";
7869
static constexpr char kDirNamePrefix[] = "io";
7970
std::string tmp_dir_;
8071
std::shared_ptr<FileSystem> file_system_;

‎src/paimon/core/disk/io_manager_test.cpp‎

Lines changed: 0 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -34,35 +34,6 @@ TEST(IOManagerTest, CreateShouldReturnManagerWithGivenTempDir) {
3434
ASSERT_EQ(manager->GetTempDir(), tmp_dir->Str());
3535
}
3636

37-
TEST(IOManagerTest, GenerateTempFilePathShouldContainPrefixAndSuffix) {
38-
auto tmp_dir = UniqueTestDirectory::Create();
39-
const std::string prefix = "spill";
40-
41-
auto manager = std::make_unique<IOManager>(tmp_dir->Str(), tmp_dir->GetFileSystem());
42-
ASSERT_OK_AND_ASSIGN(std::string temp_path, manager->GenerateTempFilePath(prefix));
43-
44-
std::string expected_prefix = PathUtil::JoinPath(tmp_dir->Str(), "");
45-
ASSERT_TRUE(StringUtils::StartsWith(temp_path, expected_prefix));
46-
47-
std::string file_name = PathUtil::GetName(temp_path);
48-
std::string file_prefix = prefix + "-";
49-
ASSERT_TRUE(StringUtils::StartsWith(file_name, file_prefix));
50-
51-
const std::string suffix = ".channel";
52-
ASSERT_GE(file_name.size(), file_prefix.size() + suffix.size() + 1);
53-
ASSERT_TRUE(StringUtils::EndsWith(file_name, suffix));
54-
}
55-
56-
TEST(IOManagerTest, GenerateTempFilePathShouldBeDifferentAcrossCalls) {
57-
auto tmp_dir = UniqueTestDirectory::Create();
58-
auto manager = std::make_unique<IOManager>(tmp_dir->Str(), tmp_dir->GetFileSystem());
59-
60-
ASSERT_OK_AND_ASSIGN(std::string path1, manager->GenerateTempFilePath("spill"));
61-
ASSERT_OK_AND_ASSIGN(std::string path2, manager->GenerateTempFilePath("spill"));
62-
63-
ASSERT_NE(path1, path2);
64-
}
65-
6637
TEST(IOManagerTest, CreateChannelShouldReturnValidAndUniquePaths) {
6738
auto tmp_dir = UniqueTestDirectory::Create();
6839
auto manager = std::make_shared<IOManager>(tmp_dir->Str(), tmp_dir->GetFileSystem());

‎src/paimon/core/mergetree/lookup_levels.cpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -219,7 +219,8 @@ Result<std::shared_ptr<LookupFile>> LookupLevels<T>::CreateLookupFile(
219219
PAIMON_ASSIGN_OR_RAISE(
220220
std::string prefix,
221221
LookupFile::LocalFilePrefix(partition_schema_, partition_, bucket_, file->file_name));
222-
PAIMON_ASSIGN_OR_RAISE(std::string kv_file_path, io_manager_->GenerateTempFilePath(prefix));
222+
PAIMON_ASSIGN_OR_RAISE(FileIOChannel::ID channel_id, io_manager_->CreateChannel(prefix));
223+
std::string kv_file_path = channel_id.GetPath();
223224

224225
int64_t schema_id = table_schema_->Id();
225226
std::string file_ser_version = serializer_factory_->Version();

‎src/paimon/core/mergetree/lookup_levels_test.cpp‎

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ class LookupLevelsTest : public testing::Test {
5656
dir_ = UniqueTestDirectory::Create("local");
5757
fs_ = dir_->GetFileSystem();
5858
noop_compact_manager_ = std::make_shared<NoopCompactManager>();
59+
io_manager_ = std::make_shared<IOManager>(tmp_dir_->Str(), tmp_dir_->GetFileSystem());
5960
}
6061

6162
void TearDown() override {}
@@ -145,7 +146,6 @@ class LookupLevelsTest : public testing::Test {
145146
PAIMON_ASSIGN_OR_RAISE(auto table_schema, schema_manager->ReadSchema(0));
146147
PAIMON_ASSIGN_OR_RAISE(CoreOptions options, CoreOptions::FromMap(table_schema->Options()));
147148

148-
auto io_manager = std::make_shared<IOManager>(tmp_dir_->Str(), tmp_dir_->GetFileSystem());
149149
auto processor_factory =
150150
std::make_shared<PersistValueAndPosProcessor::Factory>(arrow_schema_);
151151
auto serializer_factory = std::make_shared<DefaultLookupSerializerFactory>();
@@ -161,8 +161,8 @@ class LookupLevelsTest : public testing::Test {
161161
options.GetLookupCacheFileRetentionMs(), options.GetLookupCacheMaxDiskSize());
162162
}
163163
return LookupLevels<PositionedKeyValue>::Create(
164-
fs_, BinaryRow::EmptyRow(), /*bucket=*/0, options, schema_manager,
165-
std::move(io_manager), path_factory, table_schema, levels,
164+
fs_, BinaryRow::EmptyRow(), /*bucket=*/0, options, schema_manager, io_manager_,
165+
path_factory, table_schema, levels,
166166
/*dv_factory=*/{}, processor_factory, serializer_factory, lookup_store_factory,
167167
lookup_file_cache,
168168
/*remote_lookup_file_manager=*/nullptr, pool_);
@@ -176,6 +176,7 @@ class LookupLevelsTest : public testing::Test {
176176
std::unique_ptr<UniqueTestDirectory> dir_;
177177
std::shared_ptr<FileSystem> fs_;
178178
std::shared_ptr<NoopCompactManager> noop_compact_manager_;
179+
std::shared_ptr<IOManager> io_manager_;
179180
};
180181

181182
TEST_F(LookupLevelsTest, TestMultiLevels) {
@@ -234,13 +235,17 @@ TEST_F(LookupLevelsTest, TestMultiLevels) {
234235
// test lookup file in tmp dir
235236
std::vector<std::unique_ptr<BasicFileStatus>> file_status_list;
236237
ASSERT_OK(fs_->ListDir(tmp_dir_->Str(), &file_status_list));
238+
ASSERT_EQ(file_status_list.size(), 1);
239+
auto channel_dir = file_status_list[0]->GetPath();
240+
file_status_list.clear();
241+
ASSERT_OK(fs_->ListDir(channel_dir, &file_status_list));
237242
ASSERT_EQ(file_status_list.size(), 2);
238243
ASSERT_EQ(levels->drop_file_callbacks_.size(), 1);
239244

240245
// test close will rm local lookup file
241246
ASSERT_OK(lookup_levels->Close());
242247
file_status_list.clear();
243-
ASSERT_OK(fs_->ListDir(tmp_dir_->Str(), &file_status_list));
248+
ASSERT_OK(fs_->ListDir(channel_dir, &file_status_list));
244249
ASSERT_TRUE(file_status_list.empty());
245250
ASSERT_TRUE(levels->drop_file_callbacks_.empty());
246251
ASSERT_EQ(lookup_levels->lookup_file_cache_->Size(), 0);
@@ -732,6 +737,10 @@ TEST_F(LookupLevelsTest, TestLookupFileCacheIntegration) {
732737
// Collect local file paths for later verification
733738
std::vector<std::unique_ptr<BasicFileStatus>> tmp_files;
734739
ASSERT_OK(fs_->ListDir(tmp_dir_->Str(), &tmp_files));
740+
ASSERT_EQ(tmp_files.size(), 1);
741+
auto channel_dir = tmp_files[0]->GetPath();
742+
tmp_files.clear();
743+
ASSERT_OK(fs_->ListDir(channel_dir, &tmp_files));
735744
ASSERT_EQ(tmp_files.size(), 3);
736745

737746
// --- Scenario 4: Close instance 1 invalidates only its own files ---
@@ -750,7 +759,7 @@ TEST_F(LookupLevelsTest, TestLookupFileCacheIntegration) {
750759

751760
// All local lookup files should be deleted
752761
tmp_files.clear();
753-
ASSERT_OK(fs_->ListDir(tmp_dir_->Str(), &tmp_files));
762+
ASSERT_OK(fs_->ListDir(channel_dir, &tmp_files));
754763
ASSERT_TRUE(tmp_files.empty());
755764
}
756765

‎src/paimon/global_index/lucene/lucene_defs.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@ static inline const char kLuceneReadBufferSize[] = "read.buffer-size";
3232
// default is false
3333
static inline const char kLuceneWriteOmitTermFreqAndPositions[] =
3434
"write.omit-term-freq-and-position";
35+
// no default value, must explicit set
36+
static inline const char kLuceneWriteTmpDir[] = "write.tmp.directory";
3537

3638
static inline const char kJiebaDictDirEnv[] = "PAIMON_JIEBA_DICT_DIR";
3739

‎src/paimon/global_index/lucene/lucene_global_index_test.cpp‎

Lines changed: 49 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,8 @@ class LuceneGlobalIndexTest : public ::testing::Test,
6969
const std::shared_ptr<arrow::DataType>& data_type,
7070
const std::map<std::string, std::string>& options,
7171
const std::shared_ptr<arrow::Array>& array,
72-
const Range& expected_range) const {
72+
const Range& expected_range,
73+
const std::string& tmp_dir) const {
7374
auto global_index = std::make_shared<LuceneGlobalIndex>(options);
7475
auto path_factory = std::make_shared<FakeIndexPathFactory>(index_root);
7576
auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_, path_factory);
@@ -84,12 +85,25 @@ class LuceneGlobalIndexTest : public ::testing::Test,
8485
std::iota(row_ids.begin(), row_ids.end(), 0);
8586
PAIMON_RETURN_NOT_OK(global_writer->AddBatch(&c_array, std::move(row_ids)));
8687
PAIMON_ASSIGN_OR_RAISE(auto result_metas, global_writer->Finish());
88+
89+
// check tmp dir
90+
std::vector<std::unique_ptr<BasicFileStatus>> file_status_list;
91+
EXPECT_OK(fs_->ListDir(tmp_dir, &file_status_list));
92+
EXPECT_EQ(file_status_list.size(), 1);
93+
8794
// check meta
8895
EXPECT_EQ(result_metas.size(), 1);
8996
auto file_name = PathUtil::GetName(result_metas[0].file_path);
9097
EXPECT_TRUE(StringUtils::StartsWith(file_name, "lucene-fts-global-index-"));
9198
EXPECT_TRUE(StringUtils::EndsWith(file_name, ".index"));
9299
EXPECT_TRUE(result_metas[0].metadata);
100+
101+
// after reset writer, rm tmp files
102+
global_writer.reset();
103+
file_status_list.clear();
104+
EXPECT_OK(fs_->ListDir(tmp_dir, &file_status_list));
105+
EXPECT_TRUE(file_status_list.empty());
106+
93107
return result_metas[0];
94108
}
95109

@@ -128,14 +142,16 @@ class LuceneGlobalIndexTest : public ::testing::Test,
128142

129143
TEST_P(LuceneGlobalIndexTest, TestSimple) {
130144
int32_t read_buffer_size = GetParam();
131-
132145
auto test_root_dir = paimon::test::UniqueTestDirectory::Create();
133146
ASSERT_TRUE(test_root_dir);
147+
auto tmp_dir = paimon::test::UniqueTestDirectory::Create();
148+
ASSERT_TRUE(tmp_dir);
134149
std::string test_root = test_root_dir->Str();
135150

136151
std::map<std::string, std::string> options = {
137152
{"lucene-fts.write.omit-term-freq-and-position", "false"},
138-
{"lucene-fts.read.buffer-size", std::to_string(read_buffer_size)}};
153+
{"lucene-fts.read.buffer-size", std::to_string(read_buffer_size)},
154+
{"lucene-fts.write.tmp.directory", tmp_dir->Str()}};
139155
std::shared_ptr<arrow::Array> array = arrow::ipc::internal::json::ArrayFromJSON(data_type_,
140156
R"([
141157
["This is an test document."],
@@ -146,11 +162,13 @@ TEST_P(LuceneGlobalIndexTest, TestSimple) {
146162
.ValueOrDie();
147163

148164
// write index
149-
ASSERT_OK_AND_ASSIGN(auto meta,
150-
WriteGlobalIndex(test_root, data_type_, options, array, Range(0, 3)));
165+
ASSERT_OK_AND_ASSIGN(auto meta, WriteGlobalIndex(test_root, data_type_, options, array,
166+
Range(0, 3), tmp_dir->Str()));
151167
if (read_buffer_size == 10) {
152-
ASSERT_EQ(std::string(meta.metadata->data(), meta.metadata->size()),
153-
R"({"read.buffer-size":"10","write.omit-term-freq-and-position":"false"})");
168+
ASSERT_EQ(
169+
std::string(meta.metadata->data(), meta.metadata->size()),
170+
R"({"read.buffer-size":"10","write.omit-term-freq-and-position":"false","write.tmp.directory":")" +
171+
tmp_dir->Str() + R"("})");
154172
}
155173

156174
// create reader
@@ -293,13 +311,15 @@ TEST_P(LuceneGlobalIndexTest, TestSimpleChinese) {
293311

294312
auto test_root_dir = paimon::test::UniqueTestDirectory::Create();
295313
ASSERT_TRUE(test_root_dir);
314+
auto tmp_dir = paimon::test::UniqueTestDirectory::Create();
315+
ASSERT_TRUE(tmp_dir);
296316
std::string test_root = test_root_dir->Str();
297317

298318
std::map<std::string, std::string> options = {
299319
{"lucene-fts.write.omit-term-freq-and-position", "false"},
300320
{"lucene-fts.read.buffer-size", std::to_string(read_buffer_size)},
301321
{"lucene-fts.jieba.tokenize-mode", "query"},
302-
};
322+
{"lucene-fts.write.tmp.directory", tmp_dir->Str()}};
303323

304324
std::shared_ptr<arrow::Array> array = arrow::ipc::internal::json::ArrayFromJSON(data_type_,
305325
R"([
@@ -312,12 +332,13 @@ TEST_P(LuceneGlobalIndexTest, TestSimpleChinese) {
312332
.ValueOrDie();
313333

314334
// write index
315-
ASSERT_OK_AND_ASSIGN(auto meta,
316-
WriteGlobalIndex(test_root, data_type_, options, array, Range(0, 4)));
335+
ASSERT_OK_AND_ASSIGN(auto meta, WriteGlobalIndex(test_root, data_type_, options, array,
336+
Range(0, 4), tmp_dir->Str()));
317337
if (read_buffer_size == 10) {
318338
ASSERT_EQ(
319339
std::string(meta.metadata->data(), meta.metadata->size()),
320-
R"({"jieba.tokenize-mode":"query","read.buffer-size":"10","write.omit-term-freq-and-position":"false"})");
340+
R"({"jieba.tokenize-mode":"query","read.buffer-size":"10","write.omit-term-freq-and-position":"false","write.tmp.directory":")" +
341+
tmp_dir->Str() + R"("})");
321342
}
322343

323344
// create reader
@@ -406,6 +427,23 @@ TEST_P(LuceneGlobalIndexTest, TestSimpleChinese) {
406427
}
407428
}
408429

430+
TEST_F(LuceneGlobalIndexTest, TestInvalidWithoutTmpDir) {
431+
auto test_root_dir = paimon::test::UniqueTestDirectory::Create();
432+
ASSERT_TRUE(test_root_dir);
433+
std::string test_root = test_root_dir->Str();
434+
435+
std::map<std::string, std::string> options = {
436+
{"lucene-fts.write.omit-term-freq-and-position", "false"}};
437+
std::shared_ptr<arrow::Array> array = arrow::ipc::internal::json::ArrayFromJSON(data_type_,
438+
R"([
439+
["This is an test document."]
440+
])")
441+
.ValueOrDie();
442+
443+
// write index
444+
ASSERT_NOK_WITH_MSG(WriteGlobalIndex(test_root, data_type_, options, array, Range(0, 0), ""),
445+
"key write.tmp.directory does not exist in map");
446+
}
409447
INSTANTIATE_TEST_SUITE_P(ReadBufferSize, LuceneGlobalIndexTest,
410448
::testing::ValuesIn(std::vector<int32_t>({10, 100, 1024})));
411449

‎src/paimon/global_index/lucene/lucene_global_index_writer.cpp‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -58,9 +58,11 @@ Result<std::shared_ptr<LuceneGlobalIndexWriter>> LuceneGlobalIndexWriter::Create
5858
if (!UUID::Generate(&uuid)) {
5959
return Status::Invalid("generate uuid for lucene tmp path failed.");
6060
}
61-
// create a local tmp path
62-
std::string tmp_path = PathUtil::JoinPath(std::filesystem::temp_directory_path().string(),
63-
"paimon-lucene-" + uuid);
61+
// get local tmp path
62+
PAIMON_ASSIGN_OR_RAISE(std::string tmp_dir, OptionsUtils::GetValueFromMap<std::string>(
63+
options, std::string(kLuceneWriteTmpDir)));
64+
std::string tmp_path = PathUtil::JoinPath(tmp_dir, "paimon-lucene-" + uuid);
65+
6466
auto lucene_dir = Lucene::FSDirectory::open(LuceneUtils::StringToWstring(tmp_path),
6567
Lucene::NoLockFactory::getNoLockFactory());
6668
// TODO(xinyu.lxy): support other tokenizer

‎test/inte/global_index_test.cpp‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2432,8 +2432,11 @@ TEST_P(GlobalIndexTest, TestDataEvolutionBatchScanWithRangeBitmapAndBitmap) {
24322432
TEST_P(GlobalIndexTest, TestLuceneWriteCommitScanReadIndexWithScore) {
24332433
arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
24342434
arrow::field("f1", arrow::int32())};
2435+
auto tmp_dir = paimon::test::UniqueTestDirectory::Create();
2436+
ASSERT_TRUE(tmp_dir);
24352437
std::map<std::string, std::string> lucene_options = {
2436-
{"lucene-fts.write.omit-term-freq-and-position", "false"}};
2438+
{"lucene-fts.write.omit-term-freq-and-position", "false"},
2439+
{"lucene-fts.write.tmp.directory", tmp_dir->Str()}};
24372440
auto schema = arrow::schema(fields);
24382441
std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
24392442
{Options::FILE_FORMAT, file_format_},

0 commit comments

Comments
 (0)