CapnProtoSerializer.cpp

#include "config.h"

#if USE_CAPNP

#include <Formats/CapnProtoSerializer.h>
#include <Formats/FormatSettings.h>
#include <Formats/CapnProtoSchema.h>
#include <DataTypes/DataTypeArray.h>
#include <DataTypes/DataTypeEnum.h>
#include <DataTypes/DataTypeLowCardinality.h>
#include <DataTypes/DataTypeNullable.h>
#include <DataTypes/DataTypeTuple.h>
#include <DataTypes/DataTypeMap.h>
#include <DataTypes/IDataType.h>
#include <DataTypes/Serializations/SerializationArray.h>
#include <DataTypes/Serializations/SerializationTuple.h>
#include <Columns/ColumnArray.h>
#include <Columns/ColumnNullable.h>
#include <Columns/ColumnFixedString.h>
#include <Columns/ColumnTuple.h>
#include <Columns/ColumnLowCardinality.h>
#include <Columns/ColumnsDateTime.h>
#include <Columns/ColumnMap.h>
#include <base/find_symbols.h>

#include <boost/algorithm/string.hpp>


namespace DB
{

namespace ErrorCodes
{
    extern const int THERE_IS_NO_COLUMN;
    extern const int CAPN_PROTO_BAD_CAST;
    extern const int INCORRECT_DATA;
    extern const int ILLEGAL_COLUMN;
}

namespace
{
    std::pair<String, String> splitFieldName(const String & name)
    {
        const auto * begin = name.data();
        const auto * end = name.data() + name.size();
        const auto * it = find_first_symbols<'_', '.'>(begin, end);
        String first = String(begin, it);
        String second = it == end ? "" : String(it + 1, end);
        return {first, second};
    }

    std::optional<capnp::StructSchema::Field> findFieldByName(const capnp::StructSchema & struct_schema, const String & name)
    {
        const auto & fields = struct_schema.getFields();
        for (auto field : fields)
        {
            auto field_name = String(field.getProto().getName());
            if (boost::to_lower_copy(name) == boost::to_lower_copy(field_name))
                return field;
        }
        return std::nullopt;
    }

    [[noreturn]] void throwCannotConvert(const DataTypePtr & type, const String & name, const capnp::Type & capnp_type)
    {
        throw Exception(
            ErrorCodes::CAPN_PROTO_BAD_CAST,
            "Cannot convert ClickHouse column \"{}\" with type {} to CapnProto type {}",
            name,
            type->getName(),
            getCapnProtoFullTypeName(capnp_type));
    }

    struct FieldBuilder
    {
        virtual ~FieldBuilder() = default;
    };

    struct ListBuilder : public FieldBuilder
    {
        explicit ListBuilder(capnp::DynamicValue::Builder builder, UInt32 elements_size) : impl(builder.as<capnp::DynamicList>()), nested_builders(elements_size)
        {
        }

        capnp::DynamicList::Builder impl;
        std::vector<std::unique_ptr<FieldBuilder>> nested_builders;
    };

    struct StructBuilder : public FieldBuilder
    {
        explicit StructBuilder(capnp::DynamicStruct::Builder struct_builder, size_t fields_size) : impl(std::move(struct_builder)), field_builders(fields_size)
        {
        }

        capnp::DynamicStruct::Builder impl;
        std::vector<std::unique_ptr<FieldBuilder>> field_builders;
    };

    template <typename ParentBuilder>
    std::unique_ptr<StructBuilder> initStructBuilder(ParentBuilder & parent_builder, UInt32 offset_or_index, const capnp::_::StructSize & struct_size, size_t elements, const capnp::StructSchema & schema)
    {
        capnp::DynamicStruct::Builder builder_impl;
        if constexpr (std::is_same_v<ParentBuilder, capnp::DynamicStruct::Builder>)
            builder_impl = capnp::DynamicStruct::Builder(schema, parent_builder.getBuilderImpl().getPointerField(offset_or_index).initStruct(struct_size));
        else
            builder_impl = capnp::DynamicStruct::Builder(schema, parent_builder.getBuilderImpl().getStructElement(offset_or_index));
        return std::make_unique<StructBuilder>(std::move(builder_impl), elements);
    }

    class ICapnProtoSerializer
    {
    public:
        /// Write row as struct field.
        virtual void writeRow(
            const ColumnPtr & column,
            std::unique_ptr<FieldBuilder> & builder, /// Maybe unused for simple types, needed to initialize structs and lists.
            capnp::DynamicStruct::Builder & parent_struct_builder,
            UInt32 slot_offset,
            size_t row_num) = 0;

