thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-12-05 05:52:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move reader selection logic back to StorageFile.

Browse Source
This commit is contained in:
pufit 2022-12-11 16:15:41 -05:00
parent e38a93c45a
commit 1d6e77a29a
3 changed files with 86 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
src/Disks/IO/createReadBufferFromFileBase.cpp
View File

@ -2,7 +2,6 @@
#include <IO/ReadBufferFromEmptyFile.h> #include <IO/ReadBufferFromEmptyFile.h>
#include <IO/ReadBufferFromFile.h> #include <IO/ReadBufferFromFile.h>
#include <IO/MMapReadBufferFromFileWithCache.h> #include <IO/MMapReadBufferFromFileWithCache.h>
#include <IO/MMapReadBufferFromFile.h>
#include <IO/AsynchronousReadBufferFromFile.h> #include <IO/AsynchronousReadBufferFromFile.h>
#include <Disks/IO/ThreadPoolReader.h> #include <Disks/IO/ThreadPoolReader.h>
#include <IO/SynchronousReader.h> #include <IO/SynchronousReader.h>
@ -24,37 +23,22 @@ namespace DB
namespace ErrorCodes namespace ErrorCodes
{ {
extern const int LOGICAL_ERROR; extern const int LOGICAL_ERROR;
extern const int CANNOT_STAT;
} }
std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileOrFileDescriptorBase( std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(
const std::string & filename, const std::string & filename,
const ReadSettings & settings, const ReadSettings & settings,
std::optional<size_t> read_hint, std::optional<size_t> read_hint,
std::optional<size_t> file_size, std::optional<size_t> file_size,
int flags, int flags,
char * existing_memory, char * existing_memory,
size_t alignment, size_t alignment)
bool read_from_fd,
int fd)
{ {
if (file_size.has_value() && !*file_size) if (file_size.has_value() && !*file_size)
return std::make_unique<ReadBufferFromEmptyFile>(); return std::make_unique<ReadBufferFromEmptyFile>();
struct stat file_stat{}; size_t estimated_size = 0;
if (read_from_fd)
{
if (0 != fstat(fd, &file_stat))
throwFromErrno("Cannot stat file descriptor", ErrorCodes::CANNOT_STAT);
}
else
{
if (0 != stat(filename.c_str(), &file_stat))
throwFromErrno("Cannot stat file " + filename, ErrorCodes::CANNOT_STAT);
}
size_t estimated_size = file_stat.st_size;
if (read_hint.has_value()) if (read_hint.has_value())
estimated_size = *read_hint; estimated_size = *read_hint;
else if (file_size.has_value()) else if (file_size.has_value())
@ -63,24 +47,23 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileOrFileDescriptor
if (!existing_memory if (!existing_memory
&& settings.local_fs_method == LocalFSReadMethod::mmap && settings.local_fs_method == LocalFSReadMethod::mmap
&& settings.mmap_threshold && settings.mmap_threshold
&& estimated_size >= settings.mmap_threshold && settings.mmap_cache
&& S_ISREG(file_stat.st_mode)) && estimated_size >= settings.mmap_threshold)
{ {
try try
{ {
std::unique_ptr<ReadBufferFromFileBase> res; std::unique_ptr<MMapReadBufferFromFileWithCache> res;
if (file_size)
if (settings.mmap_cache) res = std::make_unique<MMapReadBufferFromFileWithCache>(*settings.mmap_cache, filename, 0, *file_size);
res = std::make_unique<MMapReadBufferFromFileWithCache>(*settings.mmap_cache, filename, 0, estimated_size);
else else
res = std::make_unique<MMapReadBufferFromFile>(filename, 0, estimated_size); res = std::make_unique<MMapReadBufferFromFileWithCache>(*settings.mmap_cache, filename, 0, *file_size);
ProfileEvents::increment(ProfileEvents::CreatedReadBufferMMap); ProfileEvents::increment(ProfileEvents::CreatedReadBufferMMap);
return res; return res;
} }
catch (const ErrnoException &) catch (const ErrnoException &)
{ {
/// Fallback if mmap is not supported. /// Fallback if mmap is not supported (example: pipe).
ProfileEvents::increment(ProfileEvents::CreatedReadBufferMMapFailed); ProfileEvents::increment(ProfileEvents::CreatedReadBufferMMapFailed);
} }
} }
@ -89,21 +72,13 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileOrFileDescriptor
{ {
std::unique_ptr<ReadBufferFromFileBase> res; std::unique_ptr<ReadBufferFromFileBase> res;
/// Pread works only with regular files, so we explicitly fallback to read in other cases. if (settings.local_fs_method == LocalFSReadMethod::read)
if (settings.local_fs_method == LocalFSReadMethod::read || !S_ISREG(file_stat.st_mode))
{ {
if (read_from_fd) res = std::make_unique<ReadBufferFromFile>(filename, buffer_size, actual_flags, existing_memory, alignment, file_size);
res = std::make_unique<ReadBufferFromFileDescriptor>(fd, buffer_size, existing_memory, alignment, file_size);
else
res = std::make_unique<ReadBufferFromFile>(filename, buffer_size, actual_flags, existing_memory, alignment, file_size);
} }
else if (settings.local_fs_method == LocalFSReadMethod::pread || settings.local_fs_method == LocalFSReadMethod::mmap) else if (settings.local_fs_method == LocalFSReadMethod::pread || settings.local_fs_method == LocalFSReadMethod::mmap)
{ {
if (read_from_fd) res = std::make_unique<ReadBufferFromFilePReadWithDescriptorsCache>(filename, buffer_size, actual_flags, existing_memory, alignment, file_size);
res = std::make_unique<ReadBufferFromFileDescriptorPRead>(fd, buffer_size, existing_memory, alignment, file_size);
else
res = std::make_unique<ReadBufferFromFilePReadWithDescriptorsCache>(
filename, buffer_size, actual_flags, existing_memory, alignment, file_size);
} }
else if (settings.local_fs_method == LocalFSReadMethod::pread_fake_async) else if (settings.local_fs_method == LocalFSReadMethod::pread_fake_async)
{ {
@ -112,13 +87,8 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileOrFileDescriptor
throw Exception(ErrorCodes::LOGICAL_ERROR, "Global context not initialized"); throw Exception(ErrorCodes::LOGICAL_ERROR, "Global context not initialized");
auto & reader = context->getThreadPoolReader(Context::FilesystemReaderType::SYNCHRONOUS_LOCAL_FS_READER); auto & reader = context->getThreadPoolReader(Context::FilesystemReaderType::SYNCHRONOUS_LOCAL_FS_READER);
res = std::make_unique<AsynchronousReadBufferFromFileWithDescriptorsCache>(
if (read_from_fd) reader, settings.priority, filename, buffer_size, actual_flags, existing_memory, alignment, file_size);
res = std::make_unique<AsynchronousReadBufferFromFileDescriptor>(
reader, settings.priority, fd, buffer_size, existing_memory, alignment, file_size);
else
res = std::make_unique<AsynchronousReadBufferFromFileWithDescriptorsCache>(
reader, settings.priority, filename, buffer_size, actual_flags, existing_memory, alignment, file_size);
} }
else if (settings.local_fs_method == LocalFSReadMethod::pread_threadpool) else if (settings.local_fs_method == LocalFSReadMethod::pread_threadpool)
{ {
@ -127,13 +97,8 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileOrFileDescriptor
throw Exception(ErrorCodes::LOGICAL_ERROR, "Global context not initialized"); throw Exception(ErrorCodes::LOGICAL_ERROR, "Global context not initialized");
auto & reader = context->getThreadPoolReader(Context::FilesystemReaderType::ASYNCHRONOUS_LOCAL_FS_READER); auto & reader = context->getThreadPoolReader(Context::FilesystemReaderType::ASYNCHRONOUS_LOCAL_FS_READER);
res = std::make_unique<AsynchronousReadBufferFromFileWithDescriptorsCache>(
if (read_from_fd) reader, settings.priority, filename, buffer_size, actual_flags, existing_memory, alignment, file_size);
res = std::make_unique<AsynchronousReadBufferFromFileDescriptor>(
reader, settings.priority, fd, buffer_size, existing_memory, alignment, file_size);
else
res = std::make_unique<AsynchronousReadBufferFromFileWithDescriptorsCache>(
reader, settings.priority, filename, buffer_size, actual_flags, existing_memory, alignment, file_size);
} }
else else
throw Exception(ErrorCodes::LOGICAL_ERROR, "Unknown read method"); throw Exception(ErrorCodes::LOGICAL_ERROR, "Unknown read method");
@ -209,26 +174,4 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileOrFileDescriptor
return create(buffer_size, flags); return create(buffer_size, flags);
} }
std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(
const std::string & filename,
const ReadSettings & settings,
std::optional<size_t> read_hint,
std::optional<size_t> file_size,
int flags_,
char * existing_memory,
size_t alignment)
{
return createReadBufferFromFileOrFileDescriptorBase(filename, settings, read_hint, file_size, flags_, existing_memory, alignment);
}
std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileDescriptorBase(
int fd,
const ReadSettings & settings,
std::optional<size_t> read_hint,
std::optional<size_t> file_size,
char * existing_memory ,
size_t alignment)
{
return createReadBufferFromFileOrFileDescriptorBase({}, settings, read_hint, file_size, -1, existing_memory, alignment, true, fd);
}
} }

