better segmenating + move read functions to avoid linker errors

dbms/src/IO/ReadHelpers.cpp

@@ -1053,4 +1053,36 @@ void skipToUnescapedNextLineOrEOF(ReadBuffer & buf)
     }
 }
 
+/// TODO (akuzm) - write comments for this and next function.
+void saveUpToPosition(ReadBuffer & in, DB::Memory<> & memory, char * current)
+{
+    assert(current >= in.position());
+    assert(current <= in.buffer().end());
+
+    const int old_bytes = memory.size();
+    const int additional_bytes = current - in.position();
+    const int new_bytes = old_bytes + additional_bytes;
+    /// There are no new bytes to add to memory.
+    /// No need to do extra stuff.
+    if (new_bytes == 0)
+        return;
+    memory.resize(new_bytes);
+    memcpy(memory.data() + old_bytes, in.position(), additional_bytes);
+    in.position() = current;
+}
+
+bool loadAtPosition(ReadBuffer & in, DB::Memory<> & memory, char * & current)
+{
+    assert(current <= in.buffer().end());
+
+    if (current < in.buffer().end())
+        return true;
+
+    saveUpToPosition(in, memory, current);
+    bool loaded_more = !in.eof();
+    assert(in.position() == in.buffer().begin());
+    current = in.position();
+    return loaded_more;
+}
+
 }

dbms/src/IO/ReadHelpers.h

@@ -925,35 +925,7 @@ return std::make_unique<TReadBuffer>(args...);
 }
 
 /// TODO (akuzm) - write comments for this and next function.
-void saveUpToPosition(ReadBuffer & in, DB::Memory<> & memory, char * current)
-{
-    assert(current >= in.position());
-    assert(current <= in.buffer().end());
-
-    const int old_bytes = memory.size();
-    const int additional_bytes = current - in.position();
-    const int new_bytes = old_bytes + additional_bytes;
-    /// There are no new bytes to add to memory.
-    /// No need to do extra stuff.
-    if (new_bytes == 0)
-        return;
-    memory.resize(new_bytes);
-    memcpy(memory.data() + old_bytes, in.position(), additional_bytes);
-    in.position() = current;
-}
-
-bool loadAtPosition(ReadBuffer & in, DB::Memory<> & memory, char * & current)
-{
-    assert(current <= in.buffer().end());
-
-    if (current < in.buffer().end())
-        return true;
-
-    saveUpToPosition(in, memory, current);
-    bool loaded_more = !in.eof();
-    assert(in.position() == in.buffer().begin());
-    current = in.position();
-    return loaded_more;
-}
+void saveUpToPosition(ReadBuffer & in, DB::Memory<> & memory, char * current);
+bool loadAtPosition(ReadBuffer & in, DB::Memory<> & memory, char * & current);
 
 }

dbms/src/Processors/Formats/Impl/CSVRowInputFormat.cpp

@@ -424,8 +424,6 @@ void registerInputFormatProcessorCSV(FormatFactory & factory)
 
 bool fileSegmentationEngineCSVImpl(ReadBuffer & in, DB::Memory<> & memory, size_t min_chunk_size)
 {
-    skipWhitespacesAndTabs(in);
-
     char * pos = in.position();
     bool quotes = false;
     bool need_more_data = true;
@@ -474,8 +472,9 @@ bool fileSegmentationEngineCSVImpl(ReadBuffer & in, DB::Memory<> & memory, size_
             }
         }
     }
-    loadAtPosition(in, memory, pos);
-    return true;
+
+    saveUpToPosition(in, memory, pos);
+    return loadAtPosition(in, memory, pos);
 }
 
 void registerFileSegmentationEngineCSV(FormatFactory & factory)

dbms/src/Processors/Formats/Impl/JSONEachRowRowInputFormat.cpp

@@ -325,8 +325,9 @@ bool fileSegmentationEngineJSONEachRowImpl(ReadBuffer & in, DB::Memory<> & memor
             }
         }
     }
-    loadAtPosition(in, memory, pos)
-    return true;
+
+    saveUpToPosition(in, memory, pos);
+    return loadAtPosition(in, memory, pos);
 }
 
 void registerFileSegmentationEngineJSONEachRow(FormatFactory & factory)