ReadBuffer & istr_,

const Block & header_,

UInt64 server_revision_,

std::optional<FormatSettings> format_settings_,

BlockMissingValues * block_missing_values_)

: istr(istr_)

, header(header_)

, server_revision(server_revision_)

, format_settings(std::move(format_settings_))

, block_missing_values(block_missing_values_)

{

IndexForNativeFormat::Blocks::const_iterator index_block_it_,

IndexForNativeFormat::Blocks::const_iterator index_block_end_)

: istr(istr_), server_revision(server_revision_),

use_index(true), index_block_it(index_block_it_), index_block_end(index_block_end_)

{

istr_concrete = typeid_cast<CompressedReadBufferFromFile *>(&istr);

if (!istr_concrete)

throw Exception(ErrorCodes::LOGICAL_ERROR, "When need to use index for NativeReader, istr must be CompressedReadBufferFromFile.");

if (index_block_it == index_block_end)

return;

index_column_it = index_block_it->columns.begin();

/// Initialize header from the index.

for (const auto & column : index_block_it->columns)

{

auto type = DataTypeFactory::instance().get(column.type);

header.insert(ColumnWithTypeAndName{ type, column.name });

}

{

istr_concrete = nullptr;

use_index = false;

const ISerialization & serialization,

ColumnPtr & column,

ReadBuffer & istr,

const FormatSettings * format_settings,

size_t rows,

const NameAndTypePair * name_and_type,

ValueSizeMap * avg_value_size_hints_)

{

ISerialization::DeserializeBinaryBulkSettings settings;

settings.getter = [&](ISerialization::SubstreamPath) -> ReadBuffer * { return &istr; };

settings.position_independent_encoding = false;

settings.native_format = true;

settings.format_settings = format_settings;

if (name_and_type != nullptr && avg_value_size_hints_ != nullptr)

{

settings.get_avg_value_size_hint_callback = [&](const ISerialization::SubstreamPath & substream_path) -> double

{

auto stream_name = ISerialization::getFileNameForStream(*name_and_type, substream_path, {});

return (*avg_value_size_hints_)[stream_name];

};

settings.update_avg_value_size_hint_callback = [&](const ISerialization::SubstreamPath & substream_path, const IColumn & column_)

{

auto stream_name = ISerialization::getFileNameForStream(*name_and_type, substream_path, {});

IDataType::updateAvgValueSizeHint(column_, (*avg_value_size_hints_)[stream_name]);

};

}

ISerialization::DeserializeBinaryBulkStatePtr state;

serialization.deserializeBinaryBulkStatePrefix(settings, state, nullptr);

serialization.deserializeBinaryBulkWithMultipleStreams(column, 0, rows, settings, state, nullptr);

if (column->size() != rows)

throw Exception(

ErrorCodes::CANNOT_READ_ALL_DATA,

"Cannot read all data in NativeReader. Rows read: {}. Rows expected: {}",

column->size(),

rows);

{

Block res;

const DataTypeFactory & data_type_factory = DataTypeFactory::instance();

if (use_index && index_block_it == index_block_end)

return res;

if (istr.eof())

{

if (use_index)

throw Exception(ErrorCodes::CANNOT_READ_ALL_DATA, "Input doesn't contain all data for index.");

return res;

}

/// Additional information about the block.

if (server_revision > 0)

res.info.read(istr, server_revision);

/// Dimensions

size_t columns = 0;

size_t rows = 0;

if (!use_index)

{

readVarUInt(columns, istr);

readVarUInt(rows, istr);

if (columns > DEFAULT_NATIVE_BINARY_MAX_NUM_COLUMNS)

throw Exception(ErrorCodes::TOO_LARGE_ARRAY_SIZE, "Suspiciously many columns in Native format: {}", columns);

if (rows > DEFAULT_NATIVE_BINARY_MAX_NUM_ROWS)

throw Exception(ErrorCodes::TOO_LARGE_ARRAY_SIZE, "Suspiciously many rows in Native format: {}", rows);

}

else

{

columns = index_block_it->num_columns;

rows = index_block_it->num_rows;

}

if (columns == 0 && header.empty() && rows != 0)

throw Exception(ErrorCodes::INCORRECT_DATA, "Zero columns but {} rows in Native format.", rows);

for (size_t i = 0; i < columns; ++i)

{

if (use_index)

{

/// If the current position is what is required, the real seek does not occur.

istr_concrete->seek(index_column_it->location.offset_in_compressed_file, index_column_it->location.offset_in_decompressed_block);

}

ColumnWithTypeAndName column;

/// Name

readBinary(column.name, istr);

/// Type

String type_name;

if (format_settings && format_settings->native.decode_types_in_binary_format)

{

column.type = decodeDataType(istr);

type_name = column.type->getName();

}

else

{

readBinary(type_name, istr);

column.type = data_type_factory.get(type_name);

}

setVersionToAggregateFunctions(column.type, true, server_revision);

SerializationPtr serialization;

ColumnPtr read_column;

if (server_revision >= DBMS_MIN_REVISION_WITH_CUSTOM_SERIALIZATION)

{

/// NativeReader must enable all supported serializations (e.g. nullable sparse) here. Since it operates on

/// in-memory state, it should be able to handle all possible serialization variants.

auto info = column.type->createSerializationInfo(SerializationInfoSettings::enableAllSupportedSerializations());

UInt8 has_custom;

readBinary(has_custom, istr);

if (has_custom)

info->deserializeFromKindsBinary(istr);

serialization = column.type->getSerialization(*info);

auto new_column = column.type->createColumn(*serialization);

new_column->reserve(rows);

read_column = std::move(new_column);

}

else

{

serialization = column.type->getDefaultSerialization();

auto new_column = column.type->createColumn(*serialization);

new_column->reserve(rows);

read_column = std::move(new_column);

}

if (use_index)

{

/// Index allows to do more checks.

if (index_column_it->name != column.name)

throw Exception(ErrorCodes::INCORRECT_INDEX, "Index points to a column with a wrong name ({} instead of {}): corrupted index or data", index_column_it->name, column.name);

/// Note: we can't compare data types as strings, compatible data types may differ in parameters.

}

/// If no rows, nothing to read.

if (rows)

{

const auto * format = format_settings ? &*format_settings : nullptr;

NameAndTypePair name_and_type = {column.name, column.type};

readData(*serialization, read_column, istr, format, rows, &name_and_type, &avg_value_size_hints);

}

column.column = std::move(read_column);

bool use_in_result = true;

if (!header.empty())

{

if (header.has(column.name))

{

auto & header_column = header.getByName(column.name);

if (format_settings && format_settings->null_as_default)

insertNullAsDefaultIfNeeded(column, header_column, header.getPositionByName(column.name), block_missing_values);

if (!header_column.type->equals(*column.type))

{

if (format_settings && format_settings->native.allow_types_conversion)

{

try

{

column.column = castColumn(column, header_column.type);

}

catch (Exception & e)

{

e.addMessage(fmt::format(

"while converting column \"{}\" from type {} to type {}",

column.name,

column.type->getName(),

header_column.type->getName()));

throw;

}

}

else

{

/// Support insert from old clients without low cardinality type.

column.column = recursiveLowCardinalityTypeConversion(column.column, column.type, header_column.type);

}

column.type = header_column.type;

}

}

else

{

if (format_settings && !format_settings->skip_unknown_fields)

throw Exception(ErrorCodes::INCORRECT_DATA, "Unknown column with name {} found while reading data in Native format", column.name);

use_in_result = false;

}

}

if (use_in_result)

res.insert(std::move(column));

if (use_index)

++index_column_it;

}

if (use_index)

{

if (index_column_it != index_block_it->columns.end())

throw Exception(ErrorCodes::INCORRECT_INDEX, "Inconsistent index: not all columns were read");

++index_block_it;

if (index_block_it != index_block_end)

index_column_it = index_block_it->columns.begin();

}

if (rows && !header.empty())

{

/// Allow to skip columns. We will fill them with default values later.

Block tmp_res;

for (size_t column_i = 0; column_i != header.columns(); ++column_i)

{

auto & col = header.getByPosition(column_i);

if (res.has(col.name))

{

tmp_res.insert(res.getByName(col.name));

}

else

{

tmp_res.insert({col.type->createColumn()->cloneResized(rows), col.type, col.name});

if (block_missing_values)

block_missing_values->setBits(column_i, rows);

}

}

tmp_res.info = res.info;

res.swap(tmp_res);

}

if (res.rows() != rows)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Row count mismatch after deserialization, got: {}, expected: {}", res.rows(), rows);

return res;