{"record":{"id":"eddba9f055468c36","repo":"pathwaycom/pathway","slug":"negative-timestamp-cannot-be-used","errorCode":null,"errorMessage":"negative timestamp cannot be used","messagePattern":"negative timestamp cannot be used","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/debug/__init__.py","lineNumber":548,"sourceCode":"        is a mapping from worker id to list of rows processed in this batch by this worker,\n        and each row is tuple (diff, key, values).\n\n        Note: unless you need to specify timestamps and keys, consider using\n        `table_from_list_of_batches` and `table_from_list_of_batches_by_workers`.\n\n        Args:\n            batches: dictionary with specified batches to be put in the table\n            schema: schema of the table\n        \"\"\"\n        unique_name = self._get_next_unique_name()\n        workers = {worker for batch in batches.values() for worker in batch}\n        for worker in workers:\n            self.events[(unique_name, worker)] = []\n\n        timestamps = set(batches.keys())\n\n        if any(timestamp for timestamp in timestamps if timestamp < 0):\n            raise ValueError(\"negative timestamp cannot be used\")\n        elif any(timestamp for timestamp in timestamps if timestamp == 0):\n            warn(\n                \"rows with timestamp 0 are only backfilled and are not processed by output connectors\"\n            )\n\n        if any(timestamp for timestamp in timestamps if timestamp % 2 == 1):\n            warn(\n                \"timestamps are required to be even; all timestamps will be doubled\",\n                stacklevel=_stacklevel + 1,\n            )\n            batches = {2 * timestamp: batches[timestamp] for timestamp in batches}\n\n        for timestamp in sorted(batches):\n            self._advance_time_for_all_workers(unique_name, workers, timestamp)\n            batch = batches[timestamp]\n            for worker, changes in batch.items():\n                for diff, key, values in changes:\n                    if diff == 1:","sourceCodeStart":530,"sourceCodeEnd":566,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/debug/__init__.py#L530-L566","documentation":"pw.io.airbyte.read validates the connector's configured catalog before running it: every stream must have sync_mode set to either \"incremental\" or \"full_refresh\". Any other value in the catalog JSON is rejected up front.","triggerScenarios":"Passing a streams list (or catalog dict) to pw.io.airbyte.read where a stream's sync_mode field is misspelled or unsupported, e.g. {\"stream\": \"users\", \"sync_mode\": \"full_refresh\"} vs a value like \"incremental_\" or \"append\".","commonSituations":"Hand-writing the streams argument from the Airbyte spec and using a destination sync mode (e.g. \"append\", \"overwrite\") instead of the source sync mode; or a catalog JSON exported from Airbyte containing an empty/renamed sync_mode.","solutions":["Set each stream's sync_mode to \"incremental\" or \"full_refresh\".","If copying from an Airbyte catalog export, use the source's sync_mode field, not the destination's destination_sync_mode.","Let Pathway configure streams automatically (e.g. streams=\"*\") instead of hand-writing the catalog."],"exampleFix":"# before\nstreams=[{\"stream\": \"users\", \"sync_mode\": \"append\"}]\n\n# after\nstreams=[{\"stream\": \"users\", \"sync_mode\": \"incremental\"}]","handlingStrategy":"validation","validationCode":"ALLOWED = {\"incremental\", \"full_refresh\"}\nfor s in streams:\n    mode = s[\"sync_mode\"] if isinstance(s, dict) else s.sync_mode\n    assert mode in ALLOWED, f\"bad sync_mode {mode!r} for stream {s}\"\npw.io.airbyte.read(..., streams=streams)","typeGuard":"from typing import Any\n\ndef valid_stream_list(streams: Any) -> bool:\n    if not isinstance(streams, list):\n        return True  # e.g. \"*\" is fine\n    return all(\n        s.get(\"sync_mode\") in {\"incremental\", \"full_refresh\"}\n        for s in streams if isinstance(s, dict)\n    )","tryCatchPattern":null,"preventionTips":["Prefer streams=\"*\" or list-of-names forms and let Pathway build the catalog.","When hand-writing stream dicts, copy sync_mode values from a working run's log, not from destination-mode docs."],"tags":["pathway","airbyte","configuration","validation"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}