pathwaycom/pathway v0.32.1 on GitHub

Added

pw.io.chroma.write writes a Pathway table to a Chroma collection, keeping the collection in sync with the table as rows are added, changed, and removed. The columns are mapped onto Chroma's record fields explicitly: the optional primary_key becomes the record id (when omitted, the row's internal Pathway key is used instead), embedding the vector, the optional document column the stored text, and metadata_columns the record metadata. The collection must already exist. The server is addressed with host/port (plus ssl, headers, tenant, and database for authenticated deployments such as Chroma Cloud).
pw.io.qdrant.write writes a Pathway table to a Qdrant collection. Each row addition is upserted as a point and each deletion removes the corresponding point, so an update replaces a point rather than duplicating it. The pre-created collection's schema drives the writing: every named vector slot of the collection is bound to the table column with the same name — dense slots to list[float] (or 1-D numpy.ndarray) columns, sparse slots to list[tuple[int, float]] columns of (index, weight) pairs — and all vectors of a point are written atomically in one upsert, enabling native hybrid (dense + BM25) search. The remaining columns are stored in the point payload. The connector fails fast on a missing collection or any slot/column mismatch instead of degrading silently. The batch_size parameter bounds how many points are sent per request, and an optional api_key authenticates against Qdrant Cloud or a secured instance.
pw.io.duckdb.write writes a Pathway table into a DuckDB database file through a native, in-process connector, in either "stream_of_changes" or "snapshot" mode. Embeddings stored as numpy arrays or list[float] columns land in native DOUBLE[] list columns, so the result is directly searchable with DuckDB's vector-distance functions for RAG retrieval. The detach_between_batches option makes the writer close the database after every minibatch commit and reopen it for the next one, releasing the file lock in between — so a separate process (e.g. a query server) can read committed data with short-lived read-only connections while the pipeline keeps running; the writer retries the reopen with a backoff when a reader momentarily holds the lock.
pw.io.weaviate.write writes a Pathway table to a Weaviate collection, keeping it in sync with the table: additions and updates upsert objects (keyed by a UUID derived from the required primary_key), and deletions remove them. An optional vector column is stored as the object's embedding; the remaining columns become object properties. The target collection must already exist. The connector writes in parallel across workers (pathway spawn -n), with batch_size and concurrency to tune throughput, and supports api_key/headers authentication for self-hosted and Weaviate Cloud deployments.
pw.io.pinecone.write writes a Pathway table to a Pinecone index for use as a vector store in RAG pipelines. It keeps the index in sync with the current state of the table: a row is upserted under its record id and a row removed from the table is deleted from the index. By default the id is the table's internal row key (unique, and written in parallel across workers); pass a primary_key column to use your own ids instead, in which case the values must uniquely identify rows (a collision raises an error) and writing runs on a single worker. The vector column holds either a dense embedding (list[float] or a 1-D numpy.ndarray), written to a dense index, or a sparse vector (list[tuple[int, float]] of (index, weight) pairs), written to a sparse index — so a hybrid (dense + BM25) setup is two indexes fed by two write() calls off the same table, with the same record ids landing in both for client-side fusion. The remaining columns — or just those listed in metadata_columns — are stored as record metadata. Targets both Pinecone cloud and the local Pinecone emulator via the host parameter; the index must already exist, with a matching dimension for a dense one, and a mismatch between the index's kind and the column's type is reported at start-up.
pw.run and pw.run_all accept a new udf_cache_directory parameter. When set, the memoization cache of non-deterministic UDFs (any pw.udf without deterministic=True, whose results are stored so that row deletions replay the originally produced values) is kept in SQLite files in the given directory instead of in memory, so the pipeline's memory usage no longer grows with the number of cached results. Note that on many systems /tmp is a RAM-backed tmpfs — point the cache directory at a real disk to actually save memory. The on-disk cache is a runtime working set, not a durability mechanism: persistence snapshots remain the source of truth on restart, the cache files are recreated on every run and removed on shutdown. The default (None, in-memory cache) is unchanged.
pw.io.s3.read, pw.io.s3.read_from_digital_ocean, pw.io.s3.read_from_wasabi, pw.io.minio.read, pw.io.pyfilesystem.read, and pw.xpacks.connectors.sharepoint.read now support the format="only_metadata" option (already available in pw.io.fs.read and pw.io.gdrive.read). In this mode the connector tracks additions, modifications, and deletions of objects in the source but does not download their contents — the resulting table contains only the _metadata column. This is useful for monitoring changes in large buckets or directories without spending time and traffic on fetching the objects themselves. For S3 and MinIO this also skips the object downloads entirely at the engine level, not just the parsing of their contents.
Promoted TwelveLabsVideoParser and MarengoEmbedder out of the Video RAG example template and into the native pathway.xpacks.llm core library. You can now build Video RAG applications directly in Pathway by installing pip install pathway[twelvelabs]. The parser processes videos concurrently on an async executor and accepts the capacity, retry_strategy, async_mode, video_format and on_error parameters (on_error="skip" lets the pipeline continue when a single video fails to parse); oversized videos are rejected before the upload. TwelveLabsVideoParser requires a license key with the advanced-parser entitlement.

