@Override
    public ClickHouseDeserializer getDeserializer(ClickHouseDataConfig config, ClickHouseColumn column) {
        final ClickHouseDeserializer deserializer;
        switch (column.getDataType()) {
            case Bool:
                deserializer = BinaryDataProcessor::readBool;
                break;
            case Date:
                deserializer = BinaryDataProcessor.DateSerDe.of(config);
                break;
            case Date32:
                deserializer = BinaryDataProcessor.Date32SerDe.of(config);
                break;
            case DateTime:
                deserializer = column.getScale() > 0 ? BinaryDataProcessor.DateTime64SerDe.of(config, column)
                        : BinaryDataProcessor.DateTime32SerDe.of(config, column);
                break;
            case DateTime32:
                deserializer = BinaryDataProcessor.DateTime32SerDe.of(config, column);
                break;
            case DateTime64:
                deserializer = BinaryDataProcessor.DateTime64SerDe.of(config, column);
                break;
            case Enum8:
                deserializer = BinaryDataProcessor::readEnum8;
                break;
            case Enum16:
                deserializer = BinaryDataProcessor::readEnum16;
                break;
            case FixedString:
                deserializer = config.isUseBinaryString() ? new BinaryDataProcessor.FixedBytesSerDe(column)
                        : new BinaryDataProcessor.FixedStringSerDe(column);
                break;
            case Int8:
                deserializer = BinaryDataProcessor::readByte;
                break;
            case UInt8:
                deserializer = config.isWidenUnsignedTypes() ? BinaryDataProcessor::readUInt8AsShort
                        : BinaryDataProcessor::readByte;
                break;
            case Int16:
                deserializer = BinaryDataProcessor::readShort;
                break;
            case UInt16:
                deserializer = config.isWidenUnsignedTypes() ? BinaryDataProcessor::readUInt16AsInt
                        : BinaryDataProcessor::readShort;
                break;
            case Int32:
                deserializer = BinaryDataProcessor::readInteger;
                break;
            case UInt32:
                deserializer = config.isWidenUnsignedTypes() ? BinaryDataProcessor::readUInt32AsLong
                        : BinaryDataProcessor::readInteger;
                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:
                deserializer = BinaryDataProcessor::readLong;
                break;
            case Int128:
                deserializer = BinaryDataProcessor::readInt128;
                break;
            case UInt128:
                deserializer = BinaryDataProcessor::readUInt128;
                break;
            case Int256:
                deserializer = BinaryDataProcessor::readInt256;
                break;
            case UInt256:
                deserializer = BinaryDataProcessor::readUInt256;
                break;
            case Decimal:
                deserializer = BinaryDataProcessor.DecimalSerDe.of(column);
                break;
            case Decimal32:
                deserializer = BinaryDataProcessor.Decimal32SerDe.of(column);
                break;
            case Decimal64:
                deserializer = BinaryDataProcessor.Decimal64SerDe.of(column);
                break;
            case Decimal128:
                deserializer = BinaryDataProcessor.Decimal128SerDe.of(column);
                break;
            case Decimal256:
                deserializer = BinaryDataProcessor.Decimal256SerDe.of(column);
                break;
            case Float32:
                deserializer = BinaryDataProcessor::readFloat;
                break;
            case Float64:
                deserializer = BinaryDataProcessor::readDouble;
                break;
            case IPv4:
                deserializer = BinaryDataProcessor::readIpv4;
                break;
            case IPv6:
                deserializer = BinaryDataProcessor::readIpv6;
                break;
            case UUID:
                deserializer = BinaryDataProcessor::readUuid;
                break;
            // Geo types
            case Point:
                deserializer = BinaryDataProcessor::readGeoPoint;
                break;
            case Ring:
                deserializer = BinaryDataProcessor::readGeoRing;
                break;
            case Polygon:
                deserializer = BinaryDataProcessor::readGeoPolygon;
                break;
            case MultiPolygon:
                deserializer = BinaryDataProcessor::readGeoMultiPolygon;
                break;
            // String
            case JSON:
            case Object:
            case String:
                deserializer = config.isUseBinaryString() ? BinaryDataProcessor::readBinaryString
                        : BinaryDataProcessor::readTextString;
                break;
            // nested
            case Array: {
                if (column.getArrayNestedLevel() == 1) {
                    ClickHouseColumn baseColumn = column.getArrayBaseColumn();
                    Class<?> javaClass = baseColumn.getObjectClassForArray(config);
                    if (!baseColumn.isNullable() && javaClass.isPrimitive()) {
                        int byteLength = baseColumn.getDataType().getByteLength();
                        if (byteLength == Byte.BYTES) { // Bool, *Int8
                            deserializer = BinaryDataProcessor::readByteArray;
                        } else if (byteLength == Short.BYTES) { // *Int16
                            deserializer = BinaryDataProcessor::readShortArray;
                        } else if (int.class == javaClass) { // Int32
                            deserializer = BinaryDataProcessor::readIntegerArray;
                        } else if (long.class == javaClass) { // UInt32, *Int64
                            deserializer = byteLength == Long.BYTES ? BinaryDataProcessor::readLongArray
                                    : BinaryDataProcessor::readIntegerArray;
                        } else if (float.class == javaClass) { // Float32
                            deserializer = BinaryDataProcessor::readFloatArray;
                        } else if (double.class == javaClass) { // Float64
                            deserializer = BinaryDataProcessor::readDoubleArray;
                        } else {
                            throw new IllegalArgumentException("Unsupported primitive type: " + javaClass);
                        }
                    } else {
                        deserializer = new BinaryDataProcessor.ArrayDeserializer(config, column, true,
                                getDeserializer(config, column.getNestedColumns().get(0)));
                    }
                } else {
                    deserializer = new BinaryDataProcessor.ArrayDeserializer(config, column, true,
                            getDeserializer(config, column.getNestedColumns().get(0)));
                }
                break;
            }
            case Map:
                deserializer = new MapDeserializer(config, column,
                        getDeserializers(config, column.getNestedColumns()));
                break;
            case Nested:
                deserializer = new NestedDeserializer(config, column,
                        getDeserializers(config, column.getNestedColumns()));
                break;
            case Tuple:
                deserializer = new TupleDeserializer(config, column,
                        getDeserializers(config, column.getNestedColumns()));
                break;
            // special
            case Nothing:
                deserializer = ClickHouseDeserializer.EMPTY_VALUE;
                break;
            case SimpleAggregateFunction:
                deserializer = getDeserializer(config, column.getNestedColumns().get(0));
                break;
            case AggregateFunction:
                if (column.getAggregateFunction() != ClickHouseAggregateFunction.groupBitmap) {
                    throw new IllegalArgumentException("Only groupMap is supported at this point");
                }
                deserializer = new BitmapSerDe(config, column)::deserialize;
                break;
            case Variant:
                deserializer = new VariantDeserializer(config, column,
                        getDeserializers(config, column.getNestedColumns()));
                break;
            default:
                throw new IllegalArgumentException("Unsupported column:" + column.toString());
        }

        return column.isNullable() ? new BinaryDataProcessor.NullableDeserializer(deserializer) : deserializer;
    }
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