refine log for CodecUtils (#8670)

dbms/src/DataStreams/NativeBlockInputStream.cpp

@@ -179,9 +179,15 @@ Block NativeBlockInputStream::readImpl()
     }
 
     if (header)
+<<<<<<< HEAD
         checkColumnSize(header.columns(), columns);
     else if (!output_names.empty())
         checkColumnSize(output_names.size(), columns);
+=======
+        CodecUtils::checkColumnSize("NativeBlockInputStream", header.columns(), columns);
+    else if (!output_names.empty())
+        CodecUtils::checkColumnSize("NativeBlockInputStream", output_names.size(), columns);
+>>>>>>> 1aa8fcda4a (refine log for `CodecUtils` (#8670))
 
     for (size_t i = 0; i < columns; ++i)
     {
@@ -208,7 +214,11 @@ Block NativeBlockInputStream::readImpl()
         readBinary(type_name, istr);
         if (header)
         {
+<<<<<<< HEAD
             checkDataTypeName(i, header_datatypes[i].name, type_name);
+=======
+            CodecUtils::checkDataTypeName("NativeBlockInputStream", i, header_datatypes[i].name, type_name);
+>>>>>>> 1aa8fcda4a (refine log for `CodecUtils` (#8670))
             column.type = header_datatypes[i].type;
         }
         else

dbms/src/Flash/Coprocessor/CHBlockChunkCodec.cpp

@@ -120,6 +120,81 @@ std::unique_ptr<ChunkCodecStream> CHBlockChunkCodec::newCodecStream(const std::v
     return std::make_unique<CHBlockChunkCodecStream>(field_types);
 }
 
+<<<<<<< HEAD
+=======
+Block CHBlockChunkCodec::decodeImpl(ReadBuffer & istr, size_t reserve_size)
+{
+    Block res;
+    if (istr.eof())
+    {
+        return res;
+    }
+
+    /// Dimensions
+    size_t columns = 0;
+    size_t rows = 0;
+    readBlockMeta(istr, columns, rows);
+
+    for (size_t i = 0; i < columns; ++i)
+    {
+        ColumnWithTypeAndName column;
+        readColumnMeta(i, istr, column);
+
+        /// Data
+        MutableColumnPtr read_column = column.type->createColumn();
+        if (column.type->haveMaximumSizeOfValue())
+        {
+            if (reserve_size > 0)
+                read_column->reserve(std::max(rows, reserve_size));
+            else if (rows)
+                read_column->reserve(rows);
+        }
+
+        if (rows) /// If no rows, nothing to read.
+            readData(*column.type, *read_column, istr, rows);
+
+        column.column = std::move(read_column);
+        res.insert(std::move(column));
+    }
+    return res;
+}
+
+void CHBlockChunkCodec::readBlockMeta(ReadBuffer & istr, size_t & columns, size_t & rows) const
+{
+    readVarUInt(columns, istr);
+    readVarUInt(rows, istr);
+
+    if (header)
+        CodecUtils::checkColumnSize("CHBlockChunkCodec", header.columns(), columns);
+    else if (!output_names.empty())
+        CodecUtils::checkColumnSize("CHBlockChunkCodec", output_names.size(), columns);
+}
+
+void CHBlockChunkCodec::readColumnMeta(size_t i, ReadBuffer & istr, ColumnWithTypeAndName & column)
+{
+    /// Name
+    readBinary(column.name, istr);
+    if (header)
+        column.name = header.getByPosition(i).name;
+    else if (!output_names.empty())
+        column.name = output_names[i];
+
+    /// Type
+    String type_name;
+    readBinary(type_name, istr);
+    const DataTypeFactory & data_type_factory = DataTypeFactory::instance();
+    if (header)
+    {
+        CodecUtils::checkDataTypeName("CHBlockChunkCodec", i, header_datatypes[i].name, type_name);
+        column.type = header_datatypes[i].type;
+    }
+    else
+    {
+        column.type = data_type_factory.get(type_name);
+    }
+}
+
+>>>>>>> 1aa8fcda4a (refine log for `CodecUtils` (#8670))
 Block CHBlockChunkCodec::decode(const String & str, const DAGSchema & schema)
 {
     ReadBufferFromString read_buffer(str);