Fix transactions with compressed files

fsspec/implementations/local.py

@@ -437,6 +437,11 @@ def __getstate__(self):
     def commit(self):
         if self.autocommit:
             raise RuntimeError("Can only commit if not already set to autocommit")
+        if not self.f.closed:
+            # a compression wrapper (e.g., GzipFile) does not close the file
+            # object it was given, so buffered bytes and any trailer may still
+            # be pending here. Windows also refuses to rename an open file.
+            self.f.close()
         try:
             shutil.move(self.temp, self.path)
         except PermissionError as e:

fsspec/implementations/tests/test_local.py

@@ -509,6 +509,19 @@ def test_commit_discard(tmpdir):
         assert not fs.exists(tmpdir + "/bfile")
 
 
+def test_transaction_with_compression(tmpdir):
+    path = str(tmpdir / "afile.gz")
+    fs = LocalFileSystem()
+
+    with fs.transaction:
+        with fs.open(path, "wt", compression="gzip") as f:
+            f.write("data")
+        assert not fs.exists(path)
+
+    with gzip.open(path, "rt") as f:
+        assert f.read() == "data"
+
+
 def test_same_permissions_with_and_without_transaction(tmpdir):
     tmpdir = str(tmpdir)
 

fsspec/spec.py

@@ -1363,16 +1363,15 @@ def open(
                 cache_options=cache_options,
                 **kwargs,
             )
+            if not ac and "r" not in mode:
+                self.transaction.files.append(f)
             if compression is not None:
                 from fsspec.compression import compr
                 from fsspec.core import get_compression
 
                 compression = get_compression(path, compression)
                 compress = compr[compression]
                 f = compress(f, mode=mode[0])
-
-            if not ac and "r" not in mode:
-                self.transaction.files.append(f)
             return f
 
     def touch(self, path, truncate=True, **kwargs):