        /// Write row as list element.
        virtual void writeRow(
            const ColumnPtr & column,
            std::unique_ptr<FieldBuilder> & builder, /// Maybe unused for simple types, needed to initialize structs and lists.
            capnp::DynamicList::Builder & parent_list_builder,
            UInt32 array_index,
            size_t row_num) = 0;

        /// Read row from struct field at slot_offset.
        virtual void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) = 0;

        /// Read row from list element at array_index.
        virtual void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) = 0;

        virtual ~ICapnProtoSerializer() = default;
    };

    template <typename CHNumericType, typename CapnProtoNumericType, bool convert_to_bool_on_read>
    class CapnProtoIntegerSerializer : public ICapnProtoSerializer
    {
    public:
        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().setDataField<CapnProtoNumericType>(slot_offset, getValue(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_list_builder, UInt32 array_index, size_t row_num) override
        {
            parent_list_builder.getBuilderImpl().setDataElement<CapnProtoNumericType>(array_index, getValue(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertValue(column, parent_struct_reader.getReaderImpl().getDataField<CapnProtoNumericType>(slot_offset));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertValue(column, parent_list_reader.getReaderImpl().getDataElement<CapnProtoNumericType>(array_index));
        }

    private:
        CapnProtoNumericType getValue(const ColumnPtr & column, size_t row_num)
        {
            return static_cast<CapnProtoNumericType>(assert_cast<const ColumnVector<CHNumericType> &>(*column).getElement(row_num));
        }

        void insertValue(IColumn & column, CapnProtoNumericType value)
        {
            if constexpr (convert_to_bool_on_read)
                assert_cast<ColumnUInt8 &>(column).insertValue(static_cast<bool>(value));
            else
                assert_cast<ColumnVector<CHNumericType> &>(column).insertValue(static_cast<CHNumericType>(value));
        }
    };

    template <typename NumericType, bool convert_to_bool_on_read = false>
    std::unique_ptr<ICapnProtoSerializer> createIntegerSerializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type)
    {
        switch (capnp_type.which())
        {
            case capnp::schema::Type::INT8:
                return std::make_unique<CapnProtoIntegerSerializer<NumericType, Int8, convert_to_bool_on_read>>();
            case capnp::schema::Type::INT16:
                return std::make_unique<CapnProtoIntegerSerializer<NumericType, Int16, convert_to_bool_on_read>>();
            case capnp::schema::Type::INT32:
                return std::make_unique<CapnProtoIntegerSerializer<NumericType, Int32, convert_to_bool_on_read>>();
            case capnp::schema::Type::INT64:
                return std::make_unique<CapnProtoIntegerSerializer<NumericType, Int64, convert_to_bool_on_read>>();
            case capnp::schema::Type::UINT8:
                return std::make_unique<CapnProtoIntegerSerializer<NumericType, UInt8, convert_to_bool_on_read>>();
            case capnp::schema::Type::UINT16:
                return std::make_unique<CapnProtoIntegerSerializer<NumericType, UInt16, convert_to_bool_on_read>>();
            case capnp::schema::Type::UINT32:
                return std::make_unique<CapnProtoIntegerSerializer<NumericType, UInt32, convert_to_bool_on_read>>();
            case capnp::schema::Type::UINT64:
                return std::make_unique<CapnProtoIntegerSerializer<NumericType, UInt64, convert_to_bool_on_read>>();
            case capnp::schema::Type::BOOL:
                return std::make_unique<CapnProtoIntegerSerializer<NumericType, bool, convert_to_bool_on_read>>();
            default:
                throwCannotConvert(data_type, column_name, capnp_type);
        }
    }

    template <typename CHFloatType, typename CapnProtoFloatType>
    class CapnProtoFloatSerializer : public ICapnProtoSerializer
    {
    public:
        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().setDataField<CapnProtoFloatType>(slot_offset, getValue(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_list_builder, UInt32 array_index, size_t row_num) override
        {
            parent_list_builder.getBuilderImpl().setDataElement<CapnProtoFloatType>(array_index, getValue(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertValue(column, parent_struct_reader.getReaderImpl().getDataField<CapnProtoFloatType>(slot_offset));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertValue(column, parent_list_reader.getReaderImpl().getDataElement<CapnProtoFloatType>(array_index));
        }

    private:
        CapnProtoFloatType getValue(const ColumnPtr & column, size_t row_num)
        {
            return static_cast<CapnProtoFloatType>(assert_cast<const ColumnVector<CHFloatType> &>(*column).getElement(row_num));
        }

        void insertValue(IColumn & column, CapnProtoFloatType value)
        {
            assert_cast<ColumnVector<CHFloatType> &>(column).insertValue(static_cast<CHFloatType>(value));
        }
    };

    template <typename FloatType>
    std::unique_ptr<ICapnProtoSerializer> createFloatSerializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type)
    {
        switch (capnp_type.which())
        {
            case capnp::schema::Type::FLOAT32:
                return std::make_unique<CapnProtoFloatSerializer<FloatType, Float32>>();
            case capnp::schema::Type::FLOAT64:
                return std::make_unique<CapnProtoFloatSerializer<FloatType, Float64>>();
            default:
                throwCannotConvert(data_type, column_name, capnp_type);
        }
    }

    template <typename EnumType>
    class CapnProtoEnumSerializer : public ICapnProtoSerializer
    {
    public:
        CapnProtoEnumSerializer(
            const DataTypePtr & data_type_,
            const String & column_name,
            const capnp::Type & capnp_type,
            const FormatSettings::CapnProtoEnumComparingMode enum_comparing_mode_) : data_type(data_type_), enum_comparing_mode(enum_comparing_mode_)
        {
            if (!capnp_type.isEnum())
                throwCannotConvert(data_type, column_name, capnp_type);

            const auto * enum_type = assert_cast<const DataTypeEnum<EnumType> *>(data_type.get());
            const auto & enum_values = dynamic_cast<const EnumValues<EnumType> &>(*enum_type);

            enum_schema = capnp_type.asEnum();
            auto enumerants = enum_schema.getEnumerants();
            if (enum_comparing_mode == FormatSettings::CapnProtoEnumComparingMode::BY_VALUES)
            {
                auto ch_enum_values = enum_values.getSetOfAllValues();
                std::unordered_set<UInt16> capn_enum_values;
                for (auto enumerant : enumerants)
                    capn_enum_values.insert(enumerant.getOrdinal());

                /// Check if ClickHouse values is a superset of CapnProto values.
                ch_enum_is_superset = true;
                /// In CapnProto Enum fields are numbered sequentially starting from zero.
                /// Check if max CapnProto value exceeds max ClickHouse value.
                constexpr auto max_value = std::is_same_v<EnumType, Int8> ? INT8_MAX : INT16_MAX;
                if (enumerants.size() > max_value)
                {
                    ch_enum_is_superset = false;
                }
                else
                {
                    for (auto capnp_value : capn_enum_values)
                    {
                        if (!ch_enum_values.contains(static_cast<EnumType>(capnp_value)))
                        {
                            ch_enum_is_superset = false;
                            break;
                        }
                    }
                }

                /// Check if CapnProto values is a superset of ClickHouse values.
                capnp_enum_is_superset = true;
                for (auto ch_value : ch_enum_values)
                {
                    /// Capnp doesn't support negative enum values.
                    if (ch_value < 0 || !capn_enum_values.contains(static_cast<UInt16>(ch_value)))
                    {
                        capnp_enum_is_superset = false;
                        break;
                    }
                }
            }
            else
            {
                bool to_lower = enum_comparing_mode == FormatSettings::CapnProtoEnumComparingMode::BY_NAMES_CASE_INSENSITIVE;

                auto all_values = enum_values.getValues();
                std::unordered_map<String, EnumType> ch_name_to_value;
                for (auto & [name, value] : all_values)
                    ch_name_to_value[to_lower ? boost::algorithm::to_lower_copy(name) : name] = value;

                std::unordered_map<String, UInt16> capnp_name_to_value;
                for (auto enumerant : enumerants)
                {
                    String capnp_name = enumerant.getProto().getName();
                    capnp_name_to_value[to_lower ? boost::algorithm::to_lower_copy(capnp_name) : capnp_name] = enumerant.getOrdinal();
                }

                /// Check if ClickHouse names is a superset of CapnProto names.
                ch_enum_is_superset = true;
                for (auto & [capnp_name, capnp_value] : capnp_name_to_value)
                {
                    auto it = ch_name_to_value.find(capnp_name);
                    if (it == ch_name_to_value.end())
                    {
                        ch_enum_is_superset = false;
                        break;
                    }
                    capnp_to_ch_values[capnp_value] = it->second;
                }

                /// Check if CapnProto names is a superset of ClickHouse names.
                capnp_enum_is_superset = true;

                for (auto & [ch_name, ch_value] : ch_name_to_value)
                {
                    auto it = capnp_name_to_value.find(ch_name);
                    if (it == capnp_name_to_value.end())
                    {
                        capnp_enum_is_superset = false;
                        break;
                    }
                    ch_to_capnp_values[ch_value] = it->second;
                }
            }
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().setDataField<UInt16>(slot_offset, getValue(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_list_builder, UInt32 array_index, size_t row_num) override
        {
            parent_list_builder.getBuilderImpl().setDataElement<UInt16>(array_index, getValue(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertValue(column, parent_struct_reader.getReaderImpl().getDataField<UInt16>(slot_offset));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertValue(column, parent_list_reader.getReaderImpl().getDataElement<UInt16>(array_index));
        }

    private:
        UInt16 getValue(const ColumnPtr & column, size_t row_num)
        {
            if (!capnp_enum_is_superset)
                throw Exception(ErrorCodes::CAPN_PROTO_BAD_CAST, "Cannot convert ClickHouse enum to CapnProto enum: CapnProto enum values/names is not a superset of ClickHouse enum values/names");

            EnumType enum_value = assert_cast<const ColumnVector<EnumType> &>(*column).getElement(row_num);
            if (enum_comparing_mode == FormatSettings::CapnProtoEnumComparingMode::BY_VALUES)
                return static_cast<UInt16>(enum_value);
            auto it = ch_to_capnp_values.find(enum_value);
            if (it == ch_to_capnp_values.end())
                throw Exception(ErrorCodes::INCORRECT_DATA, "Unexpected value {} in ClickHouse enum", enum_value);

            return it->second;
        }

        void insertValue(IColumn & column, UInt16 capnp_enum_value)
        {
            if (!ch_enum_is_superset)
                throw Exception(ErrorCodes::CAPN_PROTO_BAD_CAST, "Cannot convert CapnProto enum to ClickHouse enum: ClickHouse enum values/names is not a superset of CapnProto enum values/names");

            if (enum_comparing_mode == FormatSettings::CapnProtoEnumComparingMode::BY_VALUES)
            {
                assert_cast<ColumnVector<EnumType> &>(column).insertValue(static_cast<EnumType>(capnp_enum_value));
            }
            else
            {
                auto it = capnp_to_ch_values.find(capnp_enum_value);
                if (it == capnp_to_ch_values.end())
                    throw Exception(ErrorCodes::INCORRECT_DATA, "Unexpected value {} in CapnProto enum", capnp_enum_value);

                assert_cast<ColumnVector<EnumType> &>(column).insertValue(it->second);
            }
        }

        DataTypePtr data_type;
        capnp::EnumSchema enum_schema;
        const FormatSettings::CapnProtoEnumComparingMode enum_comparing_mode;
        bool ch_enum_is_superset;
        bool capnp_enum_is_superset;
        std::unordered_map<EnumType, UInt16> ch_to_capnp_values;
        std::unordered_map<UInt16, EnumType> capnp_to_ch_values;
    };

    class CapnProtoDateSerializer : public ICapnProtoSerializer
    {
    public:
        CapnProtoDateSerializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type)
        {
            if (!capnp_type.isUInt16())
                throwCannotConvert(data_type, column_name, capnp_type);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().setDataField<UInt16>(slot_offset, getValue(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_list_builder, UInt32 array_index, size_t row_num) override
        {
            parent_list_builder.getBuilderImpl().setDataElement<UInt16>(array_index, getValue(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertValue(column, parent_struct_reader.getReaderImpl().getDataField<UInt16>(slot_offset));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertValue(column, parent_list_reader.getReaderImpl().getDataElement<UInt16>(array_index));
        }

    private:
        UInt16 getValue(const ColumnPtr & column, size_t row_num)
        {
            return assert_cast<const ColumnDate &>(*column).getElement(row_num);
        }

        void insertValue(IColumn & column, UInt16 value)
        {
            assert_cast<ColumnDate &>(column).insertValue(value);
        }
    };

    class CapnProtoDate32Serializer : public ICapnProtoSerializer
    {
    public:
        CapnProtoDate32Serializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type)
        {
            if (!capnp_type.isInt32())
                throwCannotConvert(data_type, column_name, capnp_type);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().setDataField<Int32>(slot_offset, getValue(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_list_builder, UInt32 array_index, size_t row_num) override
        {
            parent_list_builder.getBuilderImpl().setDataElement<Int32>(array_index, getValue(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertValue(column, parent_struct_reader.getReaderImpl().getDataField<Int32>(slot_offset));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertValue(column, parent_list_reader.getReaderImpl().getDataElement<Int32>(array_index));
        }

    private:
        Int32 getValue(const ColumnPtr & column, size_t row_num)
        {
            return assert_cast<const ColumnDate32 &>(*column).getElement(row_num);
        }

        void insertValue(IColumn & column, Int32 value)
        {
            assert_cast<ColumnDate32 &>(column).insertValue(value);
        }
    };

    class CapnProtoDateTimeSerializer : public ICapnProtoSerializer
    {
    public:
        CapnProtoDateTimeSerializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type)
        {
            if (!capnp_type.isUInt32())
                throwCannotConvert(data_type, column_name, capnp_type);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().setDataField<UInt32>(slot_offset, getValue(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_list_builder, UInt32 array_index, size_t row_num) override
        {
            parent_list_builder.getBuilderImpl().setDataElement<UInt32>(array_index, getValue(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertValue(column, parent_struct_reader.getReaderImpl().getDataField<UInt32>(slot_offset));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertValue(column, parent_list_reader.getReaderImpl().getDataElement<UInt32>(array_index));
        }

    private:
        UInt32 getValue(const ColumnPtr & column, size_t row_num)
        {
            return assert_cast<const ColumnDateTime &>(*column).getElement(row_num);
        }

        void insertValue(IColumn & column, UInt32 value)
        {
            assert_cast<ColumnDateTime &>(column).insertValue(value);
        }
    };

    class CapnProtoDateTime64Serializer : public ICapnProtoSerializer
    {
    public:
        CapnProtoDateTime64Serializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type)
        {
            if (!capnp_type.isInt64())
                throwCannotConvert(data_type, column_name, capnp_type);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().setDataField<Int64>(slot_offset, getValue(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_list_builder, UInt32 array_index, size_t row_num) override
        {
            parent_list_builder.getBuilderImpl().setDataElement<Int64>(array_index, getValue(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertValue(column, parent_struct_reader.getReaderImpl().getDataField<Int64>(slot_offset));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertValue(column, parent_list_reader.getReaderImpl().getDataElement<Int64>(array_index));
        }

    private:
        Int64 getValue(const ColumnPtr & column, size_t row_num)
        {
            return assert_cast<const ColumnDateTime64 &>(*column).getElement(row_num);
        }

        void insertValue(IColumn & column, Int64 value)
        {
            assert_cast<ColumnDateTime64 &>(column).insertValue(value);
        }
    };

    template <typename DecimalType>
    class CapnProtoDecimalSerializer : public ICapnProtoSerializer
    {
    public:
        using NativeType = typename DecimalType::NativeType;

        CapnProtoDecimalSerializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type)
        {
            auto which = WhichDataType(data_type);
            if ((!capnp_type.isInt32() && which.isDecimal32()) || (!capnp_type.isInt64() && which.isDecimal64()))
                throwCannotConvert(data_type, column_name, capnp_type);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().setDataField<NativeType>(slot_offset, getValue(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_list_builder, UInt32 array_index, size_t row_num) override
        {
            parent_list_builder.getBuilderImpl().setDataElement<NativeType>(array_index, getValue(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertValue(column, parent_struct_reader.getReaderImpl().getDataField<NativeType>(slot_offset));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertValue(column, parent_list_reader.getReaderImpl().getDataElement<NativeType>(array_index));
        }

    private:
        NativeType getValue(const ColumnPtr & column, size_t row_num)
        {
            return assert_cast<const ColumnDecimal<DecimalType> &>(*column).getElement(row_num);
        }

        void insertValue(IColumn & column, NativeType value)
        {
            assert_cast<ColumnDecimal<DecimalType> &>(column).insertValue(value);
        }
    };


    class CapnProtoIPv4Serializer : public ICapnProtoSerializer
    {
    public:
        CapnProtoIPv4Serializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type)
        {
            if (!capnp_type.isUInt32())
                throwCannotConvert(data_type, column_name, capnp_type);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().setDataField<UInt32>(slot_offset, getValue(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_list_builder, UInt32 array_index, size_t row_num) override
        {
            parent_list_builder.getBuilderImpl().setDataElement<UInt32>(array_index, getValue(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertValue(column, parent_struct_reader.getReaderImpl().getDataField<UInt32>(slot_offset));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertValue(column, parent_list_reader.getReaderImpl().getDataElement<UInt32>(array_index));
        }

    private:
        UInt32 getValue(const ColumnPtr & column, size_t row_num)
        {
            return assert_cast<const ColumnIPv4 &>(*column).getElement(row_num);
        }

        void insertValue(IColumn & column, UInt32 value)
        {
            assert_cast<ColumnIPv4 &>(column).insertValue(IPv4(value));
        }
    };

    template <typename T>
    class CapnProtoFixedSizeRawDataSerializer : public ICapnProtoSerializer
    {
    private:
        static constexpr size_t expected_value_size = sizeof(T);

    public:
        CapnProtoFixedSizeRawDataSerializer(const DataTypePtr & data_type_, const String & column_name, const capnp::Type & capnp_type) : data_type(data_type_)
        {
            if (!capnp_type.isData())
                throwCannotConvert(data_type, column_name, capnp_type);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().getPointerField(slot_offset).setBlob<capnp::Data>(getData(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_struct_builder, UInt32 array_index, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().getPointerElement(array_index).setBlob<capnp::Data>(getData(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertData(column, parent_struct_reader.getReaderImpl().getPointerField(slot_offset).getBlob<capnp::Data>(nullptr, 0));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertData(column, parent_list_reader.getReaderImpl().getPointerElement(array_index).getBlob<capnp::Data>(nullptr, 0));
        }

    private:
        capnp::Data::Reader getData(const ColumnPtr & column, size_t row_num)
        {
            auto data = column->getDataAt(row_num);
            return capnp::Data::Reader(reinterpret_cast<const kj::byte *>(data.data()), data.size());
        }

        void insertData(IColumn & column, capnp::Data::Reader data)
        {
            if (data.size() != expected_value_size)
                throw Exception(ErrorCodes::INCORRECT_DATA, "Unexpected size of {} value: {}", data_type->getName(), data.size());

            column.insertData(reinterpret_cast<const char *>(data.begin()), data.size());
        }

        DataTypePtr data_type;
    };

    template <typename CapnpType>
    class CapnProtoStringSerializer : public ICapnProtoSerializer
    {
    public:
        CapnProtoStringSerializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type)
        {
            if (!capnp_type.isData() && !capnp_type.isText())
                throwCannotConvert(data_type, column_name, capnp_type);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().getPointerField(slot_offset).setBlob<CapnpType>(getData(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_struct_builder, UInt32 array_index, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().getPointerElement(array_index).setBlob<CapnpType>(getData(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertData(column, parent_struct_reader.getReaderImpl().getPointerField(slot_offset).getBlob<CapnpType>(nullptr, 0));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertData(column, parent_list_reader.getReaderImpl().getPointerElement(array_index).getBlob<CapnpType>(nullptr, 0));
        }

    private:
        using Reader = typename CapnpType::Reader;

        Reader getData(const ColumnPtr & column, size_t row_num)
        {
            auto data = column->getDataAt(row_num);
            if constexpr (std::is_same_v<CapnpType, capnp::Data>)
                return Reader(reinterpret_cast<const kj::byte *>(data.data()), data.size());
            else
                return Reader(data.data(), data.size());
        }

        void insertData(IColumn & column, Reader data)
        {
            column.insertData(reinterpret_cast<const char *>(data.begin()), data.size());
        }
    };

    template <typename CapnpType>
    class CapnProtoFixedStringSerializer : public ICapnProtoSerializer
    {
    private:

    public:
        CapnProtoFixedStringSerializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type_) : capnp_type(capnp_type_)
        {
            if (!capnp_type.isData() && !capnp_type.isText())
                throwCannotConvert(data_type, column_name, capnp_type);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().getPointerField(slot_offset).setBlob<CapnpType>(getData(column, row_num));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> &, capnp::DynamicList::Builder & parent_struct_builder, UInt32 array_index, size_t row_num) override
        {
            parent_struct_builder.getBuilderImpl().getPointerElement(array_index).setBlob<CapnpType>(getData(column, row_num));
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            insertData(column, parent_struct_reader.getReaderImpl().getPointerField(slot_offset).getBlob<CapnpType>(nullptr, 0));
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            insertData(column, parent_list_reader.getReaderImpl().getPointerElement(array_index).getBlob<CapnpType>(nullptr, 0));
        }

    private:
        using Reader = typename CapnpType::Reader;

        Reader getData(const ColumnPtr & column, size_t row_num)
        {
            auto data = column->getDataAt(row_num);
            if constexpr (std::is_same_v<CapnpType, capnp::Data>)
            {
                return Reader(reinterpret_cast<const kj::byte *>(data.data()), data.size());
            }
            else
            {
                if (data.back() == 0)
                    return Reader(data.data(), data.size());

                /// In TEXT type data should be null-terminated, but ClickHouse FixedString data could not be.
                /// To make data null-terminated we should copy it to temporary String object and use it in capnp::Text::Reader.
                /// Note that capnp::Text::Reader works only with pointer to the data and it's size, so we should
                /// guarantee that new String object life time is longer than capnp::Text::Reader life time.
                tmp_string = data;
                return Reader(tmp_string.data(), tmp_string.size());
            }
        }

        void insertData(IColumn & column, Reader data)
        {
            auto & fixed_string_column = assert_cast<ColumnFixedString &>(column);
            if (data.size() > fixed_string_column.getN())
                throw Exception(ErrorCodes::INCORRECT_DATA, "Cannot read data with size {} to FixedString with size {}", data.size(), fixed_string_column.getN());

            fixed_string_column.insertData(reinterpret_cast<const char *>(data.begin()), data.size());
        }

        String tmp_string;
        capnp::Type capnp_type;
    };

    std::unique_ptr<ICapnProtoSerializer> createSerializer(const DataTypePtr & type, const String & name, const capnp::Type & capnp_type, const FormatSettings::CapnProto & settings);

    class CapnProtoLowCardinalitySerializer : public ICapnProtoSerializer
    {
    public:
        CapnProtoLowCardinalitySerializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type, const FormatSettings::CapnProto & settings)
        {
            nested_serializer = createSerializer(assert_cast<const DataTypeLowCardinality &>(*data_type).getDictionaryType(), column_name, capnp_type, settings);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> & field_builder, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            writeRowImpl(column, field_builder, parent_struct_builder, slot_offset, row_num);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> & field_builder, capnp::DynamicList::Builder & parent_list_builder, UInt32 array_index, size_t row_num) override
        {
            writeRowImpl(column, field_builder, parent_list_builder, array_index, row_num);
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            readRowImpl(column, parent_struct_reader, slot_offset);
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            readRowImpl(column, parent_list_reader, array_index);
        }

    private:
        template <typename ParentBuilder>
        void writeRowImpl(const ColumnPtr & column, std::unique_ptr<FieldBuilder> & field_builder, ParentBuilder & parent_builder, UInt32 offset_or_index, size_t row_num)
        {
            const auto & low_cardinality_column = assert_cast<const ColumnLowCardinality &>(*column);
            size_t index = low_cardinality_column.getIndexAt(row_num);
            const auto & dict_column = low_cardinality_column.getDictionary().getNestedColumn();
            nested_serializer->writeRow(dict_column, field_builder, parent_builder, offset_or_index, index);
        }

        template <typename ParentReader>
        void readRowImpl(IColumn & column, const ParentReader & parent_reader, UInt32 offset_or_index)
        {
            auto & low_cardinality_column = assert_cast<ColumnLowCardinality &>(column);
            auto tmp_column = low_cardinality_column.getDictionary().getNestedColumn()->cloneEmpty();
            nested_serializer->readRow(*tmp_column, parent_reader, offset_or_index);
            low_cardinality_column.insertFromFullColumn(*tmp_column, 0);
        }

        std::unique_ptr<ICapnProtoSerializer> nested_serializer;
    };

    class CapnProtoNullableSerializer : public ICapnProtoSerializer
    {
    public:
        CapnProtoNullableSerializer(const DataTypePtr & data_type, const String & column_name, const capnp::Type & capnp_type, const FormatSettings::CapnProto & settings)
        {
            if (!capnp_type.isStruct())
                throw Exception(
                    ErrorCodes::CAPN_PROTO_BAD_CAST,
                    "Cannot convert column \"{}\": Nullable can be represented only as a named union of type Void and nested type, got CapnProto type {}",
                    column_name,
                    getCapnProtoFullTypeName(capnp_type));

            /// Check that struct is a named union of type VOID and one arbitrary type.
            struct_schema = capnp_type.asStruct();
            auto node = struct_schema.getProto().getStruct();
            struct_size = capnp::_::StructSize(node.getDataWordCount(), node.getPointerCount());
            discriminant_offset = node.getDiscriminantOffset();
            if (!checkIfStructIsNamedUnion(struct_schema))
                throw Exception(
                    ErrorCodes::CAPN_PROTO_BAD_CAST,
                    "Cannot convert column \"{}\": Nullable can be represented only as a named union of type Void and nested type."
                    "Given CapnProto struct is not a named union: {}",
                    column_name,
                    getCapnProtoFullTypeName(capnp_type));

            auto union_fields = struct_schema.getUnionFields();
            if (union_fields.size() != 2)
                throw Exception(
                    ErrorCodes::CAPN_PROTO_BAD_CAST,
                    "Cannot convert column \"{}\": Nullable can be represented only as a named union of type Void and nested type."
                    "Given CapnProto union have more than 2 fields: {}",
                    column_name,
                    getCapnProtoFullTypeName(capnp_type));

            auto first = union_fields[0];
            auto second = union_fields[1];
            auto nested_type = assert_cast<const DataTypeNullable *>(data_type.get())->getNestedType();
            nested_slot_offset = first.getProto().getSlot().getOffset(); /// Both fields have the same offset.
            if (first.getType().isVoid())
            {
                nested_serializer = createSerializer(nested_type, column_name, second.getType(), settings);
                null_discriminant = 0;
                nested_discriminant = 1;
            }
            else if (second.getType().isVoid())
            {
                nested_serializer = createSerializer(nested_type, column_name, first.getType(), settings);
                null_discriminant = 1;
                nested_discriminant = 0;
            }
            else
                throw Exception(
                    ErrorCodes::CAPN_PROTO_BAD_CAST,
                    "Cannot convert column \"{}\": Nullable can be represented only as a named union of type Void and nested type."
                    "Given CapnProto union doesn't have field with type Void: {}",
                    column_name,
                    getCapnProtoFullTypeName(capnp_type));
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> & field_builder, capnp::DynamicStruct::Builder & parent_struct_builder, UInt32 slot_offset, size_t row_num) override
        {
            writeRowImpl(column, field_builder, parent_struct_builder, slot_offset, row_num);
        }

        void writeRow(const ColumnPtr & column, std::unique_ptr<FieldBuilder> & field_builder, capnp::DynamicList::Builder & parent_list_builder, UInt32 array_index, size_t row_num) override
        {
            writeRowImpl(column, field_builder, parent_list_builder, array_index, row_num);
        }

        void readRow(IColumn & column, const capnp::DynamicStruct::Reader & parent_struct_reader, UInt32 slot_offset) override
        {
            auto struct_reader = capnp::DynamicStruct::Reader(struct_schema, parent_struct_reader.getReaderImpl().getPointerField(slot_offset).getStruct(nullptr));
            readRowImpl(column, struct_reader);
        }

        void readRow(IColumn & column, const capnp::DynamicList::Reader & parent_list_reader, UInt32 array_index) override
        {
            auto struct_reader = capnp::DynamicStruct::Reader(struct_schema, parent_list_reader.getReaderImpl().getStructElement(array_index));
            readRowImpl(column, struct_reader);
        }

    private:
        template <typename ParentBuilder>
        void writeRowImpl(const ColumnPtr & column, std::unique_ptr<FieldBuilder> & field_builder, ParentBuilder & parent_builder, UInt32 offset_or_index, size_t row_num)
        {
            if (!field_builder)
                field_builder = initStructBuilder(parent_builder, offset_or_index, struct_size, 1, struct_schema);

            auto & struct_builder = assert_cast<StructBuilder &>(*field_builder);

            const auto & nullable_column = assert_cast<const ColumnNullable &>(*column);
            if (nullable_column.isNullAt(row_num))
            {
                auto struct_builder_impl = struct_builder.impl.getBuilderImpl();
                struct_builder_impl.setDataField<uint16_t>(discriminant_offset, null_discriminant);
                struct_builder_impl.setDataField<capnp::Void>(nested_slot_offset, capnp::Void());
            }
            else
            {
                const auto & nested_column = nullable_column.getNestedColumnPtr();
                struct_builder.impl.getBuilderImpl().setDataField<uint16_t>(discriminant_offset, nested_discriminant);
                nested_serializer->writeRow(nested_column, struct_builder.field_builders[0], struct_builder.impl, nested_slot_offset, row_num);
            }
        }

        void readRowImpl(IColumn & column, capnp::DynamicStruct::Reader & struct_reader)
        {
            auto & nullable_column = assert_cast<ColumnNullable &>(column);