full sqlite support (#10191)

GitOrigin-RevId: 14bf058e9448ebde1e676bec676db8b262992977

Author: zxqfd555

CHANGELOG.md

@@ -5,7 +5,11 @@ All notable changes to this project will be documented in this file.
 This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ## [Unreleased]
 
+### Added
+- `pw.io.sqlite.write` connector, which writes a Pathway table into a SQLite database file. Supports two modes: `stream_of_changes` (default) appends each event alongside `time`/`diff` metadata columns, while `snapshot` maintains the current state of the table via `INSERT ... ON CONFLICT DO UPDATE` on insertions and `DELETE` on retractions, keyed on the `primary_key` parameter. Values are encoded using the same storage-class mapping that `pw.io.sqlite.read` accepts, so `write` / `read` round-trips every supported Pathway type losslessly. `init_mode` controls whether the destination table is left as-is, auto-created, or replaced on start-up.
+
 ### Changed
+- `pw.io.sqlite.read` now parses every Pathway `Value` variant. In addition to `int`, `float`, `str`, `bytes`, `pw.Json`, and their `Optional` forms, the reader now accepts `bool`, `pw.DateTimeNaive`, `pw.DateTimeUtc`, `pw.Duration`, `pw.Pointer`, `pw.PyObjectWrapper`, homogeneous `tuple` / `list`, and `np.ndarray`. Composite types are stored as `TEXT` using the same JSON encoding that `pw.io.jsonlines.write` emits. Booleans additionally accept PostgreSQL-style textual literals (`true`/`false`, `yes`/`no`, `on`/`off`, `t`/`f`, `y`/`n`; case-insensitive, whitespace-trimmed), and `float` columns tolerate values stored with `INTEGER` storage class.
 - **BREAKING**: `pw.io.iceberg.write` to a Glue catalog no longer accepts `DateTimeUtc` columns. Glue's metastore has no timezone-aware timestamp type, so previous versions silently dropped the timezone on read-back; writes now fail with an explicit error instead of corrupting the zone. To store UTC timestamps in Glue, convert to `DateTimeNaive` with UTC-normalized values, or write through the REST catalog, which preserves the timezone.
 
 ## [0.30.1] - 2026-04-23

docs/2.developers/4.user-guide/20.connect/30.connectors-in-pathway.md

@@ -92,6 +92,7 @@ Before going into more details about the different connectors and how they work,
                 <span class="block"><a href="/developers/api-docs/pathway-io/rabbitmq">RabbitMQ</a></span>
                 <span class="block"><a href="/developers/user-guide/connect/connectors/switching-to-redpanda">Redpanda</a></span>
                 <span class="block"><a href="/developers/user-guide/connect/connectors/slack_send_alerts">Slack</a></span>
+                <span class="block"><a href="/developers/api-docs/pathway-io/sqlite">SQLite</a></span>
             </td>
             <td class="text-center !align-middle">
                 <span class="block"><a href="/developers/user-guide/connect/connectors/csv_connectors">CSV</a></span>

python/pathway/io/sqlite/__init__.py

@@ -12,7 +12,6 @@
 import random
 import re
 import socket
-import sqlite3
 import sys
 import threading
 import time
@@ -2688,81 +2687,6 @@ class InputSchema(pw.Schema):
     assert_table_equality_wo_index(result, expected)
 
 
-@needs_multiprocessing_fork
-def test_sqlite(tmp_path: pathlib.Path):
-    database_name = tmp_path / "test.db"
-    output_path = tmp_path / "output.csv"
-
-    connection = sqlite3.connect(database_name)
-    cursor = connection.cursor()
-    cursor.execute(
-        """
-        CREATE TABLE users (
-            id INTEGER,
-            login TEXT,
-            name TEXT
-        )
-        """
-    )
-    cursor.execute("INSERT INTO users (id, login, name) VALUES (1, 'alice', 'Alice')")
-    cursor.execute("INSERT INTO users (id, login, name) VALUES (2, 'bob1999', 'Bob')")
-    connection.commit()
-
-    def stream_target():
-        wait_result_with_checker(FileLinesNumberChecker(output_path, 2), 5, target=None)
-        connection = sqlite3.connect(database_name)
-        cursor = connection.cursor()
-        cursor.execute(
-            """
-            INSERT INTO users (id, login, name) VALUES (3, 'ch123', 'Charlie')"""
-        )
-        connection.commit()
-
-        wait_result_with_checker(FileLinesNumberChecker(output_path, 3), 2, target=None)
-        cursor = connection.cursor()
-        cursor.execute("UPDATE users SET name = 'Bob Smith' WHERE id = 2")
-        connection.commit()
-
-        wait_result_with_checker(FileLinesNumberChecker(output_path, 5), 2, target=None)
-        cursor = connection.cursor()
-        cursor.execute("DELETE FROM users WHERE id = 3")
-        connection.commit()
-
-    class InputSchema(pw.Schema):
-        id: int
-        login: str
-        name: str
-
-    table = pw.io.sqlite.read(
-        database_name, "users", InputSchema, autocommit_duration_ms=1
-    )
-    pw.io.jsonlines.write(table, output_path)
-
-    inputs_thread = threading.Thread(target=stream_target, daemon=True)
-    inputs_thread.start()
-
-    wait_result_with_checker(FileLinesNumberChecker(output_path, 6), 30)
-
-    events = []
-    with open(output_path) as f:
-        for row in f:
-            events.append(json.loads(row))
-
-    events.sort(key=lambda event: (event["time"], event["diff"], event["name"]))
-    events_truncated = []
-    for event in events:
-        events_truncated.append([event["name"], event["diff"]])
-
-    assert events_truncated == [
-        ["Alice", 1],
-        ["Bob", 1],
-        ["Charlie", 1],
-        ["Bob", -1],
-        ["Bob Smith", 1],
-        ["Charlie", -1],
-    ]
-
-
 def test_apply_bytes_full_cycle(tmp_path: pathlib.Path):
     input_path = tmp_path / "input.txt"
     input_full_contents = "abc\n\ndef\nghi"