diff --git a/src/datajoint/adapters/base.py b/src/datajoint/adapters/base.py index 5b4a7ab72..1aa12f52f 100644 --- a/src/datajoint/adapters/base.py +++ b/src/datajoint/adapters/base.py @@ -1093,6 +1093,28 @@ def current_timestamp_expr(self, precision: int | None = None) -> str: """ ... + @abstractmethod + def statement_timestamp_expr(self, precision: int | None = None) -> str: + """ + Expression for the time at which the current statement began. + + Unlike ``CURRENT_TIMESTAMP`` on PostgreSQL, which is the start of the + enclosing transaction, this is the start of the statement itself, so it + advances between statements within one transaction. + + Parameters + ---------- + precision : int, optional + Fractional seconds precision (0-6). + + Returns + ------- + str + SQL expression for the statement's start time (e.g., 'NOW(3)' for + MySQL, 'statement_timestamp()' for PostgreSQL). + """ + ... + @abstractmethod def interval_expr(self, value: int, unit: str) -> str: """ diff --git a/src/datajoint/adapters/mysql.py b/src/datajoint/adapters/mysql.py index 2ef9baf84..5f361204a 100644 --- a/src/datajoint/adapters/mysql.py +++ b/src/datajoint/adapters/mysql.py @@ -822,6 +822,26 @@ def current_timestamp_expr(self, precision: int | None = None) -> str: return f"CURRENT_TIMESTAMP({precision})" return "CURRENT_TIMESTAMP" + def statement_timestamp_expr(self, precision: int | None = None) -> str: + """ + Statement start time for MySQL. + + MySQL's ``NOW()`` is the time at which the statement began executing. + + Parameters + ---------- + precision : int, optional + Fractional seconds precision (0-6). + + Returns + ------- + str + NOW() or NOW(n). + """ + if precision is not None: + return f"NOW({precision})" + return "NOW()" + def interval_expr(self, value: int, unit: str) -> str: """ INTERVAL expression for MySQL. diff --git a/src/datajoint/adapters/postgres.py b/src/datajoint/adapters/postgres.py index 3a55d8eb8..c23d40a98 100644 --- a/src/datajoint/adapters/postgres.py +++ b/src/datajoint/adapters/postgres.py @@ -1051,6 +1051,27 @@ def current_timestamp_expr(self, precision: int | None = None) -> str: return f"CURRENT_TIMESTAMP({precision})" return "CURRENT_TIMESTAMP" + def statement_timestamp_expr(self, precision: int | None = None) -> str: + """ + Statement start time for PostgreSQL. + + PostgreSQL's ``CURRENT_TIMESTAMP`` is the start of the transaction; + ``statement_timestamp()`` is the start of the current statement. + + Parameters + ---------- + precision : int, optional + Fractional seconds precision (0-6). + + Returns + ------- + str + statement_timestamp(), cast to timestamptz(n) when a precision is given. + """ + if precision is not None: + return f"CAST(statement_timestamp() AS timestamptz({precision}))" + return "statement_timestamp()" + def interval_expr(self, value: int, unit: str) -> str: """ INTERVAL expression for PostgreSQL. diff --git a/src/datajoint/autopopulate.py b/src/datajoint/autopopulate.py index 61b36c5c9..1738b52e5 100644 --- a/src/datajoint/autopopulate.py +++ b/src/datajoint/autopopulate.py @@ -662,7 +662,9 @@ def _populate1( if jobs is not None and not jobs.reserve(key): return False - start_time = time.time() + # Elapsed time is measured on this host's monotonic clock. The start time + # itself is read from the server's clock when the job metadata is written. + started = time.monotonic() # if make is a generator, transaction can be delayed until the final stage is_generator = inspect.isgeneratorfunction(make) @@ -730,7 +732,7 @@ def _populate1( return key, error if return_exception_objects else error_message else: self.connection.commit_transaction() - duration = time.time() - start_time + duration = time.monotonic() - started logger.jobs(f"Success making {key} -> {self.full_table_name}") # Update hidden job metadata if table has the columns @@ -739,7 +741,6 @@ def _populate1( self._update_job_metadata( key, - start_time=datetime.datetime.fromtimestamp(start_time), duration=duration, version=_get_job_version(self.connection._config), ) @@ -852,25 +853,33 @@ def _has_job_metadata_attrs(self): all_attrs = self.heading._attributes return all_attrs is not None and "_job_start_time" in all_attrs - def _update_job_metadata(self, key, start_time, duration, version): + def _update_job_metadata(self, key, duration, version): """ Update hidden job metadata for the given key. + ``_job_start_time`` is read from the server's clock: the time at which + this statement runs, less ``duration``. Every host that populates the + schema therefore records it on one clock, whatever time zone the host + runs in. ``_job_duration`` is measured on the populating host. + Parameters ---------- key : dict Primary key identifying the row(s) to update. - start_time : datetime - When computation started. duration : float - Computation duration in seconds. + Computation duration in seconds, measured on the populating host. version : str Code version (truncated to 64 chars). """ from .condition import make_condition + adapter = self.connection.adapter pk_condition = make_condition(self, key, set()) + start_time = ( + f"{adapter.statement_timestamp_expr(3)} - {adapter.interval_expr(round(duration * 1_000_000), 'microsecond')}" + ) self.connection.query( - f"UPDATE {self.full_table_name} SET _job_start_time=%s, _job_duration=%s, _job_version=%s WHERE {pk_condition}", - args=(start_time, duration, version[:64] if version else ""), + f"UPDATE {self.full_table_name} SET _job_start_time={start_time}, _job_duration=%s, _job_version=%s " + f"WHERE {pk_condition}", + args=(duration, version[:64] if version else ""), ) diff --git a/tests/integration/test_hidden_job_metadata.py b/tests/integration/test_hidden_job_metadata.py index bcf56dde7..9171b8255 100644 --- a/tests/integration/test_hidden_job_metadata.py +++ b/tests/integration/test_hidden_job_metadata.py @@ -1,5 +1,6 @@ """Tests for hidden job metadata in computed tables.""" +import datetime import time import pytest @@ -271,3 +272,80 @@ def make(self, key): finally: schema.drop() dj.config.jobs.add_job_metadata = original_setting + + +@pytest.fixture +def schema_by_backend(connection_by_backend, db_creds_by_backend): + """A fresh schema with job metadata enabled, on each backend.""" + backend = db_creds_by_backend["backend"] + test_id = str(int(time.time() * 1000))[-8:] + schema_name = f"djtest_job_metadata_{backend}_{test_id}"[:64] + quoted = connection_by_backend.adapter.quote_identifier(schema_name) + original_setting = dj.config.jobs.add_job_metadata + dj.config.jobs.add_job_metadata = True + connection_by_backend.query(f"DROP DATABASE IF EXISTS {quoted}") + schema = dj.Schema(schema_name, connection=connection_by_backend) + yield schema + dj.config.jobs.add_job_metadata = original_setting + if connection_by_backend.is_connected: + connection_by_backend.query(f"DROP DATABASE IF EXISTS {quoted}") + + +@pytest.fixture +def host_time_zone(monkeypatch): + """Set the time zone of the populating process; the original is restored afterwards.""" + + def set_zone(zone): + monkeypatch.setenv("TZ", zone) + time.tzset() + + yield set_zone + monkeypatch.undo() + time.tzset() + + +@pytest.mark.skipif(not hasattr(time, "tzset"), reason="changing the process time zone needs time.tzset") +def test_job_start_time_uses_server_clock(schema_by_backend, host_time_zone): + """ + _job_start_time is read from the server's clock, whatever time zone the populating host runs in. + + Two populate() calls run as if from hosts on opposite sides of the date line, 26 hours apart. + Each start time must fall between two readings of the server's clock taken around the call (#1584). + """ + + @schema_by_backend + class Source(dj.Lookup): + definition = """ + source_id : int16 + """ + contents = [{"source_id": 1}, {"source_id": 2}] + + @schema_by_backend + class Timed(dj.Computed): + definition = """ + -> Source + --- + result : float32 + """ + + def make(self, key): + time.sleep(0.2) + self.insert1(dict(key, result=1.0)) + + connection = schema_by_backend.connection + tolerance = datetime.timedelta(seconds=1) + + def server_now(): + return connection.query("SELECT LOCALTIMESTAMP(3)").fetchone()[0] + + for source_id, zone in ((1, "Etc/GMT+12"), (2, "Etc/GMT-14")): + host_time_zone(zone) + before = server_now() + Timed.populate({"source_id": source_id}) + after = server_now() + start_time, duration = connection.query( + f"SELECT _job_start_time, _job_duration FROM {Timed.full_table_name} WHERE source_id = {source_id}" + ).fetchone() + assert before - tolerance <= start_time <= after, f"host zone {zone}: {start_time} is not within [{before}, {after}]" + assert duration >= 0.2 + assert start_time + datetime.timedelta(seconds=duration) <= after + tolerance diff --git a/tests/unit/test_adapters.py b/tests/unit/test_adapters.py index bd0afa9e0..0e6e65441 100644 --- a/tests/unit/test_adapters.py +++ b/tests/unit/test_adapters.py @@ -166,6 +166,11 @@ def test_current_timestamp_expr(self, adapter): assert adapter.current_timestamp_expr() == "CURRENT_TIMESTAMP" assert adapter.current_timestamp_expr(3) == "CURRENT_TIMESTAMP(3)" + def test_statement_timestamp_expr(self, adapter): + """Test statement start time expression.""" + assert adapter.statement_timestamp_expr() == "NOW()" + assert adapter.statement_timestamp_expr(3) == "NOW(3)" + def test_interval_expr(self, adapter): """Test INTERVAL expression.""" assert adapter.interval_expr(5, "second") == "INTERVAL 5 SECOND" @@ -311,6 +316,11 @@ def test_current_timestamp_expr(self, adapter): assert adapter.current_timestamp_expr() == "CURRENT_TIMESTAMP" assert adapter.current_timestamp_expr(3) == "CURRENT_TIMESTAMP(3)" + def test_statement_timestamp_expr(self, adapter): + """Test statement start time expression: not CURRENT_TIMESTAMP, which is the transaction's start.""" + assert adapter.statement_timestamp_expr() == "statement_timestamp()" + assert adapter.statement_timestamp_expr(3) == "CAST(statement_timestamp() AS timestamptz(3))" + def test_interval_expr(self, adapter): """Test INTERVAL expression with PostgreSQL syntax.""" assert adapter.interval_expr(5, "second") == "INTERVAL '5 seconds'" @@ -403,6 +413,7 @@ def test_adapter_implements_interface(self, backend): "commit_sql", "rollback_sql", "current_timestamp_expr", + "statement_timestamp_expr", "interval_expr", "json_path_expr", "format_column_definition",