fix decimal quoted csv input

dbms/src/DataTypes/DataTypesDecimal.cpp

@@ -52,10 +52,13 @@ void DataTypeDecimal<T>::serializeText(const IColumn & column, size_t row_num, W
 }
 
 template <typename T>
-void DataTypeDecimal<T>::readText(T & x, ReadBuffer & istr, UInt32 precision, UInt32 scale)
+void DataTypeDecimal<T>::readText(T & x, ReadBuffer & istr, UInt32 precision, UInt32 scale, bool csv)
 {
     UInt32 unread_scale = scale;
-    readDecimalText(istr, x, precision, unread_scale);
+    if (csv)
+        readCSVDecimalText(istr, x, precision, unread_scale);
+    else
+        readDecimalText(istr, x, precision, unread_scale);
     x *= getScaleMultiplier(unread_scale);
 }
 
@@ -67,6 +70,13 @@ void DataTypeDecimal<T>::deserializeText(IColumn & column, ReadBuffer & istr, co
     static_cast<ColumnType &>(column).getData().push_back(x);
 }
 
+template <typename T>
+void DataTypeDecimal<T>::deserializeTextCSV(IColumn & column, ReadBuffer & istr, const FormatSettings &) const
+{
+    T x;
+    readText(x, istr, true);
+    static_cast<ColumnType &>(column).getData().push_back(x);
+}
 
 template <typename T>
 T DataTypeDecimal<T>::parseFromString(const String & str) const

dbms/src/DataTypes/DataTypesDecimal.h

@@ -91,6 +91,7 @@ public:
 
     void serializeText(const IColumn & column, size_t row_num, WriteBuffer & ostr, const FormatSettings &) const override;
     void deserializeText(IColumn & column, ReadBuffer & istr, const FormatSettings &) const override;
+    void deserializeTextCSV(IColumn & column, ReadBuffer & istr, const FormatSettings &) const override;
 
     void serializeBinary(const Field & field, WriteBuffer & ostr) const override;
     void serializeBinary(const IColumn & column, size_t row_num, WriteBuffer & ostr) const override;
@@ -175,8 +176,8 @@ public:
 
     T parseFromString(const String & str) const;
 
-    void readText(T & x, ReadBuffer & istr) const { readText(x, istr, precision, scale); }
+    void readText(T & x, ReadBuffer & istr, bool csv = false) const { readText(x, istr, precision, scale, csv); }
-    static void readText(T & x, ReadBuffer & istr, UInt32 precision, UInt32 scale);
+    static void readText(T & x, ReadBuffer & istr, UInt32 precision, UInt32 scale, bool csv = false);
     static T getScaleMultiplier(UInt32 scale);
 
 private:

dbms/src/IO/readFloatText.h

@@ -668,6 +668,22 @@ inline void readDecimalText(ReadBuffer & buf, T & x, unsigned int precision, uns
     scale += exponent;
 }
 
+template <typename T>
+inline void readCSVDecimalText(ReadBuffer & buf, T & x, unsigned int precision, unsigned int & scale)
+{
+    if (buf.eof())
+        throwReadAfterEOF();
+
+    char maybe_quote = *buf.position();
+
+    if (maybe_quote == '\'' || maybe_quote == '\"')
+        ++buf.position();
+
+    readDecimalText(buf, x, precision, scale, false);
+
+    if (maybe_quote == '\'' || maybe_quote == '\"')
+        assertChar(maybe_quote, buf);
+}
 
 template <typename T> void readFloatTextPrecise(T & x, ReadBuffer & in) { readFloatTextPreciseImpl<T, void>(x, in); }
 template <typename T> bool tryReadFloatTextPrecise(T & x, ReadBuffer & in) { return readFloatTextPreciseImpl<T, bool>(x, in); }

dbms/tests/queries/0_stateless/00861_decimal_quoted_csv.reference

@@ -0,0 +1,5 @@
+1	1.00	1.00	1.00
+2	-1.00	-1.00	-1.00
+3	1.00	1.00	1.00
+4	-0.10	-0.10	-0.10
+5	0.01	0.01	0.01

dbms/tests/queries/0_stateless/00861_decimal_quoted_csv.sql

@@ -0,0 +1,17 @@
+DROP TABLE IF EXISTS test;
+CREATE TABLE test (key UInt64, d32 Decimal32(2), d64 Decimal64(2), d128 Decimal128(2)) ENGINE = Memory;
+
+INSERT INTO test FORMAT CSV "1","1","1","1"
+;
+INSERT INTO test FORMAT CSV "2","-1","-1","-1"
+;
+INSERT INTO test FORMAT CSV "3","1.0","1.0","1.0"
+;
+INSERT INTO test FORMAT CSV "4","-0.1","-0.1","-0.1"
+;
+INSERT INTO test FORMAT CSV "5","0.010","0.010","0.010"
+;
+
+SELECT * FROM test ORDER BY key;
+
+DROP TABLE test;