{"record":{"id":"05647bd2da4f868a","repo":"pathwaycom/pathway","slug":"only-diffs-of-1-and-1-are-supported","errorCode":null,"errorMessage":"only diffs of 1 and -1 are supported","messagePattern":"only diffs of 1 and -1 are supported","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/debug/__init__.py","lineNumber":573,"sourceCode":"            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:\n                        event = api.SnapshotEvent.insert(key, values)\n                        self.events[(unique_name, worker)] += [event] * diff\n                    elif diff == -1:\n                        event = api.SnapshotEvent.delete(key, values)\n                        self.events[(unique_name, worker)] += [event] * (-diff)\n                    else:\n                        raise ValueError(\"only diffs of 1 and -1 are supported\")\n\n        return read(\n            _EmptyConnectorSubject(datasource_name=\"debug.stream-generator\"),\n            name=unique_name,\n            schema=schema,\n        )\n\n    def table_from_list_of_batches_by_workers(\n        self,\n        batches: list[dict[int, list[dict[str, api.Value]]]],\n        schema: type[Schema],\n        _stacklevel: int = 1,\n    ) -> Table:\n        \"\"\"A function that creates a table from a list of batches, where each batch is a mapping\n        from worker id to a list of rows processed by this worker in this batch.\n        Each row is a mapping from column name to a value.\n\n        Args:","sourceCodeStart":555,"sourceCodeEnd":591,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/debug/__init__.py#L555-L591","documentation":"A single pw.io.airbyte.read call can wrap multiple streams, but they must all share the same sync_mode because Pathway's persistence/state handling is uniform per connector. The catalog check raises ValueError when it detects a mix (e.g. one incremental, one full_refresh).","triggerScenarios":"Passing a streams list where at least two entries have different sync_mode values, e.g. [{\"stream\":\"a\",\"sync_mode\":\"incremental\"},{\"stream\":\"b\",\"sync_mode\":\"full_refresh\"}].","commonSituations":"User replicates a multi-stream Airbyte source whose default catalog mixes modes; or merges streams from different sources into one read call during a migration.","solutions":["Make all streams use the same sync_mode (usually \"incremental\").","If modes genuinely must differ, split into two pw.io.airbyte.read calls — one per sync mode — and union the resulting tables.","Regenerate the streams list programmatically so every entry gets the same mode string."],"exampleFix":"# before\nstreams=[{\"stream\":\"a\",\"sync_mode\":\"incremental\"}, {\"stream\":\"b\",\"sync_mode\":\"full_refresh\"}]\nt = pw.io.airbyte.read(..., streams=streams)\n\n# after\nt_a = pw.io.airbyte.read(..., streams=[{\"stream\":\"a\",\"sync_mode\":\"incremental\"}])\nt_b = pw.io.airbyte.read(..., streams=[{\"stream\":\"b\",\"sync_mode\":\"incremental\"}])","handlingStrategy":"validation","validationCode":"modes = {s[\"sync_mode\"] for s in streams if isinstance(s, dict)}\nif len(modes) > 1:\n    raise ValueError(f\"mixed sync_modes {modes}; split into separate read calls\")\npw.io.airbyte.read(..., streams=streams)","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Generate the streams list from one mode constant instead of writing entries individually.","When requirements differ per stream, plan for one connector call per mode from the start."],"tags":["pathway","airbyte","configuration","multi-stream"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}