ClickHouse/src/Formats/FormatFactory.cpp

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

779 lines
39 KiB
C++
Raw Normal View History

#include <Formats/FormatFactory.h>
2019-10-28 23:43:22 +00:00
#include <algorithm>
#include <Core/Settings.h>
#include <Formats/FormatSettings.h>
#include <Interpreters/Context.h>
#include <Interpreters/ProcessList.h>
#include <IO/SharedThreadPools.h>
2019-02-19 18:41:18 +00:00
#include <Processors/Formats/IRowInputFormat.h>
#include <Processors/Formats/IRowOutputFormat.h>
#include <Processors/Formats/Impl/MySQLOutputFormat.h>
2020-06-11 23:00:49 +00:00
#include <Processors/Formats/Impl/ParallelFormattingOutputFormat.h>
#include <Processors/Formats/Impl/ParallelParsingInputFormat.h>
#include <Processors/Formats/Impl/ValuesBlockInputFormat.h>
2020-02-02 00:53:11 +00:00
#include <Poco/URI.h>
#include <Common/Exception.h>
#include <Common/KnownObjectNames.h>
#include <unistd.h>
#include <boost/algorithm/string/case_conv.hpp>
namespace DB
{
namespace ErrorCodes
{
extern const int UNKNOWN_FORMAT;
extern const int LOGICAL_ERROR;
extern const int FORMAT_IS_NOT_SUITABLE_FOR_INPUT;
extern const int FORMAT_IS_NOT_SUITABLE_FOR_OUTPUT;
2022-01-14 14:46:24 +00:00
extern const int BAD_ARGUMENTS;
}
const FormatFactory::Creators & FormatFactory::getCreators(const String & name) const
{
auto it = dict.find(name);
if (dict.end() != it)
return it->second;
throw Exception(ErrorCodes::UNKNOWN_FORMAT, "Unknown format {}", name);
}
2021-06-01 12:20:52 +00:00
FormatSettings getFormatSettings(ContextPtr context)
2019-02-19 18:41:18 +00:00
{
const auto & settings = context->getSettingsRef();
return getFormatSettings(context, settings);
}
template <typename Settings>
2021-06-01 12:20:52 +00:00
FormatSettings getFormatSettings(ContextPtr context, const Settings & settings)
2019-02-19 18:41:18 +00:00
{
FormatSettings format_settings;
format_settings.avro.allow_missing_fields = settings.input_format_avro_allow_missing_fields;
format_settings.avro.output_codec = settings.output_format_avro_codec;
format_settings.avro.output_sync_interval = settings.output_format_avro_sync_interval;
format_settings.avro.schema_registry_url = settings.format_avro_schema_registry_url.toString();
2021-07-09 16:18:22 +00:00
format_settings.avro.string_column_pattern = settings.output_format_avro_string_column_pattern.toString();
2021-10-13 08:19:37 +00:00
format_settings.avro.output_rows_in_file = settings.output_format_avro_rows_in_file;
2018-07-04 21:00:50 +00:00
format_settings.csv.allow_double_quotes = settings.format_csv_allow_double_quotes;
format_settings.csv.allow_single_quotes = settings.format_csv_allow_single_quotes;
format_settings.csv.crlf_end_of_line = settings.output_format_csv_crlf_end_of_line;
format_settings.csv.delimiter = settings.format_csv_delimiter;
2021-12-20 16:25:54 +00:00
format_settings.csv.tuple_delimiter = settings.format_csv_delimiter;
format_settings.csv.empty_as_default = settings.input_format_csv_empty_as_default;
format_settings.csv.enum_as_number = settings.input_format_csv_enum_as_number;
format_settings.csv.null_representation = settings.format_csv_null_representation;
format_settings.csv.arrays_as_nested_csv = settings.input_format_csv_arrays_as_nested_csv;
format_settings.csv.use_best_effort_in_schema_inference = settings.input_format_csv_use_best_effort_in_schema_inference;
format_settings.csv.skip_first_lines = settings.input_format_csv_skip_first_lines;
format_settings.csv.try_detect_header = settings.input_format_csv_detect_header;
format_settings.csv.trim_whitespaces = settings.input_format_csv_trim_whitespaces;
format_settings.csv.use_whitespace_tab_as_delimiter = settings.input_format_csv_use_whitespace_tab_as_delimiter;
2021-12-02 08:14:25 +00:00
format_settings.hive_text.fields_delimiter = settings.input_format_hive_text_fields_delimiter;
format_settings.hive_text.collection_items_delimiter = settings.input_format_hive_text_collection_items_delimiter;
format_settings.hive_text.map_keys_delimiter = settings.input_format_hive_text_map_keys_delimiter;
format_settings.custom.escaping_rule = settings.format_custom_escaping_rule;
format_settings.custom.field_delimiter = settings.format_custom_field_delimiter;
format_settings.custom.result_after_delimiter = settings.format_custom_result_after_delimiter;
format_settings.custom.result_before_delimiter = settings.format_custom_result_before_delimiter;
format_settings.custom.row_after_delimiter = settings.format_custom_row_after_delimiter;
format_settings.custom.row_before_delimiter = settings.format_custom_row_before_delimiter;
format_settings.custom.row_between_delimiter = settings.format_custom_row_between_delimiter;
format_settings.custom.try_detect_header = settings.input_format_custom_detect_header;
format_settings.date_time_input_format = settings.date_time_input_format;
format_settings.date_time_output_format = settings.date_time_output_format;
format_settings.input_format_ipv4_default_on_conversion_error = settings.input_format_ipv4_default_on_conversion_error;
format_settings.input_format_ipv6_default_on_conversion_error = settings.input_format_ipv6_default_on_conversion_error;
2021-11-19 05:22:44 +00:00
format_settings.bool_true_representation = settings.bool_true_representation;
format_settings.bool_false_representation = settings.bool_false_representation;
format_settings.enable_streaming = settings.output_format_enable_streaming;
format_settings.import_nested_json = settings.input_format_import_nested_json;
format_settings.input_allow_errors_num = settings.input_format_allow_errors_num;
format_settings.input_allow_errors_ratio = settings.input_format_allow_errors_ratio;
format_settings.json.array_of_rows = settings.output_format_json_array_of_rows;
format_settings.json.escape_forward_slashes = settings.output_format_json_escape_forward_slashes;
format_settings.json.write_named_tuples_as_objects = settings.output_format_json_named_tuples_as_objects;
format_settings.json.read_named_tuples_as_objects = settings.input_format_json_named_tuples_as_objects;
format_settings.json.defaults_for_missing_elements_in_named_tuple = settings.input_format_json_defaults_for_missing_elements_in_named_tuple;
format_settings.json.ignore_unknown_keys_in_named_tuple = settings.input_format_json_ignore_unknown_keys_in_named_tuple;
format_settings.json.quote_64bit_integers = settings.output_format_json_quote_64bit_integers;
format_settings.json.quote_64bit_floats = settings.output_format_json_quote_64bit_floats;
format_settings.json.quote_denormals = settings.output_format_json_quote_denormals;
2022-09-08 16:07:20 +00:00
format_settings.json.quote_decimals = settings.output_format_json_quote_decimals;
format_settings.json.read_bools_as_numbers = settings.input_format_json_read_bools_as_numbers;
2022-09-08 16:07:20 +00:00
format_settings.json.read_numbers_as_strings = settings.input_format_json_read_numbers_as_strings;
format_settings.json.read_objects_as_strings = settings.input_format_json_read_objects_as_strings;
2022-07-13 15:57:55 +00:00
format_settings.json.try_infer_numbers_from_strings = settings.input_format_json_try_infer_numbers_from_strings;
format_settings.json.validate_types_from_metadata = settings.input_format_json_validate_types_from_metadata;
format_settings.json.validate_utf8 = settings.output_format_json_validate_utf8;
format_settings.json_object_each_row.column_for_object_name = settings.format_json_object_each_row_column_for_object_name;
format_settings.json.allow_object_type = context->getSettingsRef().allow_experimental_object_type;
format_settings.null_as_default = settings.input_format_null_as_default;
2021-08-16 08:03:23 +00:00
format_settings.decimal_trailing_zeros = settings.output_format_decimal_trailing_zeros;
format_settings.parquet.row_group_rows = settings.output_format_parquet_row_group_size;
format_settings.parquet.row_group_bytes = settings.output_format_parquet_row_group_size_bytes;
2023-02-23 16:14:10 +00:00
format_settings.parquet.output_version = settings.output_format_parquet_version;
format_settings.parquet.import_nested = settings.input_format_parquet_import_nested;
2022-03-21 07:47:37 +00:00
format_settings.parquet.case_insensitive_column_matching = settings.input_format_parquet_case_insensitive_column_matching;
2023-03-28 20:28:28 +00:00
format_settings.parquet.preserve_order = settings.input_format_parquet_preserve_order;
2021-12-02 08:14:25 +00:00
format_settings.parquet.allow_missing_columns = settings.input_format_parquet_allow_missing_columns;
format_settings.parquet.skip_columns_with_unsupported_types_in_schema_inference = settings.input_format_parquet_skip_columns_with_unsupported_types_in_schema_inference;
format_settings.parquet.output_string_as_string = settings.output_format_parquet_string_as_string;
format_settings.parquet.output_fixed_string_as_fixed_byte_array = settings.output_format_parquet_fixed_string_as_fixed_byte_array;
format_settings.parquet.max_block_size = settings.input_format_parquet_max_block_size;
format_settings.parquet.output_compression_method = settings.output_format_parquet_compression_method;
format_settings.parquet.output_compliant_nested_types = settings.output_format_parquet_compliant_nested_types;
format_settings.pretty.charset = settings.output_format_pretty_grid_charset.toString() == "ASCII" ? FormatSettings::Pretty::Charset::ASCII : FormatSettings::Pretty::Charset::UTF8;
format_settings.pretty.color = settings.output_format_pretty_color;
format_settings.pretty.max_column_pad_width = settings.output_format_pretty_max_column_pad_width;
format_settings.pretty.max_rows = settings.output_format_pretty_max_rows;
format_settings.pretty.max_value_width = settings.output_format_pretty_max_value_width;
format_settings.pretty.output_format_pretty_row_numbers = settings.output_format_pretty_row_numbers;
format_settings.protobuf.input_flatten_google_wrappers = settings.input_format_protobuf_flatten_google_wrappers;
format_settings.protobuf.output_nullables_with_google_wrappers = settings.output_format_protobuf_nullables_with_google_wrappers;
format_settings.protobuf.skip_fields_with_unsupported_types_in_schema_inference = settings.input_format_protobuf_skip_fields_with_unsupported_types_in_schema_inference;
format_settings.regexp.escaping_rule = settings.format_regexp_escaping_rule;
format_settings.regexp.regexp = settings.format_regexp;
format_settings.regexp.skip_unmatched = settings.format_regexp_skip_unmatched;
format_settings.schema.format_schema = settings.format_schema;
format_settings.schema.format_schema_path = context->getFormatSchemaPath();
format_settings.schema.is_server = context->hasGlobalContext() && (context->getGlobalContext()->getApplicationType() == Context::ApplicationType::SERVER);
format_settings.skip_unknown_fields = settings.input_format_skip_unknown_fields;
2019-09-24 14:25:22 +00:00
format_settings.template_settings.resultset_format = settings.format_template_resultset;
format_settings.template_settings.row_between_delimiter = settings.format_template_rows_between_delimiter;
format_settings.template_settings.row_format = settings.format_template_row;
2020-02-03 07:40:12 +00:00
format_settings.tsv.crlf_end_of_line = settings.output_format_tsv_crlf_end_of_line;
2019-10-16 14:22:22 +00:00
format_settings.tsv.empty_as_default = settings.input_format_tsv_empty_as_default;
format_settings.tsv.enum_as_number = settings.input_format_tsv_enum_as_number;
format_settings.tsv.null_representation = settings.format_tsv_null_representation;
format_settings.tsv.use_best_effort_in_schema_inference = settings.input_format_tsv_use_best_effort_in_schema_inference;
format_settings.tsv.skip_first_lines = settings.input_format_tsv_skip_first_lines;
format_settings.tsv.try_detect_header = settings.input_format_tsv_detect_header;
format_settings.values.accurate_types_of_literals = settings.input_format_values_accurate_types_of_literals;
format_settings.values.deduce_templates_of_expressions = settings.input_format_values_deduce_templates_of_expressions;
format_settings.values.interpret_expressions = settings.input_format_values_interpret_expressions;
format_settings.with_names_use_header = settings.input_format_with_names_use_header;
format_settings.with_types_use_header = settings.input_format_with_types_use_header;
format_settings.write_statistics = settings.output_format_write_statistics;
format_settings.arrow.low_cardinality_as_dictionary = settings.output_format_arrow_low_cardinality_as_dictionary;
format_settings.arrow.import_nested = settings.input_format_arrow_import_nested;
2021-12-02 08:14:25 +00:00
format_settings.arrow.allow_missing_columns = settings.input_format_arrow_allow_missing_columns;
format_settings.arrow.skip_columns_with_unsupported_types_in_schema_inference = settings.input_format_arrow_skip_columns_with_unsupported_types_in_schema_inference;
format_settings.arrow.skip_columns_with_unsupported_types_in_schema_inference = settings.input_format_arrow_skip_columns_with_unsupported_types_in_schema_inference;
format_settings.arrow.case_insensitive_column_matching = settings.input_format_arrow_case_insensitive_column_matching;
format_settings.arrow.output_string_as_string = settings.output_format_arrow_string_as_string;
format_settings.arrow.output_fixed_string_as_fixed_byte_array = settings.output_format_arrow_fixed_string_as_fixed_byte_array;
format_settings.arrow.output_compression_method = settings.output_format_arrow_compression_method;
format_settings.orc.import_nested = settings.input_format_orc_import_nested;
2021-12-02 08:14:25 +00:00
format_settings.orc.allow_missing_columns = settings.input_format_orc_allow_missing_columns;
format_settings.orc.row_batch_size = settings.input_format_orc_row_batch_size;
format_settings.orc.skip_columns_with_unsupported_types_in_schema_inference = settings.input_format_orc_skip_columns_with_unsupported_types_in_schema_inference;
format_settings.orc.import_nested = settings.input_format_orc_import_nested;
2021-12-02 08:14:25 +00:00
format_settings.orc.allow_missing_columns = settings.input_format_orc_allow_missing_columns;
format_settings.orc.row_batch_size = settings.input_format_orc_row_batch_size;
format_settings.orc.skip_columns_with_unsupported_types_in_schema_inference = settings.input_format_orc_skip_columns_with_unsupported_types_in_schema_inference;
2022-03-21 07:47:37 +00:00
format_settings.orc.case_insensitive_column_matching = settings.input_format_orc_case_insensitive_column_matching;
format_settings.orc.output_string_as_string = settings.output_format_orc_string_as_string;
format_settings.orc.output_compression_method = settings.output_format_orc_compression_method;
format_settings.defaults_for_omitted_fields = settings.input_format_defaults_for_omitted_fields;
format_settings.capn_proto.enum_comparing_mode = settings.format_capn_proto_enum_comparising_mode;
format_settings.capn_proto.skip_fields_with_unsupported_types_in_schema_inference = settings.input_format_capn_proto_skip_fields_with_unsupported_types_in_schema_inference;
2021-10-31 19:53:24 +00:00
format_settings.seekable_read = settings.input_format_allow_seeks;
format_settings.msgpack.number_of_columns = settings.input_format_msgpack_number_of_columns;
2022-01-27 16:54:15 +00:00
format_settings.msgpack.output_uuid_representation = settings.output_format_msgpack_uuid_representation;
format_settings.max_rows_to_read_for_schema_inference = settings.input_format_max_rows_to_read_for_schema_inference;
format_settings.column_names_for_schema_inference = settings.column_names_for_schema_inference;
format_settings.schema_inference_hints = settings.schema_inference_hints;
format_settings.schema_inference_make_columns_nullable = settings.schema_inference_make_columns_nullable;
2022-04-26 10:42:56 +00:00
format_settings.mysql_dump.table_name = settings.input_format_mysql_dump_table_name;
format_settings.mysql_dump.map_column_names = settings.input_format_mysql_dump_map_column_names;
2022-06-27 18:31:57 +00:00
format_settings.sql_insert.max_batch_size = settings.output_format_sql_insert_max_batch_size;
format_settings.sql_insert.include_column_names = settings.output_format_sql_insert_include_column_names;
format_settings.sql_insert.table_name = settings.output_format_sql_insert_table_name;
format_settings.sql_insert.use_replace = settings.output_format_sql_insert_use_replace;
2022-06-30 16:14:30 +00:00
format_settings.sql_insert.quote_names = settings.output_format_sql_insert_quote_names;
2022-07-13 15:57:55 +00:00
format_settings.try_infer_integers = settings.input_format_try_infer_integers;
format_settings.try_infer_dates = settings.input_format_try_infer_dates;
format_settings.try_infer_datetimes = settings.input_format_try_infer_datetimes;
format_settings.bson.output_string_as_string = settings.output_format_bson_string_as_string;
format_settings.bson.skip_fields_with_unsupported_types_in_schema_inference = settings.input_format_bson_skip_fields_with_unsupported_types_in_schema_inference;
format_settings.max_binary_string_size = settings.format_binary_max_string_size;
2023-03-13 22:49:28 +00:00
format_settings.max_binary_array_size = settings.format_binary_max_array_size;
format_settings.native.allow_types_conversion = settings.input_format_native_allow_types_conversion;
format_settings.max_parser_depth = context->getSettingsRef().max_parser_depth;
format_settings.client_protocol_version = context->getClientProtocolVersion();
2020-02-02 00:53:11 +00:00
/// Validate avro_schema_registry_url with RemoteHostFilter when non-empty and in Server context
if (format_settings.schema.is_server)
2020-02-02 00:53:11 +00:00
{
const Poco::URI & avro_schema_registry_url = settings.format_avro_schema_registry_url;
if (!avro_schema_registry_url.empty())
context->getRemoteHostFilter().checkURL(avro_schema_registry_url);
2020-02-02 00:53:11 +00:00
}
2019-02-19 18:41:18 +00:00
return format_settings;
}
2021-06-01 12:20:52 +00:00
template FormatSettings getFormatSettings<FormatFactorySettings>(ContextPtr context, const FormatFactorySettings & settings);
2021-06-01 12:20:52 +00:00
template FormatSettings getFormatSettings<Settings>(ContextPtr context, const Settings & settings);
2019-02-19 18:41:18 +00:00
2020-05-18 10:00:22 +00:00
InputFormatPtr FormatFactory::getInput(
2019-07-08 13:00:54 +00:00
const String & name,
ReadBuffer & _buf,
2019-07-08 13:00:54 +00:00
const Block & sample,
2021-06-01 12:20:52 +00:00
ContextPtr context,
2019-07-08 13:00:54 +00:00
UInt64 max_block_size,
const std::optional<FormatSettings> & _format_settings,
std::optional<size_t> _max_parsing_threads,
std::optional<size_t> _max_download_threads,
bool is_remote_fs,
CompressionMethod compression) const
{
2023-03-28 20:28:28 +00:00
const auto& creators = getCreators(name);
if (!creators.input_creator && !creators.random_access_input_creator)
2023-01-29 21:43:36 +00:00
throw Exception(ErrorCodes::FORMAT_IS_NOT_SUITABLE_FOR_INPUT, "Format {} is not suitable for input", name);
2019-02-19 18:41:18 +00:00
auto format_settings = _format_settings ? *_format_settings : getFormatSettings(context);
const Settings & settings = context->getSettingsRef();
size_t max_parsing_threads = _max_parsing_threads.value_or(settings.max_threads);
size_t max_download_threads = _max_download_threads.value_or(settings.max_download_threads);
2023-03-28 20:28:28 +00:00
RowInputFormatParams row_input_format_params;
row_input_format_params.max_block_size = max_block_size;
row_input_format_params.allow_errors_num = format_settings.input_allow_errors_num;
row_input_format_params.allow_errors_ratio = format_settings.input_allow_errors_ratio;
row_input_format_params.max_execution_time = settings.max_execution_time;
row_input_format_params.timeout_overflow_mode = settings.timeout_overflow_mode;
2019-10-01 10:48:46 +00:00
if (context->hasQueryContext() && settings.log_queries)
context->getQueryContext()->addQueryFactoriesInfo(Context::QueryLogFactories::Format, name);
// Add ParallelReadBuffer and decompression if needed.
2020-11-05 18:07:14 +00:00
auto owned_buf = wrapReadBufferIfNeeded(_buf, compression, creators, format_settings, settings, is_remote_fs, max_download_threads);
auto & buf = owned_buf ? *owned_buf : _buf;
// Decide whether to use ParallelParsingInputFormat.
2023-03-28 20:28:28 +00:00
bool parallel_parsing = max_parsing_threads > 1 && settings.input_format_parallel_parsing && creators.file_segmentation_engine && !creators.random_access_input_creator;
2020-12-30 04:50:58 +00:00
if (settings.max_memory_usage && settings.min_chunk_bytes_for_parallel_parsing * max_parsing_threads * 2 > settings.max_memory_usage)
parallel_parsing = false;
if (settings.max_memory_usage_for_user && settings.min_chunk_bytes_for_parallel_parsing * max_parsing_threads * 2 > settings.max_memory_usage_for_user)
parallel_parsing = false;
2023-03-28 20:28:28 +00:00
if (parallel_parsing)
{
2023-03-28 20:28:28 +00:00
const auto & non_trivial_prefix_and_suffix_checker = creators.non_trivial_prefix_and_suffix_checker;
/// Disable parallel parsing for input formats with non-trivial readPrefix() and readSuffix().
if (non_trivial_prefix_and_suffix_checker && non_trivial_prefix_and_suffix_checker(buf))
parallel_parsing = false;
}
2023-03-28 20:28:28 +00:00
// Create the InputFormat in one of 3 ways.
InputFormatPtr format;
if (parallel_parsing)
2019-10-01 10:48:46 +00:00
{
const auto & input_getter = creators.input_creator;
2019-10-01 10:48:46 +00:00
2020-06-10 12:02:34 +00:00
/// Const reference is copied to lambda.
auto parser_creator = [input_getter, sample, row_input_format_params, format_settings]
(ReadBuffer & input) -> InputFormatPtr
2020-12-17 15:14:09 +00:00
{ return input_getter(input, sample, row_input_format_params, format_settings); };
2020-05-18 10:00:22 +00:00
2020-06-11 00:36:57 +00:00
ParallelParsingInputFormat::Params params{
buf, sample, parser_creator, creators.file_segmentation_engine, name, max_parsing_threads,
settings.min_chunk_bytes_for_parallel_parsing, max_block_size, context->getApplicationType() == Context::ApplicationType::SERVER};
2023-03-13 03:51:50 +00:00
2023-03-28 20:28:28 +00:00
format = std::make_shared<ParallelParsingInputFormat>(params);
2019-10-01 10:48:46 +00:00
}
2023-03-28 20:28:28 +00:00
else if (creators.random_access_input_creator)
2022-09-08 16:37:18 +00:00
{
2023-03-28 20:28:28 +00:00
format = creators.random_access_input_creator(
buf,
sample,
format_settings,
context->getReadSettings(),
is_remote_fs,
max_download_threads,
max_parsing_threads);
2022-09-08 16:37:18 +00:00
}
2023-03-28 20:28:28 +00:00
else
{
format = creators.input_creator(buf, sample, row_input_format_params, format_settings);
2023-03-28 20:28:28 +00:00
}
2019-02-19 18:41:18 +00:00
if (owned_buf)
format->addBuffer(std::move(owned_buf));
if (!settings.input_format_record_errors_file_path.toString().empty())
format->setErrorsLogger(std::make_shared<ParallelInputFormatErrorsLogger>(context));
/// It's a kludge. Because I cannot remove context from values format.
2023-03-28 20:28:28 +00:00
/// (Not needed in the parallel_parsing case above because VALUES format doesn't support it.)
if (auto * values = typeid_cast<ValuesBlockInputFormat *>(format.get()))
values->setContext(context);
return format;
2019-02-19 18:41:18 +00:00
}
std::unique_ptr<ReadBuffer> FormatFactory::wrapReadBufferIfNeeded(
ReadBuffer & buf,
2023-03-28 20:28:28 +00:00
CompressionMethod compression,
const Creators & creators,
const FormatSettings & format_settings,
const Settings & settings,
bool is_remote_fs,
2023-03-28 20:28:28 +00:00
size_t max_download_threads) const
{
std::unique_ptr<ReadBuffer> res;
bool parallel_read = is_remote_fs && max_download_threads > 1 && format_settings.seekable_read && isBufferWithFileSize(buf);
2023-03-28 20:28:28 +00:00
if (creators.random_access_input_creator)
parallel_read &= compression != CompressionMethod::None;
size_t file_size = 0;
2023-03-28 20:28:28 +00:00
if (parallel_read)
{
try
{
file_size = getFileSizeFromReadBuffer(buf);
parallel_read = file_size >= 2 * settings.max_download_buffer_size;
2023-03-28 20:28:28 +00:00
}
catch (const Poco::Exception & e)
{
parallel_read = false;
LOG_TRACE(
&Poco::Logger::get("FormatFactory"),
"Failed to setup ParallelReadBuffer because of an exception:\n{}.\n"
"Falling back to the single-threaded buffer",
e.displayText());
}
}
if (parallel_read)
{
LOG_TRACE(
&Poco::Logger::get("FormatFactory"),
"Using ParallelReadBuffer with {} workers with chunks of {} bytes",
max_download_threads,
settings.max_download_buffer_size);
res = wrapInParallelReadBufferIfSupported(
buf, threadPoolCallbackRunner<void>(getIOThreadPool().get(), "ParallelRead"),
max_download_threads, settings.max_download_buffer_size, file_size);
2023-03-28 20:28:28 +00:00
}
if (compression != CompressionMethod::None)
{
if (!res)
res = wrapReadBufferReference(buf);
2023-03-28 20:28:28 +00:00
res = wrapReadBufferWithCompressionMethod(std::move(res), compression, static_cast<int>(settings.zstd_window_log_max));
}
return res;
}
static void addExistingProgressToOutputFormat(OutputFormatPtr format, ContextPtr context)
{
2022-10-17 02:21:08 +00:00
auto element_id = context->getProcessListElement();
if (element_id)
{
/// While preparing the query there might have been progress (for example in subscalar subqueries) so add it here
auto current_progress = element_id->getProgressIn();
Progress read_progress{current_progress.read_rows, current_progress.read_bytes, current_progress.total_rows_to_read};
format->onProgress(read_progress);
/// Update the start of the statistics to use the start of the query, and not the creation of the format class
format->setStartTime(element_id->getQueryCPUStartTime(), true);
}
}
2020-12-30 03:07:30 +00:00
OutputFormatPtr FormatFactory::getOutputFormatParallelIfPossible(
const String & name,
WriteBuffer & buf,
const Block & sample,
2021-06-01 12:20:52 +00:00
ContextPtr context,
2020-12-30 03:07:30 +00:00
const std::optional<FormatSettings> & _format_settings) const
{
2021-10-11 16:11:50 +00:00
const auto & output_getter = getCreators(name).output_creator;
2020-12-30 03:07:30 +00:00
if (!output_getter)
2023-01-29 21:43:36 +00:00
throw Exception(ErrorCodes::FORMAT_IS_NOT_SUITABLE_FOR_OUTPUT, "Format {} is not suitable for output", name);
2020-12-30 03:07:30 +00:00
auto format_settings = _format_settings ? *_format_settings : getFormatSettings(context);
2020-12-30 03:07:30 +00:00
const Settings & settings = context->getSettingsRef();
2020-12-30 03:07:30 +00:00
2020-12-30 04:50:58 +00:00
if (settings.output_format_parallel_formatting && getCreators(name).supports_parallel_formatting
&& !settings.output_format_json_array_of_rows)
2020-12-30 03:07:30 +00:00
{
auto formatter_creator = [output_getter, sample, format_settings] (WriteBuffer & output) -> OutputFormatPtr
{
return output_getter(output, sample, format_settings);
};
2020-12-30 03:07:30 +00:00
ParallelFormattingOutputFormat::Params builder{buf, sample, formatter_creator, settings.max_threads};
if (context->hasQueryContext() && settings.log_queries)
context->getQueryContext()->addQueryFactoriesInfo(Context::QueryLogFactories::Format, name);
auto format = std::make_shared<ParallelFormattingOutputFormat>(builder);
addExistingProgressToOutputFormat(format, context);
return format;
2020-12-30 03:07:30 +00:00
}
return getOutputFormat(name, buf, sample, context, _format_settings);
2020-12-30 03:07:30 +00:00
}
OutputFormatPtr FormatFactory::getOutputFormat(
const String & name,
WriteBuffer & buf,
const Block & sample,
2021-06-01 12:20:52 +00:00
ContextPtr context,
const std::optional<FormatSettings> & _format_settings) const
2019-02-19 18:41:18 +00:00
{
2021-10-11 16:11:50 +00:00
const auto & output_getter = getCreators(name).output_creator;
2019-02-19 18:41:18 +00:00
if (!output_getter)
2023-01-29 21:43:36 +00:00
throw Exception(ErrorCodes::FORMAT_IS_NOT_SUITABLE_FOR_OUTPUT, "Format {} is not suitable for output", name);
2019-02-19 18:41:18 +00:00
if (context->hasQueryContext() && context->getSettingsRef().log_queries)
context->getQueryContext()->addQueryFactoriesInfo(Context::QueryLogFactories::Format, name);
2021-07-16 10:10:56 +00:00
auto format_settings = _format_settings ? *_format_settings : getFormatSettings(context);
2019-02-19 18:41:18 +00:00
/** TODO: Materialization is needed, because formats can use the functions `IDataType`,
* which only work with full columns.
*/
auto format = output_getter(buf, sample, format_settings);
/// Enable auto-flush for streaming mode. Currently it is needed by INSERT WATCH query.
if (format_settings.enable_streaming)
format->setAutoFlush();
/// It's a kludge. Because I cannot remove context from MySQL format.
if (auto * mysql = typeid_cast<MySQLOutputFormat *>(format.get()))
mysql->setContext(context);
addExistingProgressToOutputFormat(format, context);
return format;
2019-02-19 18:41:18 +00:00
}
String FormatFactory::getContentType(
const String & name,
ContextPtr context,
const std::optional<FormatSettings> & _format_settings) const
{
const auto & output_getter = getCreators(name).output_creator;
if (!output_getter)
2023-01-29 21:43:36 +00:00
throw Exception(ErrorCodes::FORMAT_IS_NOT_SUITABLE_FOR_OUTPUT, "Format {} is not suitable for output", name);
auto format_settings = _format_settings ? *_format_settings : getFormatSettings(context);
2021-12-06 20:35:29 +00:00
Block empty_block;
WriteBufferFromOwnString empty_buffer;
auto format = output_getter(empty_buffer, empty_block, format_settings);
return format->getContentType();
}
SchemaReaderPtr FormatFactory::getSchemaReader(
const String & name,
ReadBuffer & buf,
ContextPtr & context,
const std::optional<FormatSettings> & _format_settings) const
{
const auto & schema_reader_creator = dict.at(name).schema_reader_creator;
if (!schema_reader_creator)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Format {} doesn't support schema inference.", name);
auto format_settings = _format_settings ? *_format_settings : getFormatSettings(context);
auto schema_reader = schema_reader_creator(buf, format_settings);
if (schema_reader->needContext())
schema_reader->setContext(context);
return schema_reader;
}
ExternalSchemaReaderPtr FormatFactory::getExternalSchemaReader(
const String & name,
ContextPtr & context,
const std::optional<FormatSettings> & _format_settings) const
{
const auto & external_schema_reader_creator = dict.at(name).external_schema_reader_creator;
if (!external_schema_reader_creator)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Format {} doesn't support schema inference.", name);
auto format_settings = _format_settings ? *_format_settings : getFormatSettings(context);
return external_schema_reader_creator(format_settings);
}
2023-03-28 20:28:28 +00:00
void FormatFactory::registerInputFormat(const String & name, InputCreator input_creator)
{
2023-03-13 19:29:59 +00:00
chassert(input_creator);
auto & creators = dict[name];
2023-03-28 20:28:28 +00:00
if (creators.input_creator || creators.random_access_input_creator)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Input format {} is already registered", name);
2023-03-13 19:29:59 +00:00
creators.input_creator = std::move(input_creator);
2023-03-28 20:28:28 +00:00
registerFileExtension(name, name);
KnownFormatNames::instance().add(name);
}
void FormatFactory::registerRandomAccessInputFormat(const String & name, RandomAccessInputCreator input_creator)
{
chassert(input_creator);
auto & creators = dict[name];
if (creators.input_creator || creators.random_access_input_creator)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Input format {} is already registered", name);
creators.random_access_input_creator = std::move(input_creator);
registerFileExtension(name, name);
2022-11-14 18:28:19 +00:00
KnownFormatNames::instance().add(name);
}
void FormatFactory::registerNonTrivialPrefixAndSuffixChecker(const String & name, NonTrivialPrefixAndSuffixChecker non_trivial_prefix_and_suffix_checker)
{
auto & target = dict[name].non_trivial_prefix_and_suffix_checker;
if (target)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Non trivial prefix and suffix checker {} is already registered", name);
target = std::move(non_trivial_prefix_and_suffix_checker);
}
2022-01-14 15:16:18 +00:00
void FormatFactory::registerAppendSupportChecker(const String & name, AppendSupportChecker append_support_checker)
{
2022-01-14 15:16:18 +00:00
auto & target = dict[name].append_support_checker;
if (target)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Suffix checker {} is already registered", name);
2022-01-14 15:16:18 +00:00
target = std::move(append_support_checker);
}
2022-01-24 13:27:04 +00:00
void FormatFactory::markFormatHasNoAppendSupport(const String & name)
{
2022-01-14 16:17:06 +00:00
registerAppendSupportChecker(name, [](const FormatSettings &){ return false; });
}
2022-01-14 15:16:18 +00:00
bool FormatFactory::checkIfFormatSupportAppend(const String & name, ContextPtr context, const std::optional<FormatSettings> & format_settings_)
{
auto format_settings = format_settings_ ? *format_settings_ : getFormatSettings(context);
2022-01-14 15:16:18 +00:00
auto & append_support_checker = dict[name].append_support_checker;
/// By default we consider that format supports append
return !append_support_checker || append_support_checker(format_settings);
}
2021-10-11 16:11:50 +00:00
void FormatFactory::registerOutputFormat(const String & name, OutputCreator output_creator)
2019-02-19 18:41:18 +00:00
{
2021-10-11 16:11:50 +00:00
auto & target = dict[name].output_creator;
2019-02-19 18:41:18 +00:00
if (target)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Output format {} is already registered", name);
2019-02-19 18:41:18 +00:00
target = std::move(output_creator);
registerFileExtension(name, name);
2022-11-14 18:28:19 +00:00
KnownFormatNames::instance().add(name);
}
2022-01-07 05:16:41 +00:00
void FormatFactory::registerFileExtension(const String & extension, const String & format_name)
{
file_extension_formats[boost::to_lower_copy(extension)] = format_name;
2022-01-07 05:16:41 +00:00
}
String FormatFactory::getFormatFromFileName(String file_name, bool throw_if_not_found)
2022-01-07 05:16:41 +00:00
{
if (file_name == "stdin")
return getFormatFromFileDescriptor(STDIN_FILENO);
2022-01-07 05:16:41 +00:00
CompressionMethod compression_method = chooseCompressionMethod(file_name, "");
if (CompressionMethod::None != compression_method)
{
auto pos = file_name.find_last_of('.');
if (pos != String::npos)
file_name = file_name.substr(0, pos);
}
auto pos = file_name.find_last_of('.');
if (pos == String::npos)
2022-01-14 14:46:24 +00:00
{
if (throw_if_not_found)
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot determine the file format by it's extension");
2022-01-07 05:16:41 +00:00
return "";
2022-01-14 14:46:24 +00:00
}
2022-01-07 05:16:41 +00:00
String file_extension = file_name.substr(pos + 1, String::npos);
boost::algorithm::to_lower(file_extension);
auto it = file_extension_formats.find(file_extension);
if (it == file_extension_formats.end())
{
if (throw_if_not_found)
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot determine the file format by it's extension");
return "";
}
return it->second;
2022-01-07 05:16:41 +00:00
}
String FormatFactory::getFormatFromFileDescriptor(int fd)
{
#ifdef OS_LINUX
2022-05-27 20:51:37 +00:00
std::string proc_path = fmt::format("/proc/self/fd/{}", fd);
char file_path[PATH_MAX] = {'\0'};
2022-05-27 20:51:37 +00:00
if (readlink(proc_path.c_str(), file_path, sizeof(file_path) - 1) != -1)
return getFormatFromFileName(file_path, false);
return "";
#elif defined(OS_DARWIN)
char file_path[PATH_MAX] = {'\0'};
if (fcntl(fd, F_GETPATH, file_path) != -1)
return getFormatFromFileName(file_path, false);
return "";
#else
(void)fd;
return "";
#endif
}
2019-10-01 10:48:46 +00:00
void FormatFactory::registerFileSegmentationEngine(const String & name, FileSegmentationEngine file_segmentation_engine)
{
auto & target = dict[name].file_segmentation_engine;
if (target)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: File segmentation engine {} is already registered", name);
target = std::move(file_segmentation_engine);
2019-10-01 10:48:46 +00:00
}
void FormatFactory::registerSchemaReader(const String & name, SchemaReaderCreator schema_reader_creator)
{
auto & target = dict[name].schema_reader_creator;
if (target)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Schema reader {} is already registered", name);
target = std::move(schema_reader_creator);
}
void FormatFactory::registerExternalSchemaReader(const String & name, ExternalSchemaReaderCreator external_schema_reader_creator)
{
auto & target = dict[name].external_schema_reader_creator;
if (target)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Schema reader {} is already registered", name);
target = std::move(external_schema_reader_creator);
}
2020-10-06 14:02:01 +00:00
void FormatFactory::markOutputFormatSupportsParallelFormatting(const String & name)
{
auto & target = dict[name].supports_parallel_formatting;
if (target)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Output format {} is already marked as supporting parallel formatting", name);
2020-10-06 14:02:01 +00:00
target = true;
}
2022-05-13 18:39:19 +00:00
void FormatFactory::markFormatSupportsSubsetOfColumns(const String & name)
2021-03-30 21:25:37 +00:00
{
2022-05-20 14:57:27 +00:00
auto & target = dict[name].supports_subset_of_columns;
2021-03-30 21:25:37 +00:00
if (target)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Format {} is already marked as supporting subset of columns", name);
2021-04-01 00:08:02 +00:00
target = true;
2021-03-30 21:25:37 +00:00
}
void FormatFactory::markFormatSupportsSubcolumns(const String & name)
{
auto & target = dict[name].supports_subcolumns;
if (target)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Format {} is already marked as supporting subcolumns", name);
target = true;
}
void FormatFactory::markOutputFormatPrefersLargeBlocks(const String & name)
{
auto & target = dict[name].prefers_large_blocks;
if (target)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: Format {} is already marked as preferring large blocks", name);
target = true;
}
bool FormatFactory::checkIfFormatSupportsSubcolumns(const String & name) const
{
const auto & target = getCreators(name);
return target.supports_subcolumns;
}
2021-03-30 21:25:37 +00:00
2022-05-20 14:57:27 +00:00
bool FormatFactory::checkIfFormatSupportsSubsetOfColumns(const String & name) const
2021-03-30 21:25:37 +00:00
{
const auto & target = getCreators(name);
2022-05-20 14:57:27 +00:00
return target.supports_subset_of_columns;
2021-03-30 21:25:37 +00:00
}
void FormatFactory::registerAdditionalInfoForSchemaCacheGetter(
const String & name, AdditionalInfoForSchemaCacheGetter additional_info_for_schema_cache_getter)
{
auto & target = dict[name].additional_info_for_schema_cache_getter;
if (target)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FormatFactory: additional info for schema cache getter {} is already registered", name);
target = std::move(additional_info_for_schema_cache_getter);
}
String FormatFactory::getAdditionalInfoForSchemaCache(const String & name, ContextPtr context, const std::optional<FormatSettings> & format_settings_)
{
const auto & additional_info_getter = getCreators(name).additional_info_for_schema_cache_getter;
if (!additional_info_getter)
return "";
auto format_settings = format_settings_ ? *format_settings_ : getFormatSettings(context);
return additional_info_getter(format_settings);
}
2021-09-16 17:18:34 +00:00
bool FormatFactory::isInputFormat(const String & name) const
{
auto it = dict.find(name);
2023-03-28 20:28:28 +00:00
return it != dict.end() && (it->second.input_creator || it->second.random_access_input_creator);
2021-09-16 17:18:34 +00:00
}
bool FormatFactory::isOutputFormat(const String & name) const
{
auto it = dict.find(name);
2021-10-11 16:11:50 +00:00
return it != dict.end() && it->second.output_creator;
2021-09-16 17:18:34 +00:00
}
2022-05-20 14:57:27 +00:00
bool FormatFactory::checkIfFormatHasSchemaReader(const String & name) const
{
const auto & target = getCreators(name);
return bool(target.schema_reader_creator);
}
2022-05-20 14:57:27 +00:00
bool FormatFactory::checkIfFormatHasExternalSchemaReader(const String & name) const
{
const auto & target = getCreators(name);
return bool(target.external_schema_reader_creator);
}
2022-05-20 14:57:27 +00:00
bool FormatFactory::checkIfFormatHasAnySchemaReader(const String & name) const
{
return checkIfFormatHasSchemaReader(name) || checkIfFormatHasExternalSchemaReader(name);
}
bool FormatFactory::checkIfOutputFormatPrefersLargeBlocks(const String & name) const
{
const auto & target = getCreators(name);
return target.prefers_large_blocks;
}
bool FormatFactory::checkParallelizeOutputAfterReading(const String & name, ContextPtr context) const
{
if (name == "Parquet" && context->getSettingsRef().input_format_parquet_preserve_order)
return false;
return true;
}
2022-05-23 12:48:48 +00:00
void FormatFactory::checkFormatName(const String & name) const
{
auto it = dict.find(name);
if (it == dict.end())
throw Exception(ErrorCodes::UNKNOWN_FORMAT, "Unknown format {}", name);
2022-05-23 12:48:48 +00:00
}
FormatFactory & FormatFactory::instance()
{
static FormatFactory ret;
return ret;
}
}