Changed

BREAKING: pw.io.airbyte.read replaced the refresh_interval_ms parameter (milliseconds) with refresh_interval, which takes a number of seconds or a datetime.timedelta / pw.Duration and defaults to 60 seconds (equal to the previous default of 60000 ms). Passing refresh_interval_ms now raises an error that includes the converted value to use. Migration: replace refresh_interval_ms=60000 with refresh_interval=60 (or refresh_interval=datetime.timedelta(seconds=60)).
Duration parameters of connectors and xpack components now uniformly accept a number of seconds (int or float) or a datetime.timedelta / pw.Duration. This covers refresh_interval in pw.io.gdrive.read, pw.io.pyfilesystem.read, pw.xpacks.connectors.sharepoint.read and pw.io.airbyte.read, poll_interval and max_transaction_duration in pw.io.elasticsearch.read, quick_access_window, compression_frequency and retention_period in pw.io.deltalake.TableOptimizer, idle_duration in pw.io.SynchronizedColumn, asset_poll_interval and asset_timeout in TwelveLabsVideoParser, and timeout in DocumentStoreClient and RAGClient. Plain numbers keep meaning seconds wherever they did before; the new public alias pw.io.DurationLike names the accepted type. Invalid durations (negative, non-finite, wrong type) are now rejected at call time with an error naming the parameter. A zero polling interval stays allowed and means polling as often as possible; timeout parameters (asset_timeout, client timeout) must be strictly positive. This also fixes pw.io.pyfilesystem.read rejecting an integer refresh_interval (e.g. refresh_interval=30).
pw.xpacks.connectors.sharepoint.read now supports the max_backlog_size parameter, already available in the other input connectors. It limits the number of entries read from SharePoint and kept in processing at any moment, which helps to avoid memory spikes when the source emits a large initial burst of data.
pw.io.mongodb.write now distributes writes across all workers when Pathway runs with several workers (pathway spawn -n N), so write throughput scales with the worker count up to the capacity of the target MongoDB/Atlas deployment. Each document is still written by a single worker, so the result is identical to a single-worker run. Output that requests a global sort_by order continues to run on a single worker, since that is required to preserve the order.
pw.io.fs.read now reads in parallel when Pathway runs with several workers and persistence is not enabled. Each file is assigned to exactly one worker by a stable hash of its path, so the read scales with the number of workers instead of running on a single one. The assignment is deterministic across workers and processes, requiring no coordination between them. All files present at start-up are still delivered as a single minibatch at one shared timestamp across every worker, so stateful operators see the initial snapshot atomically rather than split across the parallel readers. When persistence is enabled, the read runs on a single worker to keep recovery exactly consistent.
pw.io.postgres.write now streams each batch into PostgreSQL through the binary COPY protocol instead of issuing one INSERT per row, giving a large throughput improvement (up to ~100x) on bulk writes. Both output modes use it: stream-of-changes copies straight into the target, while snapshot mode stages each batch in a temporary table and merges it with a single set-based upsert/delete.
pw.io.mssql.write now streams each batch into SQL Server through the native bulk-load protocol (INSERT BULK) instead of issuing one INSERT/MERGE per row, giving a large throughput improvement (roughly ~25x) on bulk writes. Stream-of-changes mode bulk-loads straight into the target table when it is safe to do so (the connector created the table and no column name needs quoting), otherwise it stages each batch in a temporary table and applies it with a single set-based INSERT ... SELECT; snapshot mode stages each batch and applies it with one set-based MERGE/DELETE upsert.
pw.io.mysql.write no longer issues one INSERT per row; it now sends each batch in bulk, giving a large throughput improvement. At start-up the connector probes the server and picks the fastest write path it permits: when the server allows LOAD DATA LOCAL INFILE (the local_infile setting is on), batches stream through it — straight into the target in stream-of-changes mode, or via a temporary staging table merged with a single set-based upsert in snapshot mode; otherwise it falls back to chunked multi-row INSERT statements, which work against any reachable server. Both paths produce identical results and require no configuration change.

