Keep multipart CSV rows intact when a part lacks a trailing newline

frictionless/schemes/multipart/__spec__/test_loader.py

@@ -3,8 +3,8 @@
 
 import pytest
 
-from frictionless import FrictionlessException, platform, schemes
-from frictionless.resources import TableResource
+from frictionless import Dialect, FrictionlessException, platform, schemes
+from frictionless.resources import FileResource, TableResource
 
 BASEURL = "https://raw.githubusercontent.com/frictionlessdata/frictionless-py/master/%s"
 
@@ -175,3 +175,48 @@ def test_multipart_loader_with_compressed_parts_issue_1215():
             {"id": 1, "name": "english"},
             {"id": 2, "name": "中国人"},
         ]
+
+
+def test_multipart_loader_part_without_trailing_newline_issue_1778(tmpdir):
+    # The last row of a part without a trailing newline was glued onto the
+    # first row of the next part ({"id": 1, "name": "english2"}).
+    path1 = str(tmpdir.join("chunk1.csv"))
+    path2 = str(tmpdir.join("chunk2.csv"))
+    with open(path1, "wb") as file:
+        file.write(b"id,name\n1,english")
+    with open(path2, "wb") as file:
+        file.write(b"id,name\n2,german\n")
+    with TableResource(path=path1, extrapaths=[path2]) as resource:
+        assert resource.read_rows() == [
+            {"id": 1, "name": "english"},
+            {"id": 2, "name": "german"},
+        ]
+
+
+def test_multipart_loader_headless_part_without_trailing_newline_issue_1778(tmpdir):
+    path1 = str(tmpdir.join("chunk1.csv"))
+    path2 = str(tmpdir.join("chunk2.csv"))
+    with open(path1, "wb") as file:
+        file.write(b"1,english")
+    with open(path2, "wb") as file:
+        file.write(b"2,german\n")
+    resource = TableResource(
+        path=path1, extrapaths=[path2], dialect=Dialect(header=False)
+    )
+    with resource:
+        assert resource.read_rows() == [
+            {"field1": 1, "field2": "english"},
+            {"field1": 2, "field2": "german"},
+        ]
+
+
+def test_multipart_loader_binary_parts_stay_verbatim_issue_1778(tmpdir):
+    # Non-tabular parts are chunks of one file: no newline may be inserted.
+    path1 = str(tmpdir.join("part1.bin"))
+    path2 = str(tmpdir.join("part2.bin"))
+    with open(path1, "wb") as file:
+        file.write(b"\x00\x01NOEOL")
+    with open(path2, "wb") as file:
+        file.write(b"\x02\x03\n")
+    with FileResource(path=path1, extrapaths=[path2]) as resource:
+        assert resource.read_file() == b"\x00\x01NOEOL\x02\x03\n"

frictionless/schemes/multipart/loader.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import tempfile
-from typing import Any, List
+from typing import Any
 
 from ... import helpers, types
 from ...resources import FileResource
@@ -21,12 +21,14 @@ class MultipartLoader(Loader):
     def read_byte_stream_create(self):  # type: ignore
         assert self.resource.normpath
         remote = self.resource.remote
+        tabular = self.resource.format == "csv"
         headless = self.resource.dialect.header is False
-        headless = headless or self.resource.format != "csv"
+        headless = headless or not tabular
         return MultipartByteStream(
             self.resource.normpaths,
             remote=remote,
             headless=headless,
+            tabular=tabular,
         )
 
     # Write
@@ -50,10 +52,11 @@ def write_byte_stream_save(self, byte_stream: types.IByteStream):
 
 
 class MultipartByteStream:
-    def __init__(self, paths: List[str], *, remote: bool, headless: bool):
+    def __init__(self, paths: list[str], *, remote: bool, headless: bool, tabular: bool):
         self.__paths = paths
         self.__remote = remote
         self.__headless = headless
+        self.__tabular = tabular
         self.__line_stream = self.read_line_stream()
 
     def __enter__(self):
@@ -105,8 +108,22 @@ def read(self, size: int):
 
     def read_line_stream(self):
         for number, path in enumerate(self.__paths, start=1):
+            last_line = None
             with FileResource(path=path) as resource:
                 for line_number, line in enumerate(resource.byte_stream, start=1):
                     if not self.__headless and number > 1 and line_number == 1:
                         continue
-                    yield line
+                    if last_line is not None:
+                        yield last_line
+                    last_line = line
+            if last_line is not None:
+                # A tabular part that does not end with a newline would glue its
+                # last row onto the first row of the next part. Binary parts are
+                # concatenated verbatim: they are chunks of one file.
+                if (
+                    self.__tabular
+                    and number < len(self.__paths)
+                    and not last_line.endswith(b"\n")
+                ):
+                    last_line += b"\n"
+                yield last_line