Method readImpl

src/Formats/MySQLBlockInputStream.cpp:217–277

Block MySQLBlockInputStream::readImpl()
{
    auto row = connection->result.fetch();
    if (!row)
    {
        if (settings->auto_close)
           connection->entry.disconnect();

        return {};
    }

    MutableColumns columns(description.sample_block.columns());
    for (const auto i : collections::range(0, columns.size()))
        columns[i] = description.sample_block.getByPosition(i).column->cloneEmpty();

    size_t num_rows = 0;
    size_t read_bytes_size = 0;

    while (row)
    {
        for (size_t index = 0; index < position_mapping.size(); ++index)
        {
            const auto value = row[position_mapping[index]];
            const auto & sample = description.sample_block.getByPosition(index);

            bool is_type_nullable = description.types[index].second;

            if (!value.isNull())
            {
                if (is_type_nullable)
                {
                    ColumnNullable & column_nullable = assert_cast<ColumnNullable &>(*columns[index]);
                    const auto & data_type = assert_cast<const DataTypeNullable &>(*sample.type);
                    insertValue(*data_type.getNestedType(), column_nullable.getNestedColumn(), description.types[index].first, value, read_bytes_size);
                    column_nullable.getNullMapData().emplace_back(false);
                }
                else
                {
                    insertValue(*sample.type, *columns[index], description.types[index].first, value, read_bytes_size);
                }
            }
            else
            {
                insertDefaultValue(*columns[index], *sample.column);

                if (is_type_nullable)
                {
                    ColumnNullable & column_nullable = assert_cast<ColumnNullable &>(*columns[index]);
                    column_nullable.getNullMapData().back() = true;
                }
            }
        }

        ++num_rows;
        if (num_rows == settings->max_read_mysql_row_nums || (settings->max_read_mysql_bytes_size && read_bytes_size >= settings->max_read_mysql_bytes_size))
            break;

        row = connection->result.fetch();