@Override
    public ClickHouseSerializer getSerializer(ClickHouseDataConfig config, ClickHouseColumn column) {
        final ClickHouseSerializer serializer;
        switch (column.getDataType()) {
            case Bool:
                serializer = BinaryDataProcessor::writeBool;
                break;
            case Date:
                serializer = BinaryDataProcessor.DateSerDe.of(config);
                break;
            case Date32:
                serializer = BinaryDataProcessor.Date32SerDe.of(config);
                break;
            case DateTime:
                serializer = column.getScale() > 0 ? BinaryDataProcessor.DateTime64SerDe.of(config, column)
                        : BinaryDataProcessor.DateTime32SerDe.of(config, column);
                break;
            case DateTime32:
                serializer = BinaryDataProcessor.DateTime32SerDe.of(config, column);
                break;
            case DateTime64:
                serializer = BinaryDataProcessor.DateTime64SerDe.of(config, column);
                break;
            case Enum8:
                serializer = BinaryDataProcessor::writeEnum8;
                break;
            case Enum16:
                serializer = BinaryDataProcessor::writeEnum16;
                break;
            case FixedString:
                serializer = config.isUseBinaryString() ? new BinaryDataProcessor.FixedBytesSerDe(column)
                        : new BinaryDataProcessor.FixedStringSerDe(column);
                break;
            case Int8:
            case UInt8:
                serializer = BinaryDataProcessor::writeByte;
                break;
            case Int16:
            case UInt16:
                serializer = BinaryDataProcessor::writeShort;
                break;
            case Int32:
            case UInt32:
                serializer = BinaryDataProcessor::writeInteger;
                break;
            case Int64:
            case IntervalYear:
            case IntervalQuarter:
            case IntervalMonth:
            case IntervalWeek:
            case IntervalDay:
            case IntervalHour:
            case IntervalMinute:
            case IntervalSecond:
            case IntervalMicrosecond:
            case IntervalMillisecond:
            case IntervalNanosecond:
            case UInt64:
                serializer = BinaryDataProcessor::writeLong;
                break;
            case Int128:
                serializer = BinaryDataProcessor::writeInt128;
                break;
            case UInt128:
                serializer = BinaryDataProcessor::writeUInt128;
                break;
            case Int256:
                serializer = BinaryDataProcessor::writeInt256;
                break;
            case UInt256:
                serializer = BinaryDataProcessor::writeUInt256;
                break;
            case Decimal:
                serializer = BinaryDataProcessor.DecimalSerDe.of(column);
                break;
            case Decimal32:
                serializer = BinaryDataProcessor.Decimal32SerDe.of(column);
                break;
            case Decimal64:
                serializer = new BinaryDataProcessor.Decimal64SerDe(column);
                break;
            case Decimal128:
                serializer = new BinaryDataProcessor.Decimal128SerDe(column);
                break;
            case Decimal256:
                serializer = new BinaryDataProcessor.Decimal256SerDe(column);
                break;
            case Float32:
                serializer = BinaryDataProcessor::writeFloat;
                break;
            case Float64:
                serializer = BinaryDataProcessor::writeDouble;
                break;
            case IPv4:
                serializer = BinaryDataProcessor::writeIpv4;
                break;
            case IPv6:
                serializer = BinaryDataProcessor::writeIpv6;
                break;
            case UUID:
                serializer = BinaryDataProcessor::writeUuid;
                break;
            // Geo types
            case Point:
                serializer = BinaryDataProcessor::writeGeoPoint;
                break;
            case Ring:
                serializer = BinaryDataProcessor::writeGeoRing;
                break;
            case Polygon:
                serializer = BinaryDataProcessor::writeGeoPolygon;
                break;
            case MultiPolygon:
                serializer = BinaryDataProcessor::writeGeoMultiPolygon;
                break;
            // String
            case JSON:
            case Object:
            case String:
                serializer = config.isUseBinaryString() ? BinaryDataProcessor::writeBinaryString
                        : BinaryDataProcessor::writeTextString;
                break;
            // nested
            case Array:
                serializer = new BinaryDataProcessor.ArraySerializer(config, column, true,
                        getSerializer(config, column.getNestedColumns().get(0)));
                break;
            case Map:
                serializer = new MapSerializer(config, column, getSerializers(config, column.getNestedColumns()));
                break;
            case Nested:
                serializer = new NestedSerializer(config, column, getSerializers(config, column.getNestedColumns()));
                break;
            case Tuple:
                serializer = new TupleSerializer(config, column, getSerializers(config, column.getNestedColumns()));
                break;
            // special
            case Nothing:
                serializer = ClickHouseSerializer.DO_NOTHING;
                break;
            case SimpleAggregateFunction:
                serializer = getSerializer(config, column.getNestedColumns().get(0));
                break;
            case AggregateFunction:
                if (column.getAggregateFunction() != ClickHouseAggregateFunction.groupBitmap) {
                    throw new IllegalArgumentException("Only groupMap is supported at this point");
                }
                serializer = new BitmapSerDe(config, column)::serialize;
                break;
            case Variant:
                serializer = new VariantSerializer(config, column, getSerializers(config, column.getNestedColumns()));
                break;
            default:
                throw new IllegalArgumentException("Unsupported column:" + column.toString());
        }

        return column.isNullable() ? new BinaryDataProcessor.NullableSerializer(serializer) : serializer;
    }
Default constructor.

@param config   non-null confinguration contains information like format
@param input    input stream for deserialization, can be null when
                {@code output} is available
@param output   outut stream for serialization, can be null when
                {@code input} is available
@param columns  nullable columns
@param settings nullable settings
@throws IOException when failed to read columns from input stream
clickhouse-data/src/main/java/com/clickhouse/data/format/ClickHouseRowBinaryProcessor.java