disabled custom serialization (#807)

src/Client/Connection.cpp

@@ -589,6 +589,8 @@ void Connection::sendData(const Block & block, const String & name, bool scalar)
             maybe_compressed_out = out;
 
         block_out = std::make_unique<NativeWriter>(*maybe_compressed_out, block.cloneEmpty(), server_revision);
+        if (compatible_with_clickhouse)
+            block_out->setCompatibleWithClickHouse();
     }
 
     if (scalar)
@@ -1107,6 +1109,8 @@ void Connection::setCompatibleWithClickHouse()
         block_logs_in->setCompatibleWithClickHouse();
     if (block_profile_events_in)
         block_profile_events_in->setCompatibleWithClickHouse();
+    if (block_out)
+        block_out->setCompatibleWithClickHouse();
 }
 /// proton: ends
 

src/Formats/NativeReader.cpp

@@ -145,7 +145,12 @@ Block NativeReader::read()
         setVersionToAggregateFunctions(column.type, true, server_revision);
 
         SerializationPtr serialization;
-        if (server_revision >= DBMS_MIN_REVISION_WITH_CUSTOM_SERIALIZATION)
+        /// proton: starts
+        /// Because our drivers does not support custom serialization, we can only check the serialization when `compatible_with_clickhouse` is `true`,
+        /// i.e. this reader is reading packets from clickhouse (for ClickHouse external tables).
+        if (compatible_with_clickhouse &&
+            /// proton: ends
+            server_revision >= DBMS_MIN_REVISION_WITH_CUSTOM_SERIALIZATION)
         {
             auto info = column.type->createSerializationInfo({});
 

src/Formats/NativeWriter.cpp

@@ -116,7 +116,12 @@ void NativeWriter::write(const Block & block)
 
         /// Serialization. Dynamic, if client supports it.
         SerializationPtr serialization;
-        if (client_revision >= DBMS_MIN_REVISION_WITH_CUSTOM_SERIALIZATION)
+        /// proton: starts
+        /// Because our drivers does not support custom serialization, we can only check the serialization when `compatible_with_clickhouse` is `true`,
+        /// i.e. this writer is writing packets to clickhouse (for ClickHouse external tables).
+        if (compatible_with_clickhouse &&
+            /// proton: ends
+            client_revision >= DBMS_MIN_REVISION_WITH_CUSTOM_SERIALIZATION)
         {
             auto info = column.type->getSerializationInfo(*column.column);
             serialization = column.type->getSerialization(*info);

src/Formats/NativeWriter.h

@@ -32,6 +32,10 @@ class NativeWriter
 
     static String getContentType() { return "application/octet-stream"; }
 
+    /// proton: starts
+    void setCompatibleWithClickHouse() { compatible_with_clickhouse = true; }
+    /// proton: ends
+
 private:
     WriteBuffer & ostr;
     Block header;
@@ -40,6 +44,10 @@ class NativeWriter
     size_t initial_size_of_file;    /// The initial size of the data file, if `append` done. Used for the index.
     /// If you need to write index, then `ostr` must be a CompressedWriteBuffer.
     CompressedWriteBuffer * ostr_concrete = nullptr;
+
+    /// proton: starts
+    bool compatible_with_clickhouse{false};
+    /// proton: ends
 };
 
 }