Skip to content
29 changes: 28 additions & 1 deletion src/pyfilesystem.cpp
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
#include "duckdb_python/pyfilesystem.hpp"

#include "duckdb/common/exception.hpp"
#include "duckdb/common/string_util.hpp"
#include "duckdb_python/nb/casters.hpp"

Expand Down Expand Up @@ -202,12 +203,38 @@ void PythonFilesystem::RemoveFile(const string &filename, optional_ptr<FileOpene
auto remove = filesystem.attr("rm");
remove(nb::str(filename.c_str(), filename.size()));
}

static bool IsUnsupportedModificationTimeError(const nb::python_error &error) {
if (error.matches(PyExc_NotImplementedError)) {
return true;
}
if (!error.matches(PyExc_KeyError)) {
return false;
}
// gcsfs: GCSFileSystem.modified() indexes info(path)["mtime"] and raises KeyError('mtime') when
// object metadata (or synthesized directory entries) has no mtime.
try {
nb::tuple args = nb::cast<nb::tuple>(error.value().attr("args"));
return nb::len(args) == 1 && nb::cast<string>(nb::str(args[0])) == "mtime";
} catch (...) {
return false;
}
}

timestamp_t PythonFilesystem::GetLastModifiedTime(FileHandle &handle) {
D_ASSERT(!duckdb::PyUtil::GilCheck());
// TODO: this value should be cached on the PythonFileHandle
nb::gil_scoped_acquire gil;

auto last_mod = filesystem.attr("modified")(handle.path);
nb::object last_mod;
try {
last_mod = filesystem.attr("modified")(handle.path);
} catch (nb::python_error &e) {
if (IsUnsupportedModificationTimeError(e)) {
throw NotImplementedException("%s: GetLastModifiedTime is not implemented", GetName());
}
throw;
}

// datetime.timestamp() returns a float; truncate to int64 seconds (nb::cast<int64_t> would reject a float)
return Timestamp::FromEpochSeconds((int64_t)nb::cast<double>(last_mod.attr("timestamp")()));
Expand Down
71 changes: 71 additions & 0 deletions tests/fast/api/test_fsspec.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,43 @@

import pytest

import duckdb

fsspec = pytest.importorskip("fsspec")


def _register_blob_filesystem(duckdb_cursor, protocol, modified_fn):
"""Register a tiny fsspec filesystem that serves one in-memory blob."""

class BlobFileSystem(fsspec.AbstractFileSystem):
def ls(self, path, detail=True, **kwargs):
vals = [k for k in self._data if k.startswith(path)]
if detail:
return [
{"name": name, "size": len(self._data[name]), "type": "file", "created": 0, "islink": False}
for name in vals
]
return vals

def modified(self, path):
return modified_fn(path)

def _open(self, path, **kwargs):
return io.BytesIO(self._data[path])

def info(self, path, **kwargs):
return {"name": path, "size": len(self._data[path]), "type": "file"}

def __init__(self) -> None:
super().__init__()
self._data = {"blob": b"hello"}

BlobFileSystem.protocol = protocol
fsspec.register_implementation(protocol, BlobFileSystem, clobber=True)
duckdb_cursor.register_filesystem(fsspec.filesystem(protocol))
return f"{protocol}://blob"


class TestReadParquet:
def test_fsspec_deadlock(self, duckdb_cursor, tmp_path):
# Create test parquet data
Expand Down Expand Up @@ -103,3 +137,40 @@ def __init__(self) -> None:
"GROUP BY ALL ORDER BY file_id"
).fetchall()
assert result == [(0, 10000), (1, 10000), (2, 10000), (3, 10000)]


class TestLastModified:
def test_unsupported_modified_is_null(self, duckdb_cursor):
def raise_not_implemented(_path):
msg = "no mtime"
raise NotImplementedError(msg)

path = _register_blob_filesystem(duckdb_cursor, "nomtime", raise_not_implemented)
result = duckdb_cursor.sql(f"SELECT last_modified FROM read_blob('{path}')").fetchall()
assert result == [(None,)]

def test_mtime_keyerror_is_null(self, duckdb_cursor):
def raise_mtime_key_error(_path):
key = "mtime"
raise KeyError(key)

path = _register_blob_filesystem(duckdb_cursor, "gcsmt", raise_mtime_key_error)
result = duckdb_cursor.sql(f"SELECT last_modified FROM read_blob('{path}')").fetchall()
assert result == [(None,)]

def test_other_modified_errors_still_fail(self, duckdb_cursor):
def raise_os_error(_path):
msg = "simulated I/O failure"
raise OSError(msg)

path = _register_blob_filesystem(duckdb_cursor, "badmtime", raise_os_error)
with pytest.raises(duckdb.Error, match="simulated I/O failure"):
duckdb_cursor.sql(f"SELECT last_modified FROM read_blob('{path}')").fetchall()

def test_modified_timestamp_is_returned(self, duckdb_cursor):
def known_mtime(_path):
return datetime.datetime(2024, 1, 2, tzinfo=datetime.timezone.utc)

path = _register_blob_filesystem(duckdb_cursor, "okmtime", known_mtime)
result = duckdb_cursor.sql(f"SELECT last_modified FROM read_blob('{path}')").fetchall()
assert result[0][0] is not None