Skip to content

Commit 287c0d4

Browse files
zjw1111zjw1111
authored
fix(fs): normalize local filesystem paths and some refactoring (alibaba#335)
* fix(fs): normalize local paths * fix(fs): simplify external path position update * fix(schema): treat rowkind as special field * refactor(fs): return LocalFile create as unique ptr
1 parent 2e79a5b commit 287c0d4

13 files changed

Lines changed: 392 additions & 213 deletions

‎src/paimon/common/fs/external_path_provider.h‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
#pragma once
1818

19+
#include <atomic>
1920
#include <cstddef>
2021
#include <memory>
2122
#include <random>
@@ -45,12 +46,9 @@ class ExternalPathProvider {
4546
///
4647
/// @return the next external data path
4748
std::string GetNextExternalDataPath(const std::string& file_name) {
48-
position_++;
49-
if (position_ == external_table_paths_.size()) {
50-
position_ = 0;
51-
}
49+
size_t position = (++position_) % external_table_paths_.size();
5250
return PathUtil::JoinPath(
53-
PathUtil::JoinPath(external_table_paths_[position_], relative_bucket_path_), file_name);
51+
PathUtil::JoinPath(external_table_paths_[position], relative_bucket_path_), file_name);
5452
}
5553

5654
private:
@@ -66,6 +64,6 @@ class ExternalPathProvider {
6664
private:
6765
std::vector<std::string> external_table_paths_;
6866
std::string relative_bucket_path_;
69-
size_t position_;
67+
std::atomic<size_t> position_;
7068
};
7169
} // namespace paimon

‎src/paimon/common/fs/external_path_provider_test.cpp‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,11 @@
1616

1717
#include "paimon/common/fs/external_path_provider.h"
1818

19+
#include <cstdint>
20+
#include <future>
21+
#include <mutex>
1922
#include <set>
23+
#include <vector>
2024

2125
#include "gtest/gtest.h"
2226
#include "paimon/testing/utils/testharness.h"
@@ -62,4 +66,49 @@ TEST(ExternalPathProviderTest, TestGetNextExternalDataPath2) {
6266
"/tmp/external_path_c/p0=1/p1=0/bucket-0/file.orc",
6367
}));
6468
}
69+
70+
TEST(ExternalPathProviderTest, TestGetNextExternalDataPathConcurrently) {
71+
std::vector<std::string> external_table_paths;
72+
external_table_paths.emplace_back("/tmp/external_path_a/");
73+
external_table_paths.emplace_back("/tmp/external_path_b/");
74+
external_table_paths.emplace_back("/tmp/external_path_c/");
75+
std::string relative_bucket_path = "p0=1/p1=0/bucket-0";
76+
77+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<ExternalPathProvider> provider,
78+
ExternalPathProvider::Create(external_table_paths, relative_bucket_path));
79+
80+
const std::set<std::string> expected_data_paths = {
81+
"/tmp/external_path_a/p0=1/p1=0/bucket-0/file.orc",
82+
"/tmp/external_path_b/p0=1/p1=0/bucket-0/file.orc",
83+
"/tmp/external_path_c/p0=1/p1=0/bucket-0/file.orc",
84+
};
85+
std::mutex mutex;
86+
std::vector<std::string> result_data_paths;
87+
constexpr int32_t kThreadCount = 8;
88+
constexpr int32_t kPathCountPerThread = 1000;
89+
90+
std::vector<std::future<void>> futures;
91+
futures.reserve(kThreadCount);
92+
for (int32_t i = 0; i < kThreadCount; ++i) {
93+
futures.emplace_back(std::async(std::launch::async, [&]() {
94+
std::vector<std::string> local_paths;
95+
local_paths.reserve(kPathCountPerThread);
96+
for (int32_t j = 0; j < kPathCountPerThread; ++j) {
97+
local_paths.push_back(provider->GetNextExternalDataPath("file.orc"));
98+
}
99+
100+
std::lock_guard<std::mutex> lock(mutex);
101+
result_data_paths.insert(result_data_paths.end(), local_paths.begin(),
102+
local_paths.end());
103+
}));
104+
}
105+
for (auto& future : futures) {
106+
future.get();
107+
}
108+
109+
ASSERT_EQ(result_data_paths.size(), kThreadCount * kPathCountPerThread);
110+
for (const auto& data_path : result_data_paths) {
111+
ASSERT_TRUE(expected_data_paths.count(data_path)) << data_path;
112+
}
113+
}
65114
} // namespace paimon::test

