thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-21 23:21:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix abnormal termination when hdfs goes away

Browse Source
This commit is contained in:
kssenii 2021-05-17 09:55:23 +00:00
parent 6936274398
commit a8748a55df
2 changed files with 14 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/Storages/HDFS/WriteBufferFromHDFS.cpp
View File

@ -33,12 +33,13 @@ struct WriteBufferFromHDFS::WriteBufferFromHDFSImpl
, fs(createHDFSFS(builder.get()))
{
const size_t begin_of_path = hdfs_uri.find('/', hdfs_uri.find("//") + 2);
const std::string path = hdfs_uri.substr(begin_of_path);
const String path = hdfs_uri.substr(begin_of_path);
if (path.find_first_of("*?{") != std::string::npos)
throw Exception("URI '" + hdfs_uri + "' contains globs, so the table is in readonly mode", ErrorCodes::CANNOT_OPEN_FILE);
if (!hdfsExists(fs.get(), path.c_str()))
throw Exception("File: " + path + " is already exists", ErrorCodes::BAD_ARGUMENTS);
fout = hdfsOpenFile(fs.get(), path.c_str(), O_WRONLY, 0, 0, 0); /// O_WRONLY meaning create or overwrite i.e., implies O_TRUNCAT here
if (fout == nullptr)
@ -58,9 +59,11 @@ struct WriteBufferFromHDFS::WriteBufferFromHDFSImpl
int write(const char * start, size_t size) const
{
int bytes_written = hdfsWrite(fs.get(), fout, start, size);
if (bytes_written < 0)
throw Exception("Fail to write HDFS file: " + hdfs_uri + " " + std::string(hdfsGetLastError()),
ErrorCodes::NETWORK_ERROR);
return bytes_written;
}
@ -97,7 +100,8 @@ void WriteBufferFromHDFS::sync()
impl->sync();
}
WriteBufferFromHDFS::~WriteBufferFromHDFS()
void WriteBufferFromHDFS::finalize()
{
try
{
@ -109,5 +113,11 @@ WriteBufferFromHDFS::~WriteBufferFromHDFS()
}
}
WriteBufferFromHDFS::~WriteBufferFromHDFS()
{
finalize();
}
}
#endif

2
src/Storages/HDFS/WriteBufferFromHDFS.h
View File

@ -26,6 +26,8 @@ public:
~WriteBufferFromHDFS() override;
void sync() override;
void finalize() override;
};
}
#endif