18
src/Disks/IO/createReadBufferFromFileBase.h
View File

@ -14,17 +14,6 @@ namespace DB
* @param read_hint - the number of bytes to read hint * @param read_hint - the number of bytes to read hint
* @param file_size - size of file * @param file_size - size of file
*/ */
std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileOrFileDescriptorBase(
const std::string & filename,
const ReadSettings & settings,
std::optional<size_t> read_hint = {},
std::optional<size_t> file_size = {},
int flags_ = -1,
char * existing_memory = nullptr,
size_t alignment = 0,
bool read_from_fd = false,
int fd = 0);
std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase( std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(
const std::string & filename, const std::string & filename,
const ReadSettings & settings, const ReadSettings & settings,
@ -34,11 +23,4 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(
char * existing_memory = nullptr, char * existing_memory = nullptr,
size_t alignment = 0); size_t alignment = 0);
std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileDescriptorBase(
int fd,
const ReadSettings & settings,
std::optional<size_t> read_hint = {},
std::optional<size_t> file_size = {},
char * existing_memory = nullptr,
size_t alignment = 0);
} }

84
src/Storages/StorageFile.cpp
View File

@ -38,8 +38,7 @@
#include <Common/typeid_cast.h> #include <Common/typeid_cast.h>
#include <Common/parseGlobs.h> #include <Common/parseGlobs.h>
#include <Common/filesystemHelpers.h> #include <Common/filesystemHelpers.h>
#include <Common/ProfileEvents.h>
#include <Disks/IO/createReadBufferFromFileBase.h>
#include <QueryPipeline/Pipe.h> #include <QueryPipeline/Pipe.h>
#include <QueryPipeline/QueryPipelineBuilder.h> #include <QueryPipeline/QueryPipelineBuilder.h>
@ -51,6 +50,13 @@
#include <filesystem> #include <filesystem>
namespace ProfileEvents
{
extern const Event CreatedReadBufferOrdinary;
extern const Event CreatedReadBufferMMap;
extern const Event CreatedReadBufferMMapFailed;
}
namespace fs = std::filesystem; namespace fs = std::filesystem;
namespace DB namespace DB
@ -69,6 +75,7 @@ namespace ErrorCodes
extern const int FILE_DOESNT_EXIST; extern const int FILE_DOESNT_EXIST;
extern const int TIMEOUT_EXCEEDED; extern const int TIMEOUT_EXCEEDED;
extern const int INCOMPATIBLE_COLUMNS; extern const int INCOMPATIBLE_COLUMNS;
extern const int CANNOT_STAT;
extern const int LOGICAL_ERROR; extern const int LOGICAL_ERROR;
extern const int CANNOT_APPEND_TO_FILE; extern const int CANNOT_APPEND_TO_FILE;
extern const int CANNOT_EXTRACT_TABLE_STRUCTURE; extern const int CANNOT_EXTRACT_TABLE_STRUCTURE;
@ -182,6 +189,7 @@ void checkCreationIsAllowed(
std::unique_ptr<ReadBuffer> createReadBuffer( std::unique_ptr<ReadBuffer> createReadBuffer(
const String & current_path, const String & current_path,
bool use_table_fd, bool use_table_fd,
const String & storage_name,
int table_fd, int table_fd,
const String & compression_method, const String & compression_method,
ContextPtr context) ContextPtr context)
@ -189,27 +197,73 @@ std::unique_ptr<ReadBuffer> createReadBuffer(
std::unique_ptr<ReadBuffer> nested_buffer; std::unique_ptr<ReadBuffer> nested_buffer;
CompressionMethod method; CompressionMethod method;
auto read_method = context->getSettingsRef().storage_file_read_method.value; auto read_method_string = context->getSettingsRef().storage_file_read_method.value;
auto read_settings = context->getReadSettings(); LocalFSReadMethod read_method;
read_settings.mmap_threshold = 1; if (auto opt_method = magic_enum::enum_cast<LocalFSReadMethod>(read_method_string))
read_settings.mmap_cache = nullptr; /// Turn off mmap cache for Storage File read_method = *opt_method;
if (auto opt_method = magic_enum::enum_cast<LocalFSReadMethod>(read_method))
read_settings.local_fs_method = *opt_method;
else else
throwFromErrno("Unknown read method " + read_method, ErrorCodes::UNKNOWN_READ_METHOD); throwFromErrno("Unknown read method " + read_method_string, ErrorCodes::UNKNOWN_READ_METHOD);
struct stat file_stat{};
if (use_table_fd) if (use_table_fd)
{ {
nested_buffer = createReadBufferFromFileDescriptorBase(table_fd, read_settings); /// Check if file descriptor allows random reads (and reading it twice).
if (0 != fstat(table_fd, &file_stat))
throwFromErrno("Cannot stat table file descriptor, inside " + storage_name, ErrorCodes::CANNOT_STAT);
method = chooseCompressionMethod("", compression_method); method = chooseCompressionMethod("", compression_method);
} }
else else
{ {
nested_buffer = createReadBufferFromFileBase(current_path, read_settings); /// Check if file descriptor allows random reads (and reading it twice).
if (0 != stat(current_path.c_str(), &file_stat))
throwFromErrno("Cannot stat file " + current_path, ErrorCodes::CANNOT_STAT);
method = chooseCompressionMethod(current_path, compression_method); method = chooseCompressionMethod(current_path, compression_method);
} }
bool mmap_failed = false;
if (S_ISREG(file_stat.st_mode) && read_method == LocalFSReadMethod::mmap)
{
try
{
if (use_table_fd)
nested_buffer = std::make_unique<MMapReadBufferFromFileDescriptor>(table_fd, 0);
else
nested_buffer = std::make_unique<MMapReadBufferFromFile>(current_path, 0);
ProfileEvents::increment(ProfileEvents::CreatedReadBufferMMap);
}
catch (const ErrnoException &)
{
/// Fallback if mmap is not supported.
ProfileEvents::increment(ProfileEvents::CreatedReadBufferMMapFailed);
mmap_failed = true;
}
}
if (S_ISREG(file_stat.st_mode) && (read_method == LocalFSReadMethod::pread || mmap_failed))
{
if (use_table_fd)
nested_buffer = std::make_unique<ReadBufferFromFileDescriptorPRead>(table_fd);
else
nested_buffer = std::make_unique<ReadBufferFromFilePRead>(current_path, context->getSettingsRef().max_read_buffer_size);
ProfileEvents::increment(ProfileEvents::CreatedReadBufferOrdinary);
}
else
{
if (use_table_fd)
nested_buffer = std::make_unique<ReadBufferFromFileDescriptor>(table_fd);
else
nested_buffer = std::make_unique<ReadBufferFromFile>(current_path, context->getSettingsRef().max_read_buffer_size);
ProfileEvents::increment(ProfileEvents::CreatedReadBufferOrdinary);
}
/// For clickhouse-local and clickhouse-client add progress callback to display progress bar. /// For clickhouse-local and clickhouse-client add progress callback to display progress bar.
if (context->getApplicationType() == Context::ApplicationType::LOCAL if (context->getApplicationType() == Context::ApplicationType::LOCAL
|| context->getApplicationType() == Context::ApplicationType::CLIENT) || context->getApplicationType() == Context::ApplicationType::CLIENT)
@ -277,7 +331,7 @@ ColumnsDescription StorageFile::getTableStructureFromFileDescriptor(ContextPtr c
{ {
/// We will use PeekableReadBuffer to create a checkpoint, so we need a place /// We will use PeekableReadBuffer to create a checkpoint, so we need a place
/// where we can store the original read buffer. /// where we can store the original read buffer.
read_buffer_from_fd = createReadBuffer("", true, table_fd, compression_method, context); read_buffer_from_fd = createReadBuffer("", true, getName(), table_fd, compression_method, context);
auto read_buf = std::make_unique<PeekableReadBuffer>(*read_buffer_from_fd); auto read_buf = std::make_unique<PeekableReadBuffer>(*read_buffer_from_fd);
read_buf->setCheckpoint(); read_buf->setCheckpoint();
return read_buf; return read_buf;
@ -326,7 +380,7 @@ ColumnsDescription StorageFile::getTableStructureFromFile(
if (it == paths.end()) if (it == paths.end())
return nullptr; return nullptr;
return createReadBuffer(*it++, false, -1, compression_method, context); return createReadBuffer(*it++, false, "File", -1, compression_method, context);
}; };
ColumnsDescription columns; ColumnsDescription columns;
@ -543,7 +597,7 @@ public:
} }
if (!read_buf) if (!read_buf)
read_buf = createReadBuffer(current_path, storage->use_table_fd, storage->table_fd, storage->compression_method, context); read_buf = createReadBuffer(current_path, storage->use_table_fd, storage->getName(), storage->table_fd, storage->compression_method, context);
auto format auto format
= context->getInputFormat(storage->format_name, *read_buf, block_for_format, max_block_size, storage->format_settings); = context->getInputFormat(storage->format_name, *read_buf, block_for_format, max_block_size, storage->format_settings);