Fixed

Constructing a KNN index factory (BruteForceKnnFactory, UsearchKnnFactory, LshKnnFactory) with an OpenAIEmbedder no longer sends a request to the OpenAI API. To learn the vector size, the factory used to ask the embedder to embed ".", so building the dataflow graph required network access and a working API key, and a transient network failure aborted the whole pipeline before it started. For the known embedding models the dimension is now taken from a lookup table; the embedder is still queried for unknown models, and when an explicit dimensions parameter shortens the returned vectors.
Errors raised inside get_embedding_dimension are no longer swallowed. When the call was made from a context with a running event loop (most notably a Jupyter notebook), the coroutine ran in a helper thread whose exception was discarded, so an API timeout surfaced as an unrelated TypeError: 'NoneType' object is not subscriptable with the real cause only printed to stderr. The original exception is now re-raised to the caller.
pathway.xpacks.llm (including pathway.xpacks.llm.parsers and pathway.xpacks.llm.document_store) imports again in an environment with only pathway[xpack-llm] installed. The module used to fail at import time with ModuleNotFoundError for pdf2image and unstructured, which belong to the xpack-llm-docs extra; these imports are lazy again, so parsers that need the missing packages fail only at construction, with an error naming the extra to install.
pw.io.nats.write no longer loses messages or fails with a "too many requests" (429) error when writing to a JetStream stream at scale, especially with several workers (pathway spawn -n N). The JetStream writer used to fire every publish in a minibatch without waiting for acknowledgements, so a large batch — multiplied across workers publishing to the same stream — overwhelmed the server and most of the batch was dropped. The writer now bounds the number of un-acknowledged publishes in flight (applying backpressure) and retries transient failures with exponential backoff, so the whole batch is delivered.
A failed run no longer leaves the failing input connector running in the background. When an input connector exhausts its error retries, the error is reported and pw.run raises — but the connector kept retrying and logging the same error inside the process indefinitely, which polluted logs and could destabilize subprocesses started afterwards. The connector now stops as soon as it reports the fatal error.
pw.io.mqtt.read and pw.io.mqtt.write are now more robust: payloads larger than 10 KB are handled by default, the reader re-subscribes and rides out transient broker disconnects instead of stalling, and the writer no longer blocks the pipeline indefinitely when the broker keeps the connection alive but never confirms deliveries. Publish topics are also validated up front (empty or wildcard topics are rejected with a clear error).
pw.io.milvus.write no longer intermittently fails with a "server unavailable" / "connect failed" error when pointed at a local .db file. The embedded local Milvus server reports itself as started before it actually accepts connections, so under load the first connection could lose the race against the server coming up; the connector now retries the initial connection until the local server is ready.
BedrockChat now correctly routes top_k and other model-specific arguments to the AWS Converse API via additionalModelRequestFields.
Improved concurrent write handling in pw.io.sqlite.write for SQLite databases. Writes to the same database file now produce deterministic output in multi-worker and multi-table setups.
pw.io.elasticsearch.write no longer fails when a minibatch is big enough that its Elasticsearch _bulk request would exceed a server-side limit. The connector reads both the cluster's http.max_content_length (the 413 Request Entity Too Large limit) and indexing_pressure.memory.limit (the 429 Too Many Requests limit, which on a small-heap node trips well below 100 MB) at start-up, and splits the buffered documents across as many bulk requests as needed to stay under whichever is hit first — so large batches are still written in as few requests as possible instead of being rejected. (Both limits fall back to a conservative default if they cannot be read.)
pw.io.elasticsearch.write now retries transient bulk failures with backoff instead of failing the run on the first hiccup. A whole-request rejection or an individual document failing with 429/503 (back-pressure / temporary unavailability) is retried — resending only the documents the server reports as not yet applied, so a retry never duplicates data — while deterministic per-document failures (e.g. a type-mismatched value rejected with 400) are now logged and skipped rather than silently dropped.
pw.io.kafka.read in mode="static" no longer intermittently reads zero rows (or fewer rows than were present) from a topic. The static reader subscribed to the topic and relied on a consumer-group rebalance to assign partitions before its first fetch; under load that assignment could fail to complete within the reader's polling budget, so the read finished having consumed nothing. The static reader now assigns its partitions explicitly and reads each up to the end offset captured at start-up, removing the consumer-group dependency entirely.
Glob pattern filtering in KNN LSH has been optimized; a polynomial algorithm is used instead of exponential.
pw.io.airbyte.read with execution_type="local" (venv method) no longer hangs indefinitely when the download of the connector package from PyPI stalls. Each pip install attempt now runs with a timeout and is retried, and a clear error is raised once all attempts fail. The connector package is also installed exactly once instead of three times, making the connector start-up faster.
Pipelines with persistence enabled no longer intermittently lose the most recent input when a pw.iterate is present. The internal snapshot of the iterated state was written on a branch of the computation graph that no output depends on, so a checkpoint could be committed before those writes finished; after a restart the affected rows were neither replayed nor found in the snapshot, surfacing as "key missing in output table" errors. Checkpoint commits now wait for these writes. Additionally, tables passed to pw.iterate as extra (non-iterated) arguments are now included in the persisted state, so the iteration logic sees them correctly after a restart.
Under pw.PersistenceMode.OPERATOR_PERSISTING, joins that preserve the left side's keys (id=left.id), including join_left, no longer misreport a row update arriving after a restart as a duplicate key. Previously the update was logged as a duplicate-key error and the affected row was replaced with an error value.
With persistence enabled, a pipeline that is restarted more than once in quick succession no longer risks losing input rows. A restarted run could commit a checkpoint whose logical time fell into the previous (killed) run's range, accidentally certifying that run's partially-written state: the next restart then resumed from input offsets whose data was missing from the persisted operator state, so those rows were never replayed. Checkpoint commits are now clamped to the current run's own time range.
With persistence enabled, data read right after a restart (new or modified files) now enters the computation at one shared timestamp across all input sources. Previously each source resumed on its own clock, and cross-source operators with key contracts (.ix, join with id=..., with_universe_of) could observe intermediate states, producing spurious "key missing" or duplicate-key errors on restart. When autocommit_duration_ms=None is set explicitly, the previous per-source behavior is kept, since there is no timer to close the shared start-up batch.
Bumped the beartype dependency upper bound from < 0.16.0 to < 0.22.9 to allow installation with recent Beartype releases. Closes #243.