‎src/paimon/common/fs/file_system_test.cpp‎

Lines changed: 80 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -159,7 +159,12 @@ class FileSystemTest : public ::testing::Test, public ::testing::WithParamInterf
159159
std::string GetTestDir() const {
160160
std::string file_system = GetParam();
161161
if (file_system == "local") {
162-
return paimon::test::GetDataDir();
162+
std::string data_dir = paimon::test::GetDataDir();
163+
if (data_dir.empty() || data_dir[0] != '/') {
164+
EXPECT_OK_AND_ASSIGN(std::string current_path, PathUtil::GetWorkingDirectory());
165+
data_dir = PathUtil::JoinPath(current_path, data_dir);
166+
}
167+
return data_dir;
163168
} else if (file_system == "jindo") {
164169
return "oss://paimon-unittest/test_data/";
165170
}
@@ -209,6 +214,28 @@ TEST_P(FileSystemTest, TestCreate) {
209214
ASSERT_NOK_WITH_MSG(fs_->Create(path, /*overwrite=*/false), "already exists");
210215
}
211216

217+
TEST_P(FileSystemTest, TestCreateRelativeFileInCurrentDirectory) {
218+
if (GetParam() != "local") {
219+
GTEST_SKIP() << "this test is only tested for the local file system";
220+
}
221+
222+
std::string path = "relative_file_" + RandomName();
223+
ASSERT_OK_AND_ASSIGN(auto out, fs_->Create(path, /*overwrite=*/true));
224+
std::string content = "content";
225+
ASSERT_OK_AND_ASSIGN(int32_t write_len, out->Write(content.data(), content.size()));
226+
ASSERT_EQ(write_len, content.size());
227+
ASSERT_OK_AND_ASSIGN(std::string uri, out->GetUri());
228+
ASSERT_FALSE(uri.empty());
229+
ASSERT_EQ(uri[0], '/');
230+
ASSERT_EQ(PathUtil::GetName(uri), path);
231+
ASSERT_OK(out->Close());
232+
233+
std::string read_content;
234+
ASSERT_OK(fs_->ReadFile(path, &read_content));
235+
ASSERT_EQ(read_content, content);
236+
ASSERT_OK(fs_->Delete(path));
237+
}
238+
212239
// --- write&read
213240
TEST_P(FileSystemTest, TestSimpleWriteAndRead) {
214241
std::string content = "abcdefghijk";
@@ -648,6 +675,27 @@ TEST_P(FileSystemTest, TestRename) {
648675
"src file is not a dir");
649676
}
650677

678+
TEST_P(FileSystemTest, TestRenameWithFileSchemeUsesNormalizedPath) {
679+
if (GetParam() != "local") {
680+
GTEST_SKIP() << "this test is only tested for the local file system";
681+
}
682+
683+
const std::string src = "file:" + test_root_ + "/scheme_src.txt";
684+
const std::string dst = "file:" + test_root_ + "/scheme_dst.txt";
685+
686+
ASSERT_OK(fs_->WriteFile(src, "content", /*overwrite=*/false));
687+
ASSERT_OK(fs_->Rename(src, dst));
688+
689+
ASSERT_OK_AND_ASSIGN(bool src_exists, fs_->Exists(src));
690+
ASSERT_FALSE(src_exists);
691+
ASSERT_OK_AND_ASSIGN(bool dst_exists, fs_->Exists(dst));
692+
ASSERT_TRUE(dst_exists);
693+
694+
std::string content;
695+
ASSERT_OK(fs_->ReadFile(dst, &content));
696+
ASSERT_EQ(content, "content");
697+
}
698+
651699
TEST_P(FileSystemTest, TestRename2) {
652700
{
653701
// test rename dir
@@ -781,6 +829,19 @@ TEST_P(FileSystemTest, TestExists) {
781829
ASSERT_TRUE(is_exist);
782830
}
783831

832+
TEST_P(FileSystemTest, TestExistsInLocalFileSystem) {
833+
if (GetParam() != "local") {
834+
GTEST_SKIP() << "this test is only tested for the local file system";
835+
}
836+
837+
ASSERT_OK_AND_ASSIGN(bool is_exist, fs_->Exists("/"));
838+
ASSERT_TRUE(is_exist);
839+
ASSERT_OK_AND_ASSIGN(is_exist, fs_->Exists(""));
840+
ASSERT_TRUE(is_exist);
841+
ASSERT_OK_AND_ASSIGN(is_exist, fs_->Exists("."));
842+
ASSERT_TRUE(is_exist);
843+
}
844+
784845
// --- delete
785846
TEST_P(FileSystemTest, TestExistingFileDeletion) {
786847
auto check = [&](bool recursive) {
@@ -985,8 +1046,16 @@ TEST_P(FileSystemTest, TestMkdir) {
9851046
ASSERT_OK(fs_->Mkdirs(test_root_ + "/tmp/local/f/1"));
9861047
ASSERT_OK(fs_->Mkdirs(test_root_ + "/tmp1"));
9871048
ASSERT_OK(fs_->Mkdirs(test_root_ + "/tmp1/f2/"));
1049+
}
1050+
1051+
TEST_P(FileSystemTest, TestMkdirInLocalFileSystem) {
1052+
if (GetParam() != "local") {
1053+
GTEST_SKIP() << "this test is only tested for the local file system";
1054+
}
1055+
9881056
ASSERT_OK(fs_->Mkdirs("/"));
989-
ASSERT_NOK_WITH_MSG(fs_->Mkdirs(""), "path is an empty string.");
1057+
ASSERT_OK(fs_->Mkdirs(""));
1058+
ASSERT_OK(fs_->Mkdirs("."));
9901059
}
9911060

9921061
TEST_P(FileSystemTest, TestMkdir2) {
@@ -1399,6 +1468,14 @@ TEST_P(FileSystemTest, TestAtomicStoreAlreadyExist) {
13991468
ASSERT_TRUE(is_exist);
14001469
}
14011470

1402-
INSTANTIATE_TEST_SUITE_P(UseLocal, FileSystemTest, ::testing::Values("local" /*, "jindo"*/));
1471+
std::vector<std::string> GetTestValuesForFileSystemTest() {
1472+
std::vector<std::string> values;
1473+
values.emplace_back("local");
1474+
// values.emplace_back("jindo");
1475+
return values;
1476+
}
1477+
1478+
INSTANTIATE_TEST_SUITE_P(FsType, FileSystemTest,
1479+
::testing::ValuesIn(GetTestValuesForFileSystemTest()));
14031480

14041481
} // namespace paimon::test

‎src/paimon/common/table/special_fields.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,8 @@ struct SpecialFields {
6565

6666
static bool IsSpecialFieldName(const std::string& field_name) {
6767
if (field_name == SequenceNumber().Name() || field_name == ValueKind().Name() ||
68-
field_name == RowId().Name() || field_name == IndexScore().Name()) {
68+
field_name == RowKind().Name() || field_name == RowId().Name() ||
69+
field_name == IndexScore().Name()) {
6970
return true;
7071
}
7172
return false;

‎src/paimon/common/table/special_fields_test.cpp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@ TEST(SpecialFieldsTest, TestIsSpecialFieldName) {
6161
ASSERT_TRUE(SpecialFields::IsSpecialFieldName("_SEQUENCE_NUMBER"));
6262
ASSERT_TRUE(SpecialFields::IsSpecialFieldName("_VALUE_KIND"));
6363
ASSERT_FALSE(SpecialFields::IsSpecialFieldName("VALUE_KIND"));
64+
ASSERT_TRUE(SpecialFields::IsSpecialFieldName("rowkind"));
6465
ASSERT_TRUE(SpecialFields::IsSpecialFieldName("_ROW_ID"));
6566
ASSERT_TRUE(SpecialFields::IsSpecialFieldName("_INDEX_SCORE"));
6667
}

‎src/paimon/common/utils/path_util.cpp‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,13 @@
1616

1717
#include "paimon/common/utils/path_util.h"
1818

19+
#include <unistd.h>
20+
21+
#include <cerrno>
1922
#include <cstddef>
2023
#include <cstdint>
24+
#include <cstdlib>
25+
#include <cstring>
2126
#include <utility>
2227

2328
#include "fmt/format.h"
@@ -142,6 +147,17 @@ void PathUtil::TrimLastDelim(std::string* dir_path) noexcept {
142147
}
143148
}
144149

150+
Result<std::string> PathUtil::GetWorkingDirectory() noexcept {
151+
char* path = getcwd(nullptr, 0);
152+
if (path != nullptr) {
153+
std::string ret(path);
154+
free(path);
155+
return ret;
156+
}
157+
return Status::IOError(
158+
fmt::format("get working directory failed, ec: {}", std::strerror(errno)));
159+
}
160+
145161
Result<std::string> PathUtil::CreateTempPath(const std::string& path) noexcept {
146162
std::string uuid;
147163
if (!UUID::Generate(&uuid)) {

‎src/paimon/common/utils/path_util.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ class PAIMON_EXPORT PathUtil {
4444
static std::string GetParentDirPath(const std::string& path) noexcept;
4545
static std::string GetName(const std::string& path) noexcept;
4646
static void TrimLastDelim(std::string* dir_path) noexcept;
47+
static Result<std::string> GetWorkingDirectory() noexcept;
4748
static Result<std::string> CreateTempPath(const std::string& path) noexcept;
4849
static Result<Path> ToPath(const std::string& path) noexcept;
4950
static Result<std::string> NormalizePath(const std::string& path) noexcept;

‎src/paimon/common/utils/path_util_test.cpp‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,12 @@ TEST(PathUtilsTest, TestTrimLastDelim) {
9191
}
9292
}
9393

94+
TEST(PathUtilsTest, TestGetWorkingDirectory) {
95+
ASSERT_OK_AND_ASSIGN(std::string current_path, PathUtil::GetWorkingDirectory());
96+
ASSERT_FALSE(current_path.empty());
97+
ASSERT_EQ(current_path[0], '/');
98+
}
99+
94100
TEST(PathUtilsTest, TestToPath) {
95101
{
96102
std::string test_path = "";
@@ -128,6 +134,22 @@ TEST(PathUtilsTest, TestToPath) {
128134
ASSERT_EQ(path.path, "/tmp/index");
129135
ASSERT_EQ(path.ToString(), "/tmp/index");
130136
}
137+
{
138+
std::string test_path = ".";
139+
ASSERT_OK_AND_ASSIGN(Path path, PathUtil::ToPath(test_path));
140+
ASSERT_EQ(path.scheme, "");
141+
ASSERT_EQ(path.authority, "");
142+
ASSERT_EQ(path.path, ".");
143+
ASSERT_EQ(path.ToString(), ".");
144+
}
145+
{
146+
std::string test_path = "relative/path";
147+
ASSERT_OK_AND_ASSIGN(Path path, PathUtil::ToPath(test_path));
148+
ASSERT_EQ(path.scheme, "");
149+
ASSERT_EQ(path.authority, "");
150+
ASSERT_EQ(path.path, "relative/path");
151+
ASSERT_EQ(path.ToString(), "relative/path");
152+
}
131153
}
132154

133155
TEST(PathUtilsTest, TestGetName) {

0 commit comments

Comments
 (0)