diff --git a/CHANGELOG.md b/CHANGELOG.md index 69eb8c1a..930b8e84 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## 2026-07-30 1. Gated the UI behind Supabase email/password auth (issue #124): a login page, a protected route group that redirects unauthenticated visitors to sign-in and returns them to their intended destination, and a sign-out control showing the signed-in user's email. Access is invite-only, with users added directly in Supabase. Also removed job-polling status flicker. [PR #140](https://github.com/METResearchGroup/lab_data_integrations_interface/pull/140) +2. Bluesky Jetstream ingestion now commits to Iceberg tables in the Glue catalog (`bluesky_raw`) instead of writing Parquet to disk, retrying transient commit failures and dead-lettering batches to `s3://lab-data-integrations-interface/dead_letter/` when they cannot land. Replaced the `data_platform` Terraform stack with `terraform/bluesky_ingestion_jetstream/`, which destroyed the previous Glue database, both Athena workgroups, and the pipeline-runs DynamoDB table that `backend/routes/posts.py` still references. [PR #141](https://github.com/METResearchGroup/lab_data_integrations_interface/pull/141) ## 2026-07-29 diff --git a/bluesky_ingestion_jetstream/aws/__init__.py b/bluesky_ingestion_jetstream/aws/__init__.py new file mode 100644 index 00000000..8c50896c --- /dev/null +++ b/bluesky_ingestion_jetstream/aws/__init__.py @@ -0,0 +1 @@ +"""AWS-facing pieces of the Jetstream pipeline: Glue catalog and Iceberg tables.""" diff --git a/bluesky_ingestion_jetstream/aws/bootstrap.py b/bluesky_ingestion_jetstream/aws/bootstrap.py new file mode 100644 index 00000000..564499dc --- /dev/null +++ b/bluesky_ingestion_jetstream/aws/bootstrap.py @@ -0,0 +1,97 @@ +"""One-shot creation of the four Iceberg tables. Run by hand, not by the ingester. + + python -m bluesky_ingestion_jetstream.aws.bootstrap + +Idempotent: tables that already exist are reported and left alone, so a partial +failure can be resolved by re-running. Dropping a table is not offered here -- +it discards data, and doing it by hand is the point. + +The Glue database itself is Terraform's +(`terraform/bluesky_ingestion_jetstream/main.tf`); the +tables are not, because Iceberg rewrites a table's schema, partition spec, and +snapshot pointer on every commit, which an `aws_glue_catalog_table` resource +would read as drift and revert on the next apply. +""" + +from pyiceberg.catalog.glue import GlueCatalog +from pyiceberg.exceptions import NoSuchNamespaceError, TableAlreadyExistsError +from pyiceberg.table import Table +from pyiceberg.transforms import DayTransform + +from bluesky_ingestion_jetstream.aws.catalog import build_catalog +from bluesky_ingestion_jetstream.aws.constants import ( + GLUE_DATABASE, + PARTITION_FIELD_NAME, + PARTITION_SOURCE_COLUMN, + TABLE_LOCATIONS, + TABLE_PROPERTIES, +) +from bluesky_ingestion_jetstream.schemas.arrow_schemas import RECORD_TYPE_TO_SCHEMA + + +def require_namespace(catalog: GlueCatalog) -> None: + """Fail if Terraform has not created the Glue database yet.""" + + try: + catalog.list_tables(GLUE_DATABASE) + except NoSuchNamespaceError as error: + raise RuntimeError( + f"Glue database {GLUE_DATABASE!r} does not exist. It is managed by " + "Terraform -- run `terraform apply` in " + "terraform/bluesky_ingestion_jetstream first." + ) from error + + +def create_table(catalog: GlueCatalog, record_type: str) -> Table: + """Create one partitioned table from its Arrow schema. + + The Arrow schema is passed through untouched so PyIceberg assigns the field + IDs itself. Hand-written IDs are renumbered on create, and because Iceberg + resolves columns by ID rather than name, any mismatch reads back as NULL for + every affected column rather than failing. Not writing IDs at all removes + that failure mode instead of guarding against it. + """ + + table = catalog.create_table( + identifier=(GLUE_DATABASE, record_type), + schema=RECORD_TYPE_TO_SCHEMA[record_type], + location=TABLE_LOCATIONS[record_type], + properties=TABLE_PROPERTIES, + ) + + # Partition after creation rather than passing a `PartitionSpec`, because + # `add_field` takes the column name and resolves the source ID itself. + table.update_spec().add_field( + PARTITION_SOURCE_COLUMN, DayTransform(), PARTITION_FIELD_NAME + ).commit() + + return table + + +def bootstrap(catalog: GlueCatalog | None = None) -> dict[str, Table]: + """Create every missing table. Returns record type -> table for all four.""" + + catalog = catalog or build_catalog() + require_namespace(catalog) + + tables: dict[str, Table] = {} + for record_type in RECORD_TYPE_TO_SCHEMA: + try: + tables[record_type] = create_table(catalog, record_type) + print(f"created {GLUE_DATABASE}.{record_type} -> {TABLE_LOCATIONS[record_type]}") + except TableAlreadyExistsError: + tables[record_type] = catalog.load_table((GLUE_DATABASE, record_type)) + print(f"exists {GLUE_DATABASE}.{record_type}") + + return tables + + +def main() -> None: + """CLI entry point.""" + + for record_type, table in bootstrap().items(): + print(f"{record_type}: {table.spec()}") + + +if __name__ == "__main__": + main() diff --git a/bluesky_ingestion_jetstream/aws/catalog.py b/bluesky_ingestion_jetstream/aws/catalog.py new file mode 100644 index 00000000..0ec9d902 --- /dev/null +++ b/bluesky_ingestion_jetstream/aws/catalog.py @@ -0,0 +1,98 @@ +"""Connect to the Glue catalog and load the four Iceberg tables. + +Runs on every process start, unlike `bootstrap.py`. Nothing here creates or +alters a table: an ingester that can issue DDL is one typo in a database name +away from quietly writing a full day of data into a second, empty set of tables +that looks fine until someone queries the real ones. Missing tables raise. +""" + +import boto3 +from botocore.config import Config +from pyiceberg.catalog.glue import GlueCatalog +from pyiceberg.exceptions import NoSuchTableError +from pyiceberg.table import Table + +from bluesky_ingestion_jetstream.aws.constants import ( + AWS_REGION, + GLUE_CONNECT_TIMEOUT_SECONDS, + GLUE_DATABASE, + GLUE_MAX_ATTEMPTS, + GLUE_READ_TIMEOUT_SECONDS, + S3_BUCKET, + S3_CONNECT_TIMEOUT_SECONDS, + S3_PREFIX, + S3_REQUEST_TIMEOUT_SECONDS, +) +from bluesky_ingestion_jetstream.constants import RECORD_TYPES + + +class MissingTablesError(RuntimeError): + """Raised when the catalog is missing tables `bootstrap.py` should have created.""" + + +def build_glue_client(): + """A Glue client with a bounded worst case. + + Built here rather than left to PyIceberg because its default is `standard` + retry mode with ten attempts over a 60s read timeout, which puts an + open-ended retry loop underneath every commit. The commit is retried at a + higher level where the failure can be dead-lettered, so this layer only needs + to cover a single dropped packet, not an outage. + """ + + return boto3.client( + "glue", + region_name=AWS_REGION, + config=Config( + retries={"max_attempts": GLUE_MAX_ATTEMPTS, "mode": "standard"}, + connect_timeout=GLUE_CONNECT_TIMEOUT_SECONDS, + read_timeout=GLUE_READ_TIMEOUT_SECONDS, + ), + ) + + +def build_catalog() -> GlueCatalog: + """Construct the Glue-backed catalog. + + Left on PyIceberg's default PyArrowFileIO. The Iceberg experiment pinned + FsspecFileIO so its S3 meter could see every request through aiobotocore; + production has nothing to meter and PyArrow's client is faster. + """ + + return GlueCatalog( + name="bluesky", + client=build_glue_client(), + **{ + "warehouse": f"s3://{S3_BUCKET}/{S3_PREFIX}", + "glue.region": AWS_REGION, + "s3.region": AWS_REGION, + "s3.connect-timeout": S3_CONNECT_TIMEOUT_SECONDS, + "s3.request-timeout": S3_REQUEST_TIMEOUT_SECONDS, + }, + ) + + +def load_tables(catalog: GlueCatalog) -> dict[str, Table]: + """Load every record type's table, or raise naming all the ones missing. + + Called once at startup rather than per flush, because each load is a Glue + `GetTable` call. Every missing table is collected before raising, so a fresh + environment reports all four in one go instead of one per re-run. + """ + + tables: dict[str, Table] = {} + missing: list[str] = [] + + for record_type in RECORD_TYPES: + try: + tables[record_type] = catalog.load_table((GLUE_DATABASE, record_type)) + except NoSuchTableError: + missing.append(record_type) + + if missing: + raise MissingTablesError( + f"Glue database {GLUE_DATABASE!r} is missing table(s): {', '.join(missing)}. " + "Run `python -m bluesky_ingestion_jetstream.aws.bootstrap` to create them." + ) + + return tables diff --git a/bluesky_ingestion_jetstream/aws/constants.py b/bluesky_ingestion_jetstream/aws/constants.py new file mode 100644 index 00000000..04e4cf9c --- /dev/null +++ b/bluesky_ingestion_jetstream/aws/constants.py @@ -0,0 +1,75 @@ +"""AWS identifiers and Iceberg table configuration for this pipeline.""" + +from bluesky_ingestion_jetstream.constants import RECORD_TYPES + +AWS_REGION = "us-east-2" +S3_BUCKET = "lab-data-integrations-interface" +S3_PREFIX = "bluesky/raw" + +# Created by Terraform (`terraform/bluesky_ingestion_jetstream/main.tf`). +GLUE_DATABASE = "bluesky_raw" + +# One table per record type. Glue names cannot contain `/`, so the location is +# passed explicitly at creation. +TABLE_LOCATIONS = { + record_type: f"s3://{S3_BUCKET}/{S3_PREFIX}/{record_type}" for record_type in RECORD_TYPES +} + +# Applied at table creation only; edits here do not reach existing tables. +TABLE_PROPERTIES = { + "format-version": "2", + "write.parquet.compression-codec": "zstd", + "write.target-file-size-bytes": str(256 * 1024 * 1024), + # Iceberg's default salts a hash into the data path; unnecessary at this + # volume, and it costs a browsable `created_at_day=.../` layout. + "write.object-storage.enabled": "false", + "write.metadata.delete-after-commit.enabled": "true", + "write.metadata.previous-versions-max": "100", + # Inert while ingestion is append-only. Set now for the duplicates backfill + # will introduce; PyIceberg cannot write delete files, so that merge needs + # Athena or Spark. + "write.delete.mode": "merge-on-read", + "write.update.mode": "merge-on-read", + "write.merge.mode": "merge-on-read", +} + +# Iceberg's default name for a `day()` transform, as it appears on disk. +PARTITION_SOURCE_COLUMN = "created_at" +PARTITION_FIELD_NAME = "created_at_day" + +# --------------------------------------------------------------------------- +# Client bounds +# +# Overrides PyIceberg's Glue defaults +# --------------------------------------------------------------------------- + +GLUE_MAX_ATTEMPTS = 2 +GLUE_CONNECT_TIMEOUT_SECONDS = 3.0 +GLUE_READ_TIMEOUT_SECONDS = 10.0 + +# Passed to the PyArrow S3 filesystem (pyiceberg/io/pyarrow.py:445-448). +S3_CONNECT_TIMEOUT_SECONDS = 3.0 +S3_REQUEST_TIMEOUT_SECONDS = 15.0 + +# --------------------------------------------------------------------------- +# Commit retry +# +# Three attempts, two sleeps +# --------------------------------------------------------------------------- + +COMMIT_MAX_ATTEMPTS = 3 +COMMIT_INITIAL_DELAY_SECONDS = 1.0 +COMMIT_MAX_DELAY_SECONDS = 8.0 + +# Stamped on the snapshot, so a retry can tell a failed commit from a lost reply. +SNAPSHOT_FLUSH_ID_TAG = "flush_id" + +# --------------------------------------------------------------------------- +# Dead letter +# +# Outside `S3_PREFIX`: orphan cleanup deletes unreferenced files under the +# warehouse root, and these are unreferenced by definition. +# --------------------------------------------------------------------------- + +DEAD_LETTER_PREFIX = "dead_letter/bluesky/raw" +DEAD_LETTER_ROOT = f"{S3_BUCKET}/{DEAD_LETTER_PREFIX}" diff --git a/bluesky_ingestion_jetstream/aws/dead_letter.py b/bluesky_ingestion_jetstream/aws/dead_letter.py new file mode 100644 index 00000000..2e2424aa --- /dev/null +++ b/bluesky_ingestion_jetstream/aws/dead_letter.py @@ -0,0 +1,134 @@ +"""Durable landing place for batches whose Iceberg commit could not be made. + +These rows are not in the tables. Every query run afterwards is short by exactly +this batch, and nothing about the table itself will say so -- the log line and +the presence of files under this prefix are the only signals, which is why both +matter more than they look. + +Two decisions worth keeping: + +The path sits above `bluesky/`, well outside the warehouse root. A dead-letter +file is unreferenced by any Iceberg table by definition, and unreferenced files +under a table's location are precisely what orphan cleanup deletes. Storing them +in the warehouse means a future maintenance sweep silently destroys the only +remaining copy of unrecovered data. + +The Parquet is built from the *module* schema, not the table's. That is what +keeps this path working in the case it is most needed: if the declared schema and +the table's have diverged, the append fails on incompatibility while this write +still succeeds, because the rows came from parsers that produce exactly this +shape. +""" + +import logging +from datetime import UTC, datetime +from uuid import uuid4 + +import pyarrow as pa +import pyarrow.parquet as pq +from pyarrow.fs import FileSystem, S3FileSystem + +from bluesky_ingestion_jetstream.aws.constants import ( + AWS_REGION, + DEAD_LETTER_ROOT, + S3_CONNECT_TIMEOUT_SECONDS, + S3_REQUEST_TIMEOUT_SECONDS, +) +from bluesky_ingestion_jetstream.schemas.arrow_schemas import RECORD_TYPE_TO_SCHEMA +from lib.timestamp_utils import CREATED_AT_FORMAT + +logger = logging.getLogger(__name__) + + +class DeadLetterError(RuntimeError): + """Raised when a batch could not be written to the dead letter either.""" + + +def build_filesystem() -> S3FileSystem: + """S3 filesystem with the same bounds as the commit path. + + Unbounded here would defeat the point: this write happens while the read loop + is already stalled by a failed commit. + """ + + return S3FileSystem( + region=AWS_REGION, + connect_timeout=S3_CONNECT_TIMEOUT_SECONDS, + request_timeout=S3_REQUEST_TIMEOUT_SECONDS, + ) + + +def build_path( + record_type: str, run_id: str, now: datetime | None = None, root: str | None = None +) -> str: + """Where one dead-lettered batch lands. + + Partitioned by record type and then by day so a recovery can read one day of + one table rather than listing the whole prefix, and `dt=` spelled Hive-style + so Athena or a crawler can be pointed at it without moving anything. The + timestamp uses the repo's zero-padded format, so files sort chronologically; + `run_id` names the process that gave up. + + The random suffix is not decoration. The timestamp resolves only to the + second, and S3 has no "create if absent" -- two dead letters landing in the + same second would silently overwrite each other, destroying unrecovered data, + which is the one thing this module exists to prevent. + """ + + now = now or datetime.now(UTC) + root = root or DEAD_LETTER_ROOT + day = now.strftime("%Y-%m-%d") + timestamp = now.strftime(CREATED_AT_FORMAT) + return f"{root}/{record_type}/dt={day}/{timestamp}-{run_id}-{uuid4().hex[:8]}.parquet" + + +def build_table(record_type: str, rows: list[dict]) -> pa.Table: + """Arrow table from the declared schema, independent of the catalog.""" + + return pa.Table.from_pylist(rows, schema=RECORD_TYPE_TO_SCHEMA[record_type]) + + +def write_dead_letter( + record_type: str, + rows: list[dict], + run_id: str, + filesystem: FileSystem | None = None, + root: str | None = None, +) -> str: + """Persist a batch that could not be committed, and return where it went. + + Attempted twice, then given up on. Holding the rows in memory until S3 + recovers would convert a transient failure into unbounded buffer growth and + eventually a crash, which loses every buffered batch rather than this one. + """ + + filesystem = filesystem or build_filesystem() + path = build_path(record_type, run_id, root=root) + table = build_table(record_type, rows) + + for attempt in (1, 2): + try: + pq.write_table(table, path, filesystem=filesystem, compression="zstd") + except Exception: + logger.warning( + "dead letter write failed (attempt %d/2) for %d %s rows at %s", + attempt, + len(rows), + record_type, + path, + exc_info=True, + ) + else: + logger.error( + "DEAD LETTER: %d %s rows are not in Iceberg; written to %s", + len(rows), + record_type, + path, + ) + return path + + # Both the commit and its fallback failed, which in practice means S3 itself + # is unreachable. Nothing durable is left to try. + raise DeadLetterError( + f"dropped {len(rows)} {record_type} rows: dead letter write to {path} failed twice" + ) diff --git a/bluesky_ingestion_jetstream/aws/iceberg_writer.py b/bluesky_ingestion_jetstream/aws/iceberg_writer.py new file mode 100644 index 00000000..be48d136 --- /dev/null +++ b/bluesky_ingestion_jetstream/aws/iceberg_writer.py @@ -0,0 +1,37 @@ +"""The Iceberg append itself. + +One `append` is a whole commit: data files to S3, then a manifest, then a +manifest list, then a new `metadata.json`, then a Glue `UpdateTable` that swaps +the pointer. Only that last step makes anything visible, so a commit either lands +whole or not at all -- there is no partial table for a caller to repair. +""" + +import pyarrow as pa +from pyiceberg.table import Table + +from bluesky_ingestion_jetstream.aws.constants import SNAPSHOT_FLUSH_ID_TAG + + +def build_append_table(table: Table, rows: list[dict]) -> pa.Table: + """Build the table to append, using the schema the catalog assigned. + + Iceberg resolves columns by field id, so a declared schema whose ids have + drifted from the table's does not raise -- the affected columns just read + back as NULL. + """ + + return pa.Table.from_pylist(rows, schema=table.schema().as_arrow()) + + +def append_once(table: Table, append_table: pa.Table, flush_id: str) -> None: + table.append(append_table, snapshot_properties={SNAPSHOT_FLUSH_ID_TAG: flush_id}) + + +def already_committed(table: Table, flush_id: str) -> bool: + """Checks to see if a flush exists in the table's snapshot history.""" + + table.refresh() + return any( + snapshot.summary is not None and snapshot.summary.get(SNAPSHOT_FLUSH_ID_TAG) == flush_id + for snapshot in table.snapshots() + ) diff --git a/bluesky_ingestion_jetstream/aws/retry.py b/bluesky_ingestion_jetstream/aws/retry.py new file mode 100644 index 00000000..9cd38dd6 --- /dev/null +++ b/bluesky_ingestion_jetstream/aws/retry.py @@ -0,0 +1,71 @@ +"""Retry policy for Iceberg commits. + +Mirrors `data_platform/ingestion/bluesky_retry.py`: tenacity, an explicit +predicate for what counts as transient, and `reraise=True` so the caller sees the +original error rather than a `RetryError` wrapper. +""" + +import logging +from collections.abc import Callable +from typing import ParamSpec, TypeVar + +from botocore.exceptions import BotoCoreError, ClientError +from pyiceberg.exceptions import ( + CommitFailedException, + NoSuchTableError, + ServerError, +) +from tenacity import ( + before_sleep_log, + retry, + retry_if_exception, + stop_after_attempt, + wait_exponential_jitter, +) + +from bluesky_ingestion_jetstream.aws.constants import ( + COMMIT_INITIAL_DELAY_SECONDS, + COMMIT_MAX_ATTEMPTS, + COMMIT_MAX_DELAY_SECONDS, +) + +P = ParamSpec("P") +R = TypeVar("R") +logger = logging.getLogger(__name__) + +# A commit that failed for one of these reasons will fail again in a second's +# time. Retrying a schema mismatch just delays the dead letter by the length of +# the backoff, and hides a code bug behind two pointless round trips. +NON_RETRYABLE = (ValueError, TypeError, NoSuchTableError) + + +def is_retryable_commit_error(error: BaseException) -> bool: + """Whether re-issuing the same commit could plausibly succeed. + + Deliberately a denylist rather than an allowlist. The commit path spans + PyArrow's S3 client, botocore, and PyIceberg, and an unrecognised error from + that surface is far more likely to be a transport failure than a logic bug -- + so the default is to retry, and only the known-permanent cases opt out. + """ + + if isinstance(error, NON_RETRYABLE): + return False + return isinstance( + error, CommitFailedException | ServerError | ClientError | BotoCoreError | OSError + ) + + +def retry_iceberg_commit( + max_attempts: int = COMMIT_MAX_ATTEMPTS, + initial_delay: float = COMMIT_INITIAL_DELAY_SECONDS, + max_delay: float = COMMIT_MAX_DELAY_SECONDS, +) -> Callable[[Callable[P, R]], Callable[P, R]]: + """Retry a commit on transient failures, with jittered exponential backoff.""" + + return retry( + stop=stop_after_attempt(max_attempts), + wait=wait_exponential_jitter(initial=initial_delay, max=max_delay), + retry=retry_if_exception(is_retryable_commit_error), + before_sleep=before_sleep_log(logger, logging.WARNING), + reraise=True, + ) diff --git a/bluesky_ingestion_jetstream/constants.py b/bluesky_ingestion_jetstream/constants.py index 1986303c..085124b2 100644 --- a/bluesky_ingestion_jetstream/constants.py +++ b/bluesky_ingestion_jetstream/constants.py @@ -1,6 +1,6 @@ """Shared constants.""" -from pathlib import Path +from datetime import UTC, datetime, timedelta POSTS = "posts" LIKES = "likes" @@ -23,6 +23,14 @@ COMMON_REQUIRED_KEYS = ("uri", "did", "created_at", "ingested_at") +# `created_at` is the client's clock and the Iceberg partition key, so a row +# claiming 1970 mints a permanent one-file partition. Dropped, not clamped. +EARLIEST_VALID_CREATED_AT = datetime(2022, 1, 1, tzinfo=UTC) + +# How far ahead of `ingested_at` a `created_at` may be before the row is dropped. +# Compared against the broker's clock, not ours, so replays stay deterministic. +MAX_CREATED_AT_SKEW = timedelta(days=1) + POST_REQUIRED_KEYS = COMMON_REQUIRED_KEYS LIKE_REQUIRED_KEYS = (*COMMON_REQUIRED_KEYS, "subject_uri") REPOST_REQUIRED_KEYS = LIKE_REQUIRED_KEYS @@ -40,8 +48,6 @@ MAX_BUFFER_SIZE_BYTES = 2 * 1024 * 1024 * 1024 MAX_BUFFER_AGE_SECONDS = 30.0 -DATA_DIR = Path(__file__).parent / "data" - # Reconnect backoff, doubling from the first to the second. INITIAL_BACKOFF_SECONDS = 1.0 MAX_BACKOFF_SECONDS = 60.0 diff --git a/bluesky_ingestion_jetstream/event_parsing/shared.py b/bluesky_ingestion_jetstream/event_parsing/shared.py index 33bc0dec..cbd1471e 100644 --- a/bluesky_ingestion_jetstream/event_parsing/shared.py +++ b/bluesky_ingestion_jetstream/event_parsing/shared.py @@ -3,6 +3,11 @@ from collections.abc import Iterable from datetime import UTC, datetime, timedelta +from bluesky_ingestion_jetstream.constants import ( + EARLIEST_VALID_CREATED_AT, + MAX_CREATED_AT_SKEW, +) + EPOCH = datetime(1970, 1, 1, tzinfo=UTC) @@ -71,6 +76,16 @@ def parse_ingested_at(value: object) -> datetime | None: return None +def is_created_at_valid(created_at: datetime, ingested_at: datetime | None) -> bool: + """Whether a client-supplied `created_at` is plausible enough to partition on.""" + + if created_at < EARLIEST_VALID_CREATED_AT: + return False + if ingested_at is not None and created_at > ingested_at + MAX_CREATED_AT_SKEW: + return False + return True + + def parse_shared(event: dict) -> dict: """Extract the columns every commit type has.""" @@ -81,14 +96,22 @@ def parse_shared(event: dict) -> dict: collection = as_str(commit.get("collection")) rkey = as_str(commit.get("rkey")) + ingested_at = parse_ingested_at(event.get("time_us")) + created_at = parse_created_at(record.get("createdAt")) + + # An invalid timestamp is nulled, not flagged: `created_at` is a required + # key, so `validate_non_null_fields` then drops the row downstream. + if created_at is not None and not is_created_at_valid(created_at, ingested_at): + created_at = None + return { # Not on the wire: Jetstream sends the parts, so the AT-URI is rebuilt. "uri": f"at://{did}/{collection}/{rkey}" if did and collection and rkey else None, "did": did, "cid": as_str(commit.get("cid")), "rev": as_str(commit.get("rev")), - "created_at": parse_created_at(record.get("createdAt")), - "ingested_at": parse_ingested_at(event.get("time_us")), + "created_at": created_at, + "ingested_at": ingested_at, } diff --git a/bluesky_ingestion_jetstream/main.py b/bluesky_ingestion_jetstream/main.py index a4b8ab85..dc123570 100644 --- a/bluesky_ingestion_jetstream/main.py +++ b/bluesky_ingestion_jetstream/main.py @@ -1,15 +1,28 @@ -"""Entry point: stream from Jetstream, buffer, write to disk.""" +"""Entry point: stream from Jetstream, buffer, commit to Iceberg.""" import asyncio -from pathlib import Path +import logging +from uuid import uuid4 -from bluesky_ingestion_jetstream.constants import DATA_DIR +from bluesky_ingestion_jetstream.aws.catalog import build_catalog, load_tables from bluesky_ingestion_jetstream.network.connection import stream_events +from bluesky_ingestion_jetstream.sinks.base import Sink +from bluesky_ingestion_jetstream.sinks.iceberg import IcebergSink from bluesky_ingestion_jetstream.storage.buffer import BufferSet, flush +logger = logging.getLogger(__name__) -async def run(data_dir: Path) -> None: - """Consume the stream, buffering rows and writing them out when full.""" + +def new_run_id() -> str: + return str(uuid4()) + + +def build_sink(run_id: str) -> IcebergSink: + return IcebergSink(load_tables(build_catalog()), run_id) + + +async def run(sink: Sink) -> None: + """Consume the stream, buffering rows and committing them when full.""" buffers = BufferSet() @@ -17,13 +30,16 @@ async def run(data_dir: Path) -> None: buffers.add(record_type, row) if buffers.should_flush(): - flush(buffers, data_dir) + flush(buffers, sink) def main() -> None: """CLI entry point.""" - asyncio.run(run(DATA_DIR)) + logging.basicConfig(level=logging.INFO) + run_id = new_run_id() + logger.info("starting ingestion run %s", run_id) + asyncio.run(run(build_sink(run_id))) if __name__ == "__main__": diff --git a/bluesky_ingestion_jetstream/schemas/arrow_schemas.py b/bluesky_ingestion_jetstream/schemas/arrow_schemas.py index 675b88b4..989250fa 100644 --- a/bluesky_ingestion_jetstream/schemas/arrow_schemas.py +++ b/bluesky_ingestion_jetstream/schemas/arrow_schemas.py @@ -4,6 +4,12 @@ from bluesky_ingestion_jetstream.constants import FOLLOWS, LIKES, POSTS, REPOSTS +# Common fields that are stamped on by the writer at flush time rather than +# produced by the parsers. +WRITE_STAMPED_FIELDS = [ + pa.field("run_id", pa.string()), +] + # Present in all four tables. COMMON_FIELDS = [ pa.field("uri", pa.string()), @@ -12,6 +18,7 @@ pa.field("rev", pa.string()), pa.field("created_at", pa.timestamp("us", tz="UTC")), pa.field("ingested_at", pa.timestamp("us", tz="UTC")), + *WRITE_STAMPED_FIELDS, ] POST_SCHEMA: pa.Schema = pa.schema( diff --git a/bluesky_ingestion_jetstream/sinks/__init__.py b/bluesky_ingestion_jetstream/sinks/__init__.py new file mode 100644 index 00000000..283c37cd --- /dev/null +++ b/bluesky_ingestion_jetstream/sinks/__init__.py @@ -0,0 +1 @@ +"""Destinations a flush can be written to.""" diff --git a/bluesky_ingestion_jetstream/sinks/base.py b/bluesky_ingestion_jetstream/sinks/base.py new file mode 100644 index 00000000..f937feaf --- /dev/null +++ b/bluesky_ingestion_jetstream/sinks/base.py @@ -0,0 +1,17 @@ +"""The contract between the buffers and wherever their rows end up.""" + +from typing import Protocol + + +class Sink(Protocol): + """Somewhere a flush of one record type's rows can be written.""" + + def write(self, record_type: str, rows: list[dict]) -> None: + """Persist `rows`, or dispose of them durably if that proves impossible. + + Implementations must not raise for a batch they have dealt with -- a + dead-lettered batch is handled, not failed. Raising means the rows are + still the caller's problem, which for the ingester means the flush is + abandoned and the buffer keeps them. + """ + ... diff --git a/bluesky_ingestion_jetstream/sinks/iceberg.py b/bluesky_ingestion_jetstream/sinks/iceberg.py new file mode 100644 index 00000000..ee8af498 --- /dev/null +++ b/bluesky_ingestion_jetstream/sinks/iceberg.py @@ -0,0 +1,101 @@ +"""Commit flushed rows to Iceberg, or dead-letter them. + +The retry wraps the whole `append`, because the append *is* the S3 writes and the +Glue commit together -- there is no separate upload step to retry on its own, and +the Glue swap at the end is the only point where anything becomes visible. So a +retried attempt cannot double-write a table; it can only leave orphan files +behind from the attempt it abandoned. + +This runs synchronously inside the async read loop, so its worst case is time the +Jetstream socket spends undrained. That is the reason for three attempts rather +than ten, and for the client timeouts in `aws/constants.py`. +""" + +import logging +from collections.abc import Callable +from uuid import uuid4 + +from pyiceberg.table import Table + +from bluesky_ingestion_jetstream.aws.dead_letter import write_dead_letter +from bluesky_ingestion_jetstream.aws.iceberg_writer import ( + already_committed, + append_once, + build_append_table, +) +from bluesky_ingestion_jetstream.aws.retry import retry_iceberg_commit + +logger = logging.getLogger(__name__) + + +class IcebergSink: + """Writes each record type to its own Iceberg table. + + Holds the `run_id` because it is fixed for the life of the process; threading + it down through the flush signature would be churn for a value that cannot + vary per row. + """ + + def __init__( + self, + tables: dict[str, Table], + run_id: str, + dead_letter: Callable[..., str] = write_dead_letter, + ) -> None: + self.tables = tables + self.run_id = run_id + self.dead_letter = dead_letter + + def write(self, record_type: str, rows: list[dict]) -> None: + """Commit one record type's batch, dead-lettering it if that fails. + + Called once per record type per flush rather than once per flush, so a + table that is failing does not hold up the three that are not. + """ + + if not rows: + return + + # Stamped in place: a flush batch can be large, and copying every row to + # add one fixed key costs more than it explains. The buffer clears + # straight afterwards, so nothing else observes the mutation. + for row in rows: + row["run_id"] = self.run_id + + try: + self._commit(record_type, rows) + except Exception: + logger.warning( + "commit failed for %d %s rows; dead-lettering", + len(rows), + record_type, + exc_info=True, + ) + self.dead_letter(record_type, rows, self.run_id) + + def _commit(self, record_type: str, rows: list[dict]) -> None: + """Append with retries, skipping the work if a retry finds it already done.""" + + table = self.tables[record_type] + append_table = build_append_table(table, rows) + flush_id = str(uuid4()) + attempts = 0 + + @retry_iceberg_commit() + def commit() -> None: + nonlocal attempts + attempts += 1 + + # Only from the second attempt on: on the first there is nothing to + # have succeeded yet, and the check costs a Glue GetTable. + if attempts > 1 and already_committed(table, flush_id): + logger.warning( + "commit for %d %s rows had already landed; not repeating it", + len(rows), + record_type, + ) + return + + append_once(table, append_table, flush_id) + + commit() diff --git a/bluesky_ingestion_jetstream/storage/buffer.py b/bluesky_ingestion_jetstream/storage/buffer.py index 944820ca..26ef993f 100644 --- a/bluesky_ingestion_jetstream/storage/buffer.py +++ b/bluesky_ingestion_jetstream/storage/buffer.py @@ -3,14 +3,13 @@ import json import time from dataclasses import dataclass, field -from pathlib import Path from bluesky_ingestion_jetstream.constants import ( MAX_BUFFER_AGE_SECONDS, MAX_BUFFER_SIZE_BYTES, RECORD_TYPES, ) -from bluesky_ingestion_jetstream.writer import write +from bluesky_ingestion_jetstream.sinks.base import Sink def row_bytes(row: dict) -> int: @@ -102,15 +101,17 @@ def mark_flushed(self) -> None: self.last_flush = time.monotonic() -def flush(buffers: BufferSet, data_dir: Path) -> None: - """Write every non-empty buffer to disk and empty it. +def flush(buffers: BufferSet, sink: Sink) -> None: + """Write every non-empty buffer to the sink and empty it. - Each buffer is cleared only after its write succeeds; clearing first would - lose the batch if the write raised. + Each buffer is cleared only after its write returns; clearing first would + lose the batch if the write raised. A sink that dead-letters a batch has + dealt with it and returns normally, so those rows are cleared too -- keeping + them would write them twice on the next flush. """ for record_type, buffer in buffers.buffers.items(): if buffer.rows: - write(record_type, buffer.rows, data_dir) + sink.write(record_type, buffer.rows) buffer.clear() buffers.mark_flushed() diff --git a/bluesky_ingestion_jetstream/writer.py b/bluesky_ingestion_jetstream/writer.py deleted file mode 100644 index aeba53f5..00000000 --- a/bluesky_ingestion_jetstream/writer.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Write buffered events to disk.""" - -from datetime import UTC, datetime -from pathlib import Path - -import pyarrow as pa -import pyarrow.parquet as pq - -from bluesky_ingestion_jetstream.schemas.arrow_schemas import RECORD_TYPE_TO_SCHEMA -from lib.timestamp_utils import CREATED_AT_FORMAT - - -def build_path(record_type: str, data_dir: Path) -> Path: - """Timestamped Parquet path for one flush of one record type. - - The repo's timestamp format is zero-padded, so filenames sort chronologically. - It only resolves to the second, so a suffix is added rather than silently - overwriting a file from a flush in the same second. - """ - - directory = data_dir / record_type - directory.mkdir(parents=True, exist_ok=True) - - timestamp = datetime.now(UTC).strftime(CREATED_AT_FORMAT) - path = directory / f"{timestamp}.parquet" - seq = 1 - while path.exists(): - path = directory / f"{timestamp}-{seq}.parquet" - seq += 1 - return path - - -def write(record_type: str, rows: list[dict], data_dir: Path) -> Path: - """Write rows to a file under `data_dir` and return the path.""" - - table = pa.Table.from_pylist(rows, schema=RECORD_TYPE_TO_SCHEMA[record_type]) - path = build_path(record_type, data_dir) - pq.write_table(table, path, compression="snappy") - return path diff --git a/docs/runbooks/HOW_TO_SETUP_ICEBERG_TABLES.md b/docs/runbooks/HOW_TO_SETUP_ICEBERG_TABLES.md new file mode 100644 index 00000000..e755d32c --- /dev/null +++ b/docs/runbooks/HOW_TO_SETUP_ICEBERG_TABLES.md @@ -0,0 +1,104 @@ +# How to Set Up the Iceberg Tables + +## Overview + +This runbook covers creating the four Iceberg tables the Jetstream ingester writes to: `posts`, `likes`, `reposts`, and `follows`, in the Glue database `bluesky_raw`. + +The split of ownership matters: + +- **Terraform** ([`terraform/bluesky_ingestion_jetstream/main.tf`](../../terraform/bluesky_ingestion_jetstream/main.tf)) creates the S3 warehouse bucket and the Glue database — the containers. +- **[`bluesky_ingestion_jetstream/aws/bootstrap.py`](../../bluesky_ingestion_jetstream/aws/bootstrap.py)** creates the tables, by hand, once. + +The tables are deliberately not Terraform resources: Iceberg rewrites a table's schema, partition spec, and snapshot pointer on every commit, which an `aws_glue_catalog_table` resource would read as drift and revert on the next apply. + +The ingester itself never issues DDL. [`aws/catalog.py`](../../bluesky_ingestion_jetstream/aws/catalog.py) only loads tables and raises `MissingTablesError` if any are absent, so this bootstrap is a required step in any fresh environment. + +--- + +## Prerequisites + +- AWS credentials in the environment for **us-east-2**, with Glue (`CreateTable`, `GetTable`, `UpdateTable`) and S3 read/write on the warehouse bucket. +- Dependencies installed: `uv sync`. + +--- + +## Step 1: Apply Terraform + +From `terraform/bluesky_ingestion_jetstream/`: + +```bash +terraform init +terraform apply +``` + +This creates the `lab-data-integrations-interface` bucket and the `bluesky_raw` Glue database. Skip if they already exist. + +--- + +## Step 2: Run the bootstrap script + +From the repo root: + +```bash +uv run python -m bluesky_ingestion_jetstream.aws.bootstrap +``` + +Expected output on a fresh environment — one line per table, then the partition spec for each: + +``` +created bluesky_raw.posts -> s3://lab-data-integrations-interface/bluesky/raw/posts +created bluesky_raw.likes -> s3://lab-data-integrations-interface/bluesky/raw/likes +... +``` + +The script is idempotent. Tables that already exist print `exists` and are left untouched, so a partial failure is resolved by simply re-running. + +Each table is created from its Arrow schema in [`schemas/arrow_schemas.py`](../../bluesky_ingestion_jetstream/schemas/arrow_schemas.py), rooted at its own S3 prefix, partitioned by `day(created_at)`, with the properties in [`aws/constants.py`](../../bluesky_ingestion_jetstream/aws/constants.py). + +--- + +## Step 3: Verify + +Re-run the bootstrap. All four tables should report `exists`: + +```bash +uv run python -m bluesky_ingestion_jetstream.aws.bootstrap +``` + +Or ask Glue directly: + +```bash +aws glue get-tables --database-name bluesky_raw --region us-east-2 \ + --query 'TableList[].Name' +``` + +Then start the ingester — it loads all four tables before the first event, so a catalog problem surfaces in the first second rather than at the first flush: + +```bash +uv run python -m bluesky_ingestion_jetstream.main +``` + +--- + +## Changing a schema or table property later + +Bootstrap **only creates**. It will not alter a table that already exists, and Iceberg stores table properties in metadata at creation time — so editing `TABLE_PROPERTIES` or an Arrow schema and re-running does nothing to live tables. + +To actually change one: + +- **Additive schema change** (new column): evolve it in place with PyIceberg's `update_schema()`, or via Athena `ALTER TABLE`. Do not drop and recreate. +- **Property change**: `ALTER TABLE ... SET TBLPROPERTIES` in Athena. +- **Incompatible change**: drop and recreate, which discards the data. Bootstrap deliberately does not offer a drop — doing it by hand is the point. + +--- + +## Troubleshooting + +**`RuntimeError: Glue database 'bluesky_raw' does not exist`** +Terraform has not run, or ran against a different account/region. Go back to Step 1. + +**`MissingTablesError` when starting the ingester** +The database exists but tables do not. Run Step 2. The error names every missing table at once. + +**`TableAlreadyExistsError` surfacing as a crash** +Bootstrap catches this per table, so it should not escape. If it does, the table exists in Glue but is not loadable as an Iceberg table — most likely a non-Iceberg Glue table sitting on the same name. diff --git a/pyproject.toml b/pyproject.toml index 6207a3e5..6f07e9f7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,12 +14,17 @@ dependencies = [ "websockets>=15.0.1", "numpy>=2.4.4", "scikit-learn>=1.8.0", - "rich>=15.0.0", + # Capped below 15 because pyiceberg pins rich<15 for its CLI's table output. + # Nothing here needs 15: the highest floor in the tree is pydocket's 13.9.4. + "rich>=13.9.4,<15", "tenacity>=9.0.0", "spacy>=3.7.0", "pandas>=2.0.0", "duckdb>=1.0.0", "pyarrow>=14.0.0", + # pyiceberg-core supplies the Rust partition transforms; partitioned writes + # are unsupported without it. + "pyiceberg[glue,pyiceberg-core]>=0.10.0", "praw>=7.7.0", "prefect>=3.0.0", "pyyaml>=6.0.0", diff --git a/strategy_planning/2026-07-24_bluesky_event_schemas.md b/strategy_planning/2026-07-24_bluesky_event_schemas.md index 3e739027..28c897f9 100644 --- a/strategy_planning/2026-07-24_bluesky_event_schemas.md +++ b/strategy_planning/2026-07-24_bluesky_event_schemas.md @@ -143,6 +143,7 @@ Present in all four tables. Each table adds its own columns on top of these. | `rev` | `string` | `commit.rev` | Orders writes to one `uri`. Nothing to order today, since only creates are ingested — captured now because Jetstream's retention window makes it unbackfillable later. | | `created_at` | `timestamp[us, tz=UTC]` | `record.createdAt` | Client-supplied. Iceberg partition source, `day()` granularity. | | `ingested_at` | `timestamp[us, tz=UTC]` | `time_us` | Broker clock, so the trustworthy end of `ingested_at - created_at` ingest lag. Microseconds are held exactly. | +| `run_id` | `string` | Generated at process start | Identifies the ingestion process that wrote the row. Not on the wire, and not produced by the parsers — the writer stamps it. | `did` is derivable from `uri`, and is stored anyway for three reasons: it is the join and group key for nearly every query; deriving it means a string split on every row of every @@ -154,6 +155,19 @@ the whole firehose, so within any row group the DID range spans nearly the entir space and prunes nothing. Min/max pruning belongs to the timestamps, which are clustered by arrival. +**`run_id` is a column, not a filename, because filenames do not survive.** Iceberg names +its own data files (`00000-0-.parquet`), so per-run traceability cannot be encoded in +the path. Compaction then makes the question permanently unanswerable from the layout: a +`BIN_PACK` rewrite merges files by size with no regard for content, so one post-compaction +file spans many runs. As a column it survives that rewrite, Iceberg keeps min/max stats for +it per file, and `table.inspect.files()` gives the file↔run mapping directly. + +It is stamped by the writer rather than the parsers. The value is fixed for the life of a +process, so passing it down through every parser signature would be churn for something +that cannot vary per row. The consequence worth knowing is that the buffer's byte +accounting does not see it — `row_bytes` measures parser output, which is already +documented as a proxy rather than a measurement. + **Partitioning on `created_at` depends on clamping it.** A client-supplied timestamp is unbounded, so a single junk `"0001-01-01"` permanently mints a year-0001 partition, and honest backdating scatters tiny files across old partitions that never seal. The mitigation diff --git a/terraform/data_platform/.terraform.lock.hcl b/terraform/bluesky_ingestion_jetstream/.terraform.lock.hcl similarity index 100% rename from terraform/data_platform/.terraform.lock.hcl rename to terraform/bluesky_ingestion_jetstream/.terraform.lock.hcl diff --git a/terraform/bluesky_ingestion_jetstream/main.tf b/terraform/bluesky_ingestion_jetstream/main.tf new file mode 100644 index 00000000..e3a43a4e --- /dev/null +++ b/terraform/bluesky_ingestion_jetstream/main.tf @@ -0,0 +1,93 @@ +terraform { + required_providers { + aws = { + source = "hashicorp/aws" + version = "~> 5.0" + } + } +} + +provider "aws" { + region = var.aws_region +} + +# --------------------------------------------------------------------------- +# Variables +# +# These four values are duplicated in `bluesky_ingestion_jetstream/aws/constants.py`. +# Terraform creates the container; PyIceberg addresses it by name at runtime, so +# a change here is only half a change until that file matches. +# --------------------------------------------------------------------------- + +variable "aws_region" { + default = "us-east-2" +} + +variable "s3_bucket" { + default = "lab-data-integrations-interface" +} + +variable "s3_prefix" { + description = "Warehouse root. Each record type gets its own table directory beneath it." + default = "bluesky/raw" +} + +variable "glue_database" { + description = "Glue database names cannot contain `/`, so this is independent of s3_prefix." + default = "bluesky_raw" +} + +# --------------------------------------------------------------------------- +# S3 — the Iceberg warehouse +# +# Versioning is left off. Iceberg already keeps history through snapshots, and +# it rewrites metadata.json on every commit, so bucket versioning would retain a +# noncurrent version per commit — thousands a day — that nothing ever reads. +# --------------------------------------------------------------------------- + +resource "aws_s3_bucket" "warehouse" { + bucket = var.s3_bucket +} + +resource "aws_s3_bucket_public_access_block" "warehouse" { + bucket = aws_s3_bucket.warehouse.id + + block_public_acls = true + block_public_policy = true + ignore_public_acls = true + restrict_public_buckets = true +} + +# --------------------------------------------------------------------------- +# Glue catalog database +# +# The database only. The four Iceberg tables are created once by +# `python -m bluesky_ingestion_jetstream.aws.bootstrap` and are deliberately not +# Terraform resources: Iceberg rewrites a table's schema, partition spec, and +# snapshot pointer on every commit, which an `aws_glue_catalog_table` would read +# as drift and revert on the next apply. +# --------------------------------------------------------------------------- + +resource "aws_glue_catalog_database" "bluesky_raw" { + name = var.glue_database + + # Not read by the pipeline — `create_table` passes `location` explicitly — but + # it makes the catalog entry point at the same place the tables actually live. + location_uri = "s3://${aws_s3_bucket.warehouse.bucket}/${var.s3_prefix}" +} + +# --------------------------------------------------------------------------- +# Outputs +# --------------------------------------------------------------------------- + +output "s3_bucket_name" { + value = aws_s3_bucket.warehouse.bucket +} + +output "warehouse_uri" { + value = "s3://${aws_s3_bucket.warehouse.bucket}/${var.s3_prefix}" +} + +output "glue_database_name" { + value = aws_glue_catalog_database.bluesky_raw.name +} diff --git a/tests/bluesky_ingestion_jetstream/aws/__init__.py b/tests/bluesky_ingestion_jetstream/aws/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/bluesky_ingestion_jetstream/aws/test_constants.py b/tests/bluesky_ingestion_jetstream/aws/test_constants.py new file mode 100644 index 00000000..66fab411 --- /dev/null +++ b/tests/bluesky_ingestion_jetstream/aws/test_constants.py @@ -0,0 +1,35 @@ +"""Tests for the Glue/Iceberg configuration. + +Deliberately short. Most of what lives in `aws/constants.py` is a literal, and a +test that restates a literal only fails when the literal is changed on purpose -- +it reports edits, not defects. What is worth asserting is the handful of facts +that span two modules, or that AWS validates too late to be cheap. +""" + +from bluesky_ingestion_jetstream.aws.constants import ( + PARTITION_SOURCE_COLUMN, + TABLE_PROPERTIES, +) +from bluesky_ingestion_jetstream.schemas.arrow_schemas import RECORD_TYPE_TO_SCHEMA + + +def test_partition_source_column_exists_in_every_schema(): + """Renaming the column in the schemas has to fail here, not at bootstrap. + + `bootstrap.create_table` commits the table to Glue *before* `add_field` + resolves this column, so a mismatch leaves a half-built table behind for + someone to clean up by hand. + """ + + for schema in RECORD_TYPE_TO_SCHEMA.values(): + assert PARTITION_SOURCE_COLUMN in schema.names + + +def test_table_properties_are_strings(): + """Iceberg stores properties as strings, and rejects anything else at commit. + + Cheap to get wrong (`256 * 1024 * 1024` without the `str()`) and otherwise + only caught against live AWS. + """ + + assert all(isinstance(value, str) for value in TABLE_PROPERTIES.values()) diff --git a/tests/bluesky_ingestion_jetstream/aws/test_dead_letter.py b/tests/bluesky_ingestion_jetstream/aws/test_dead_letter.py new file mode 100644 index 00000000..84c3ee42 --- /dev/null +++ b/tests/bluesky_ingestion_jetstream/aws/test_dead_letter.py @@ -0,0 +1,174 @@ +"""Tests for the dead letter: where batches go when Iceberg will not take them.""" + +import logging +import re +from datetime import UTC, datetime + +import pyarrow.parquet as pq +import pytest +from pyarrow.fs import LocalFileSystem + +from bluesky_ingestion_jetstream.aws import dead_letter as dead_letter_module +from bluesky_ingestion_jetstream.aws.constants import DEAD_LETTER_ROOT, S3_BUCKET, S3_PREFIX +from bluesky_ingestion_jetstream.aws.dead_letter import ( + DeadLetterError, + build_path, + build_table, + write_dead_letter, +) +from bluesky_ingestion_jetstream.constants import RECORD_TYPES +from bluesky_ingestion_jetstream.schemas.arrow_schemas import RECORD_TYPE_TO_SCHEMA +from tests.bluesky_ingestion_jetstream.conftest import RUN_ID + +MOMENT = datetime(2026, 7, 23, 6, 48, 11, tzinfo=UTC) + + +@pytest.fixture +def local_root(tmp_path): + """A local stand-in for the S3 prefix, with the day directory pre-created. + + `pq.write_table` will not create parent directories, and S3 has none to + create, so the test supplies them rather than the code doing it needlessly. + """ + + day = datetime.now(UTC).strftime("%Y-%m-%d") + for record_type in RECORD_TYPES: + (tmp_path / record_type / f"dt={day}").mkdir(parents=True) + return str(tmp_path) + + +class TestBuildPath: + def test_stays_out_of_the_warehouse(self): + """Files under a table's location are exactly what orphan cleanup deletes.""" + + assert not build_path("posts", RUN_ID).startswith(f"{S3_BUCKET}/{S3_PREFIX}/") + assert DEAD_LETTER_ROOT.startswith(f"{S3_BUCKET}/dead_letter/") + + def test_partitions_by_record_type_then_day(self): + path = build_path("likes", RUN_ID, now=MOMENT, root="root") + + assert path.startswith("root/likes/dt=2026-07-23/") + + def test_names_carry_the_timestamp_and_run_id(self): + path = build_path("likes", RUN_ID, now=MOMENT, root="root") + + assert path.split("/")[-1].startswith(f"2026_07_23-06:48:11-{RUN_ID}-") + assert path.endswith(".parquet") + + def test_two_batches_in_one_second_do_not_collide(self): + """S3 has no create-if-absent; a collision would overwrite unrecovered data.""" + + first = build_path("likes", RUN_ID, now=MOMENT, root="root") + second = build_path("likes", RUN_ID, now=MOMENT, root="root") + + assert first != second + + def test_names_sort_chronologically(self): + """The repo timestamp format is zero-padded, so lexical order is time order.""" + + earlier = build_path("posts", RUN_ID, now=MOMENT, root="root") + later = build_path( + "posts", RUN_ID, now=datetime(2026, 7, 23, 11, 4, 9, tzinfo=UTC), root="root" + ) + + assert sorted([later, earlier]) == [earlier, later] + + +class TestBuildTable: + @pytest.mark.parametrize("record_type", RECORD_TYPES) + def test_uses_the_declared_schema_not_the_catalogs(self, record_type, rows_factory): + """The catalog may be exactly what is broken, so it must not be consulted.""" + + table = build_table(record_type, rows_factory(record_type, 3)) + + assert table.schema.equals(RECORD_TYPE_TO_SCHEMA[record_type]) + assert table.num_rows == 3 + + +class TestWriteDeadLetter: + @pytest.mark.parametrize("record_type", RECORD_TYPES) + def test_rows_survive_the_round_trip(self, record_type, rows_factory, local_root): + """Rows arrive already stamped, because the sink stamps before it commits.""" + + rows = [row | {"run_id": RUN_ID} for row in rows_factory(record_type, 4)] + + path = write_dead_letter( + record_type, rows, RUN_ID, filesystem=LocalFileSystem(), root=local_root + ) + + table = pq.read_table(path) + assert table.select(RECORD_TYPE_TO_SCHEMA[record_type].names).to_pylist() == rows + + def test_the_day_is_readable_as_a_partition_column(self, rows_factory, local_root): + """`dt=` is Hive-style so a reader recovers the day without opening the file.""" + + path = write_dead_letter( + "posts", rows_factory("posts", 2), RUN_ID, filesystem=LocalFileSystem(), root=local_root + ) + + assert "dt" in pq.read_table(path).column_names + + def test_returns_where_it_wrote(self, rows_factory, local_root): + path = write_dead_letter( + "posts", rows_factory("posts", 1), RUN_ID, filesystem=LocalFileSystem(), root=local_root + ) + + assert path.startswith(local_root) + assert re.search(rf"/posts/dt=\d{{4}}-\d{{2}}-\d{{2}}/.*{RUN_ID}.*\.parquet$", path) + + def test_a_transient_failure_is_retried_once(self, rows_factory, local_root, monkeypatch): + calls = [] + real = dead_letter_module.pq.write_table + + def flaky(*args, **kwargs): + calls.append(1) + if len(calls) == 1: + raise OSError("connection reset") + return real(*args, **kwargs) + + monkeypatch.setattr(dead_letter_module.pq, "write_table", flaky) + + write_dead_letter( + "likes", rows_factory("likes", 2), RUN_ID, filesystem=LocalFileSystem(), root=local_root + ) + + assert len(calls) == 2 + + def test_two_failures_raise_rather_than_retrying_forever( + self, rows_factory, local_root, monkeypatch + ): + """Holding rows until S3 recovers trades one lost batch for the whole buffer.""" + + calls = [] + + def always_fails(*args, **kwargs): + calls.append(1) + raise OSError("s3 unreachable") + + monkeypatch.setattr(dead_letter_module.pq, "write_table", always_fails) + + with pytest.raises(DeadLetterError, match="failed twice"): + write_dead_letter( + "posts", + rows_factory("posts", 3), + RUN_ID, + filesystem=LocalFileSystem(), + root=local_root, + ) + + assert len(calls) == 2 + + def test_a_successful_write_is_logged_loudly(self, rows_factory, local_root, caplog): + """With no replay tool yet, the log line is the only signal rows went missing.""" + + with caplog.at_level(logging.ERROR): + write_dead_letter( + "follows", + rows_factory("follows", 2), + RUN_ID, + filesystem=LocalFileSystem(), + root=local_root, + ) + + assert "DEAD LETTER" in caplog.text + assert "2 follows rows" in caplog.text diff --git a/tests/bluesky_ingestion_jetstream/conftest.py b/tests/bluesky_ingestion_jetstream/conftest.py index a6d464ea..1bb8dd43 100644 --- a/tests/bluesky_ingestion_jetstream/conftest.py +++ b/tests/bluesky_ingestion_jetstream/conftest.py @@ -13,10 +13,17 @@ CREATED_AT_STR = "2026-07-23T06:48:11.102Z" CREATED_AT = datetime(2026, 7, 23, 6, 48, 11, 102000, tzinfo=UTC) -# Deliberately earlier than CREATED_AT: the two clocks are independent, and a test -# that shared one instant could not catch them being swapped. -TIME_US = 1784533137411372 -INGESTED_AT = datetime(2026, 7, 20, 7, 38, 57, 411372, tzinfo=UTC) +# A distinct instant from CREATED_AT -- the two clocks are independent, and a test +# that shared one instant could not catch them being swapped. Shortly *after* it, +# rather than before: a create reaches the firehose after it is made, and +# `is_created_at_valid` now rejects a `created_at` more than +# MAX_CREATED_AT_SKEW ahead of the broker's clock, so a fixture ordered the other +# way would be dropped by every parser test that uses it. +TIME_US = 1784789293411372 +INGESTED_AT = datetime(2026, 7, 23, 6, 48, 13, 411372, tzinfo=UTC) + +# Fixed, so a test can assert the stamped value rather than merely its presence. +RUN_ID = "f47ac10b-58cc-4372-a567-0e02b2c3d479" SUBJECT_DID = "did:plc:targetaccount0000000000" SUBJECT_URI = "at://did:plc:abc/app.bsky.feed.post/3l3qtarget" @@ -123,6 +130,17 @@ def as_messages(events: list) -> list[str]: return [json.dumps(event) for event in events] +class MemorySink: + """A `Sink` that collects writes in a list.""" + + def __init__(self) -> None: + self.writes: list[tuple[str, list[dict]]] = [] + + def write(self, record_type: str, rows: list[dict]) -> None: + # Copied, so a later `Buffer.clear()` cannot empty what was recorded. + self.writes.append((record_type, list(rows))) + + @pytest.fixture def post_event() -> dict: return make_event(POST_COLLECTION, post_record()) diff --git a/tests/bluesky_ingestion_jetstream/event_parsing/test_shared.py b/tests/bluesky_ingestion_jetstream/event_parsing/test_shared.py index 7e5a86bf..d79aa1ab 100644 --- a/tests/bluesky_ingestion_jetstream/event_parsing/test_shared.py +++ b/tests/bluesky_ingestion_jetstream/event_parsing/test_shared.py @@ -4,6 +4,11 @@ import pytest +from bluesky_ingestion_jetstream.constants import ( + COMMON_REQUIRED_KEYS, + EARLIEST_VALID_CREATED_AT, + MAX_CREATED_AT_SKEW, +) from bluesky_ingestion_jetstream.event_parsing.shared import ( as_dict, as_str, @@ -139,6 +144,70 @@ def test_out_of_range_becomes_none(self): assert parse_ingested_at(10**20) is None +class TestCreatedAtRange: + """`created_at` is client-supplied and is the Iceberg partition key. + + An out-of-range value is nulled by `parse_shared`, which makes the row fail + the required-key check and be dropped. These assert on `parse_shared` rather + than on `is_created_at_valid` alone, because the nulling is the part that + actually keeps junk out of the table. + """ + + def test_a_plausible_timestamp_survives(self, post_event): + assert parse_shared(post_event)["created_at"] == CREATED_AT + + @pytest.mark.parametrize("createdAt", ["1970-01-01T00:00:00Z", "2021-12-31T23:59:59Z"]) + def test_timestamps_before_the_floor_are_nulled(self, post_event, createdAt): + post_event["commit"]["record"]["createdAt"] = createdAt + + assert parse_shared(post_event)["created_at"] is None + + def test_the_floor_itself_is_accepted(self, post_event): + """A boundary that rejected its own limit would be off by one day of data.""" + + post_event["commit"]["record"]["createdAt"] = EARLIEST_VALID_CREATED_AT.isoformat() + + assert parse_shared(post_event)["created_at"] == EARLIEST_VALID_CREATED_AT + + def test_timestamps_far_ahead_of_the_broker_are_nulled(self, post_event): + far_future = INGESTED_AT + MAX_CREATED_AT_SKEW + timedelta(seconds=1) + post_event["commit"]["record"]["createdAt"] = far_future.isoformat() + + assert parse_shared(post_event)["created_at"] is None + + def test_a_clock_within_the_skew_allowance_is_kept(self, post_event): + """A misconfigured device clock is not junk, and dropping it loses real posts.""" + + ahead = INGESTED_AT + MAX_CREATED_AT_SKEW - timedelta(seconds=1) + post_event["commit"]["record"]["createdAt"] = ahead.isoformat() + + assert parse_shared(post_event)["created_at"] == ahead + + def test_the_ceiling_follows_the_broker_clock_not_the_wall_clock(self, post_event): + """Replay redelivers old events to a much later wall clock; they must survive.""" + + post_event["time_us"] = TIME_US + post_event["commit"]["record"]["createdAt"] = CREATED_AT.isoformat() + + assert parse_shared(post_event)["created_at"] == CREATED_AT + + def test_a_null_ingested_at_leaves_only_the_floor(self, post_event): + """With no broker clock the ceiling cannot be applied -- the row dies anyway.""" + + del post_event["time_us"] + post_event["commit"]["record"]["createdAt"] = "2099-01-01T00:00:00Z" + row = parse_shared(post_event) + + assert row["created_at"] == datetime(2099, 1, 1, tzinfo=UTC) + assert not validate_non_null_fields(row, ["ingested_at"]) + + def test_an_out_of_range_row_is_dropped_by_the_required_key_check(self, post_event): + post_event["commit"]["record"]["createdAt"] = "1999-01-01T00:00:00Z" + row = parse_shared(post_event) + + assert not validate_non_null_fields(row, COMMON_REQUIRED_KEYS) + + class TestParseShared: def test_extracts_every_common_column(self, post_event): row = parse_shared(post_event) diff --git a/tests/bluesky_ingestion_jetstream/network/test_connection.py b/tests/bluesky_ingestion_jetstream/network/test_connection.py index f30728f5..c742d797 100644 --- a/tests/bluesky_ingestion_jetstream/network/test_connection.py +++ b/tests/bluesky_ingestion_jetstream/network/test_connection.py @@ -195,6 +195,18 @@ def test_unparseable_created_at_drops_the_row(self): assert process_commit_event(make_event(POST_COLLECTION, record)) is None + @pytest.mark.parametrize("time_us", [None, "1725911162329308", True]) + def test_unusable_time_us_drops_the_row(self, time_us): + """`ingested_at` is derived from `time_us`, so junk there drops the row. + + It is the one required column the broker supplies rather than the client, + so a null here means a malformed envelope rather than a careless poster. + """ + + event = make_event(POST_COLLECTION, post_record(), time_us=time_us) + + assert process_commit_event(event) is None + def test_like_without_subject_uri_drops_the_row(self): record = interaction_record(subject={"cid": "bafyx"}) diff --git a/tests/bluesky_ingestion_jetstream/schemas/test_arrow_schemas.py b/tests/bluesky_ingestion_jetstream/schemas/test_arrow_schemas.py index f3651b83..9daac427 100644 --- a/tests/bluesky_ingestion_jetstream/schemas/test_arrow_schemas.py +++ b/tests/bluesky_ingestion_jetstream/schemas/test_arrow_schemas.py @@ -10,9 +10,11 @@ POST_SCHEMA, RECORD_TYPE_TO_SCHEMA, REPOST_SCHEMA, + WRITE_STAMPED_FIELDS, ) -COMMON_COLUMNS = {"uri", "did", "cid", "rev", "created_at", "ingested_at"} +WRITE_STAMPED_COLUMNS = {field.name for field in WRITE_STAMPED_FIELDS} +COMMON_COLUMNS = {"uri", "did", "cid", "rev", "created_at", "ingested_at"} | WRITE_STAMPED_COLUMNS class TestRecordTypeToSchema: @@ -95,8 +97,20 @@ def test_subject_did_is_a_string(self): class TestSchemasMatchParsedRows: @pytest.mark.parametrize("record_type", RECORD_TYPES) def test_parsed_rows_have_exactly_the_schema_columns(self, record_type, rows_factory): - """A drifted column would silently null out or fail the Parquet write.""" + """A drifted column would silently null out or fail the Parquet write. + + The write-stamped columns are the one legitimate gap: the writer adds them, + so a parser that produced them would be the bug. + """ row = rows_factory(record_type, 1)[0] - assert set(row) == set(RECORD_TYPE_TO_SCHEMA[record_type].names) + expected = set(RECORD_TYPE_TO_SCHEMA[record_type].names) - WRITE_STAMPED_COLUMNS + + assert set(row) == expected + + @pytest.mark.parametrize("record_type", RECORD_TYPES) + def test_write_stamped_columns_are_in_every_schema(self, record_type): + """They are absent from parser output, so nothing else would catch a typo.""" + + assert WRITE_STAMPED_COLUMNS.issubset(set(RECORD_TYPE_TO_SCHEMA[record_type].names)) diff --git a/tests/bluesky_ingestion_jetstream/sinks/__init__.py b/tests/bluesky_ingestion_jetstream/sinks/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/bluesky_ingestion_jetstream/sinks/test_iceberg.py b/tests/bluesky_ingestion_jetstream/sinks/test_iceberg.py new file mode 100644 index 00000000..8db21983 --- /dev/null +++ b/tests/bluesky_ingestion_jetstream/sinks/test_iceberg.py @@ -0,0 +1,261 @@ +"""Tests for the commit path: retries, the idempotency guard, and dead-lettering. + +Every test runs against a fake table. Nothing here touches AWS, and the backoff +sleeps are patched out, so the suite stays sub-second despite exercising a retry +policy measured in seconds. +""" + +import logging + +import pyarrow as pa +import pytest +from pyiceberg.exceptions import CommitFailedException, NoSuchTableError + +from bluesky_ingestion_jetstream.aws.constants import COMMIT_MAX_ATTEMPTS, SNAPSHOT_FLUSH_ID_TAG +from bluesky_ingestion_jetstream.constants import RECORD_TYPES +from bluesky_ingestion_jetstream.schemas.arrow_schemas import RECORD_TYPE_TO_SCHEMA +from bluesky_ingestion_jetstream.sinks.iceberg import IcebergSink +from tests.bluesky_ingestion_jetstream.conftest import RUN_ID + + +class FakeSnapshot: + def __init__(self, summary: dict): + self.summary = summary + + +class FakeTable: + """A table that records appends and can be told to fail the first N of them. + + `committed_despite_failing` models the case a retry cannot otherwise see: the + Glue update landed but the response was lost, so the caller observes an error + for a commit that actually succeeded. + """ + + def __init__(self, record_type: str, fail_times: int = 0, error: Exception | None = None): + self._schema = RECORD_TYPE_TO_SCHEMA[record_type] + self.appends: list[pa.Table] = [] + self.flush_ids: list[str] = [] + self.fail_times = fail_times + self.error = error or CommitFailedException("glue said no") + self.committed_despite_failing = False + self.refreshes = 0 + self._snapshots: list[FakeSnapshot] = [] + + def schema(self): + class Schema: + def __init__(self, arrow): + self._arrow = arrow + + def as_arrow(self): + return self._arrow + + return Schema(self._schema) + + def append(self, arrow: pa.Table, snapshot_properties: dict | None = None) -> None: + flush_id = (snapshot_properties or {}).get(SNAPSHOT_FLUSH_ID_TAG, "") + if self.fail_times > 0: + self.fail_times -= 1 + if self.committed_despite_failing: + self._snapshots.append(FakeSnapshot({SNAPSHOT_FLUSH_ID_TAG: flush_id})) + raise self.error + self.appends.append(arrow) + self.flush_ids.append(flush_id) + self._snapshots.append(FakeSnapshot({SNAPSHOT_FLUSH_ID_TAG: flush_id})) + + def refresh(self) -> None: + self.refreshes += 1 + + def snapshots(self): + return self._snapshots + + +@pytest.fixture(autouse=True) +def no_sleeping(monkeypatch): + """Run the real backoff schedule without waiting for it.""" + + monkeypatch.setattr("tenacity.nap.time.sleep", lambda _seconds: None) + + +class RecordingDeadLetter: + """Stands in for the S3 write, recording what was given up on.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, list[dict], str]] = [] + + def __call__(self, record_type: str, rows: list[dict], run_id: str, **kwargs) -> str: + self.calls.append((record_type, list(rows), run_id)) + return "s3://dead-letter/file.parquet" + + +@pytest.fixture +def recorded_dead_letters() -> RecordingDeadLetter: + return RecordingDeadLetter() + + +def build_sink(tables: dict, dead_letter) -> IcebergSink: + return IcebergSink(tables, RUN_ID, dead_letter=dead_letter) + + +class TestSuccessfulCommit: + @pytest.mark.parametrize("record_type", RECORD_TYPES) + def test_rows_reach_the_table(self, record_type, rows_factory, recorded_dead_letters): + table = FakeTable(record_type) + rows = rows_factory(record_type, 3) + + build_sink({record_type: table}, recorded_dead_letters).write(record_type, rows) + + assert len(table.appends) == 1 + assert table.appends[0].num_rows == 3 + assert recorded_dead_letters.calls == [] + + def test_every_row_is_stamped_with_the_run_id(self, rows_factory, recorded_dead_letters): + table = FakeTable("likes") + + build_sink({"likes": table}, recorded_dead_letters).write("likes", rows_factory("likes", 2)) + + assert table.appends[0].column("run_id").to_pylist() == [RUN_ID, RUN_ID] + + def test_arrow_is_built_from_the_tables_schema(self, rows_factory, recorded_dead_letters): + """Using the declared schema would write ids the table does not know. + + Iceberg matches columns by field id, so the mismatch does not raise -- the + affected columns simply read back as NULL. + """ + + table = FakeTable("posts") + + build_sink({"posts": table}, recorded_dead_letters).write("posts", rows_factory("posts", 1)) + + assert table.appends[0].schema.equals(table.schema().as_arrow()) + + def test_the_commit_is_tagged_with_a_flush_id(self, rows_factory, recorded_dead_letters): + table = FakeTable("posts") + + build_sink({"posts": table}, recorded_dead_letters).write("posts", rows_factory("posts", 1)) + + assert table.flush_ids[0] + + def test_each_flush_gets_its_own_id(self, rows_factory, recorded_dead_letters): + table = FakeTable("posts") + sink = build_sink({"posts": table}, recorded_dead_letters) + + sink.write("posts", rows_factory("posts", 1)) + sink.write("posts", rows_factory("posts", 1)) + + assert table.flush_ids[0] != table.flush_ids[1] + + def test_an_empty_batch_is_not_committed(self, recorded_dead_letters): + """An empty append would burn a full commit -- and a snapshot -- on nothing.""" + + table = FakeTable("posts") + + build_sink({"posts": table}, recorded_dead_letters).write("posts", []) + + assert table.appends == [] + + +class TestRetry: + def test_a_transient_failure_is_retried_and_can_succeed( + self, rows_factory, recorded_dead_letters + ): + table = FakeTable("posts", fail_times=2) + + build_sink({"posts": table}, recorded_dead_letters).write("posts", rows_factory("posts", 2)) + + assert len(table.appends) == 1 + assert recorded_dead_letters.calls == [] + + def test_it_gives_up_after_the_configured_attempts(self, rows_factory, recorded_dead_letters): + """Three attempts, because the read loop is stalled for every one of them.""" + + table = FakeTable("posts", fail_times=99) + rows = rows_factory("posts", 2) + + build_sink({"posts": table}, recorded_dead_letters).write("posts", rows) + + assert len(recorded_dead_letters.calls) == 1 + record_type, dead_rows, run_id = recorded_dead_letters.calls[0] + assert (record_type, len(dead_rows), run_id) == ("posts", 2, RUN_ID) + + def test_a_code_bug_is_not_retried(self, rows_factory, recorded_dead_letters): + """A schema mismatch fails identically three times; retrying only delays it.""" + + table = FakeTable("posts", fail_times=99, error=ValueError("schema mismatch")) + + build_sink({"posts": table}, recorded_dead_letters).write("posts", rows_factory("posts", 1)) + + assert table.refreshes == 0 + assert len(recorded_dead_letters.calls) == 1 + + def test_a_missing_table_is_not_retried(self, rows_factory, recorded_dead_letters): + table = FakeTable("posts", fail_times=99, error=NoSuchTableError("gone")) + + build_sink({"posts": table}, recorded_dead_letters).write("posts", rows_factory("posts", 1)) + + assert table.refreshes == 0 + + def test_a_lost_response_does_not_duplicate_the_rows( + self, rows_factory, recorded_dead_letters, caplog + ): + """The commit landed but the caller saw an error. Iceberg would not dedupe it.""" + + table = FakeTable("posts", fail_times=1) + table.committed_despite_failing = True + + with caplog.at_level(logging.WARNING): + build_sink({"posts": table}, recorded_dead_letters).write( + "posts", rows_factory("posts", 2) + ) + + assert table.appends == [] + assert table.refreshes == 1 + assert recorded_dead_letters.calls == [] + assert "already landed" in caplog.text + + def test_the_first_attempt_does_not_pay_for_the_check( + self, rows_factory, recorded_dead_letters + ): + """`already_committed` is a Glue GetTable; nothing can have landed yet.""" + + table = FakeTable("posts") + + build_sink({"posts": table}, recorded_dead_letters).write("posts", rows_factory("posts", 1)) + + assert table.refreshes == 0 + + def test_attempts_are_bounded_by_the_constant(self, rows_factory, recorded_dead_letters): + table = FakeTable("posts", fail_times=99) + start = table.fail_times + + build_sink({"posts": table}, recorded_dead_letters).write("posts", rows_factory("posts", 1)) + + assert start - table.fail_times == COMMIT_MAX_ATTEMPTS + + +class TestPerRecordTypeIsolation: + def test_one_failing_table_does_not_block_the_others(self, rows_factory, recorded_dead_letters): + """Four separate commits, so a throttled table cannot cost the other three.""" + + tables = {record_type: FakeTable(record_type) for record_type in RECORD_TYPES} + tables["likes"] = FakeTable("likes", fail_times=99) + sink = build_sink(tables, recorded_dead_letters) + + for record_type in RECORD_TYPES: + sink.write(record_type, rows_factory(record_type, 1)) + + assert [rt for rt, _, _ in recorded_dead_letters.calls] == ["likes"] + for record_type in ("posts", "reposts", "follows"): + assert len(tables[record_type].appends) == 1 + + def test_a_dead_letter_failure_propagates(self, rows_factory): + """Nothing durable is left, so the caller must not be told this was handled.""" + + def exploding_dead_letter(*args, **kwargs): + raise RuntimeError("s3 unreachable") + + table = FakeTable("posts", fail_times=99) + + with pytest.raises(RuntimeError, match="s3 unreachable"): + build_sink({"posts": table}, exploding_dead_letter).write( + "posts", rows_factory("posts", 1) + ) diff --git a/tests/bluesky_ingestion_jetstream/storage/test_buffer.py b/tests/bluesky_ingestion_jetstream/storage/test_buffer.py index 376e38f6..cf21ef15 100644 --- a/tests/bluesky_ingestion_jetstream/storage/test_buffer.py +++ b/tests/bluesky_ingestion_jetstream/storage/test_buffer.py @@ -9,15 +9,24 @@ from bluesky_ingestion_jetstream.storage.buffer import Buffer, BufferSet, flush, row_bytes +class RecordingSink: + """A sink that remembers what it was handed, and can be told to fail.""" + + def __init__(self, error: Exception | None = None): + self.calls: list[tuple[str, list[dict]]] = [] + self.error = error + + def write(self, record_type: str, rows: list[dict]) -> None: + if self.error is not None: + raise self.error + self.calls.append((record_type, list(rows))) + + @pytest.fixture -def recorded_writes(monkeypatch, tmp_path): - """Replace the writer so flush tests never touch Parquet.""" +def sink(): + """The flush tests need a destination, not a filesystem.""" - calls: list[tuple[str, int]] = [] - monkeypatch.setattr( - buffer_module, "write", lambda rt, rows, data_dir: calls.append((rt, len(rows))) - ) - return calls + return RecordingSink() @pytest.fixture @@ -225,74 +234,71 @@ def test_restarts_the_age_timer(self, rows_factory, monkeypatch): class TestFlush: - def test_writes_every_non_empty_buffer(self, filled, recorded_writes, tmp_path): - flush(filled, tmp_path) + def test_writes_every_non_empty_buffer(self, filled, sink): + flush(filled, sink) - assert dict(recorded_writes) == dict(zip(RECORD_TYPES, [1, 2, 3, 4])) + assert {rt: len(rows) for rt, rows in sink.calls} == dict(zip(RECORD_TYPES, [1, 2, 3, 4])) - def test_empty_buffers_write_nothing(self, rows_factory, recorded_writes, tmp_path): + def test_empty_buffers_write_nothing(self, rows_factory, sink): buffer_set = BufferSet() buffer_set.add("posts", rows_factory("posts", 1)[0]) - flush(buffer_set, tmp_path) + flush(buffer_set, sink) - assert [record_type for record_type, _ in recorded_writes] == ["posts"] + assert [record_type for record_type, _ in sink.calls] == ["posts"] - def test_nothing_buffered_writes_nothing(self, recorded_writes, tmp_path): - flush(BufferSet(), tmp_path) + def test_nothing_buffered_writes_nothing(self, sink): + flush(BufferSet(), sink) - assert recorded_writes == [] + assert sink.calls == [] - def test_buffers_are_empty_afterward(self, filled, recorded_writes, tmp_path): - flush(filled, tmp_path) + def test_buffers_are_empty_afterward(self, filled, sink): + flush(filled, sink) assert filled.size == 0 for buffer in filled.buffers.values(): assert buffer.rows == [] - def test_restarts_the_age_timer(self, filled, recorded_writes, tmp_path, monkeypatch): + def test_restarts_the_age_timer(self, filled, sink, monkeypatch): clock = [500.0] monkeypatch.setattr(buffer_module.time, "monotonic", lambda: clock[0]) - flush(filled, tmp_path) + flush(filled, sink) assert filled.last_flush == 500.0 - def test_timer_restarts_with_nothing_to_write(self, recorded_writes, tmp_path, monkeypatch): + def test_timer_restarts_with_nothing_to_write(self, sink, monkeypatch): """A size-triggered flush must not leave a stale timer that fires next tick.""" clock = [500.0] monkeypatch.setattr(buffer_module.time, "monotonic", lambda: clock[0]) buffer_set = BufferSet() - flush(buffer_set, tmp_path) + flush(buffer_set, sink) assert buffer_set.last_flush == 500.0 - def test_rows_survive_a_write_failure(self, filled, monkeypatch, tmp_path): - """Clearing before the write succeeds would lose the batch.""" + def test_rows_survive_a_write_failure(self, filled): + """Clearing before the write returns would lose the batch. - def boom(record_type, rows, data_dir): - raise OSError("disk full") + A sink that dead-letters has handled the batch and returns normally, so + this is the case where even that failed and the rows are still ours. + """ - monkeypatch.setattr(buffer_module, "write", boom) + failing = RecordingSink(error=OSError("s3 unreachable")) expected = {rt: len(b.rows) for rt, b in filled.buffers.items()} - with pytest.raises(OSError, match="disk full"): - flush(filled, tmp_path) + with pytest.raises(OSError, match="s3 unreachable"): + flush(filled, failing) assert {rt: len(b.rows) for rt, b in filled.buffers.items()} == expected - def test_write_receives_the_rows_it_should(self, rows_factory, monkeypatch, tmp_path): - seen: list[list[dict]] = [] - monkeypatch.setattr( - buffer_module, "write", lambda rt, rows, data_dir: seen.append(list(rows)) - ) + def test_the_sink_receives_the_rows_it_should(self, rows_factory, sink): rows = rows_factory("follows", 2) buffer_set = BufferSet() for row in rows: buffer_set.add("follows", row) - flush(buffer_set, tmp_path) + flush(buffer_set, sink) - assert seen == [rows] + assert sink.calls == [("follows", rows)] diff --git a/tests/bluesky_ingestion_jetstream/test_main.py b/tests/bluesky_ingestion_jetstream/test_main.py index 96e9aa77..814c219c 100644 --- a/tests/bluesky_ingestion_jetstream/test_main.py +++ b/tests/bluesky_ingestion_jetstream/test_main.py @@ -7,6 +7,7 @@ from bluesky_ingestion_jetstream import main as main_module from bluesky_ingestion_jetstream.constants import RECORD_TYPES from bluesky_ingestion_jetstream.main import run +from tests.bluesky_ingestion_jetstream.conftest import MemorySink @pytest.fixture @@ -15,7 +16,7 @@ def wired(monkeypatch, rows_factory): flushes: list[dict[str, int]] = [] - def fake_flush(buffers, data_dir): + def fake_flush(buffers, sink): flushes.append({rt: len(b.rows) for rt, b in buffers.buffers.items() if b.rows}) for buffer in buffers.buffers.values(): buffer.clear() @@ -34,22 +35,29 @@ async def fake_stream(): return drive +@pytest.fixture +def sink() -> MemorySink: + """`run` needs a destination; these tests stub `flush`, so it is never used.""" + + return MemorySink() + + def rows_for(rows_factory, record_type, count): return [(record_type, row) for row in rows_factory(record_type, count)] class TestRun: - def test_consumes_the_whole_stream(self, wired, rows_factory, tmp_path, monkeypatch): + def test_consumes_the_whole_stream(self, wired, rows_factory, sink, monkeypatch): flushes = wired(rows_for(rows_factory, "likes", 5)) monkeypatch.setattr( main_module.BufferSet, "should_flush", lambda self: False, raising=False ) - asyncio.run(run(tmp_path)) + asyncio.run(run(sink)) assert flushes == [] - def test_flushes_when_the_buffers_say_so(self, wired, rows_factory, tmp_path, monkeypatch): + def test_flushes_when_the_buffers_say_so(self, wired, rows_factory, sink, monkeypatch): flushes = wired(rows_for(rows_factory, "likes", 3)) calls = {"n": 0} @@ -59,11 +67,11 @@ def every_other(self): monkeypatch.setattr(main_module.BufferSet, "should_flush", every_other, raising=False) - asyncio.run(run(tmp_path)) + asyncio.run(run(sink)) assert flushes == [{"likes": 2}] - def test_routes_each_row_to_its_record_type(self, wired, rows_factory, tmp_path, monkeypatch): + def test_routes_each_row_to_its_record_type(self, wired, rows_factory, sink, monkeypatch): stream = [pair for rt in RECORD_TYPES for pair in rows_for(rows_factory, rt, 2)] flushes = wired(stream) monkeypatch.setattr( @@ -73,27 +81,27 @@ def test_routes_each_row_to_its_record_type(self, wired, rows_factory, tmp_path, raising=False, ) - asyncio.run(run(tmp_path)) + asyncio.run(run(sink)) assert flushes == [dict.fromkeys(RECORD_TYPES, 2)] - def test_empty_stream_never_flushes(self, wired, tmp_path): + def test_empty_stream_never_flushes(self, wired, sink): flushes = wired([]) - asyncio.run(run(tmp_path)) + asyncio.run(run(sink)) assert flushes == [] - def test_uses_the_real_thresholds_by_default(self, wired, rows_factory, tmp_path): + def test_uses_the_real_thresholds_by_default(self, wired, rows_factory, sink): """A handful of rows is nowhere near the size threshold.""" flushes = wired(rows_for(rows_factory, "likes", 10)) - asyncio.run(run(tmp_path)) + asyncio.run(run(sink)) assert flushes == [] - def test_passes_the_data_dir_through(self, monkeypatch, rows_factory, tmp_path): + def test_passes_the_sink_through_to_flush(self, monkeypatch, rows_factory, sink): seen: list = [] async def fake_stream(): @@ -101,9 +109,16 @@ async def fake_stream(): yield parsed monkeypatch.setattr(main_module, "stream_events", fake_stream) - monkeypatch.setattr(main_module, "flush", lambda buffers, data_dir: seen.append(data_dir)) + monkeypatch.setattr(main_module, "flush", lambda buffers, s: seen.append(s)) monkeypatch.setattr(main_module.BufferSet, "should_flush", lambda self: True, raising=False) - asyncio.run(run(tmp_path)) + asyncio.run(run(sink)) + + assert seen == [sink] + + +class TestNewRunId: + def test_is_a_distinct_value_each_call(self): + """Two processes must not share a run id, or the column cannot separate them.""" - assert seen == [tmp_path] + assert main_module.new_run_id() != main_module.new_run_id() diff --git a/tests/bluesky_ingestion_jetstream/test_writer.py b/tests/bluesky_ingestion_jetstream/test_writer.py deleted file mode 100644 index 55863940..00000000 --- a/tests/bluesky_ingestion_jetstream/test_writer.py +++ /dev/null @@ -1,179 +0,0 @@ -"""Tests for the Parquet writer and its path building.""" - -from datetime import UTC, datetime - -import pyarrow.parquet as pq -import pytest - -from bluesky_ingestion_jetstream.constants import RECORD_TYPES -from bluesky_ingestion_jetstream.schemas.arrow_schemas import RECORD_TYPE_TO_SCHEMA -from bluesky_ingestion_jetstream.writer import build_path, write - - -@pytest.fixture -def frozen_clock(monkeypatch): - """Pin the timestamp so filenames are predictable.""" - - from bluesky_ingestion_jetstream import writer - - class FixedDatetime(datetime): - @classmethod - def now(cls, tz=None): - return datetime(2026, 7, 23, 6, 48, 11, tzinfo=tz or UTC) - - monkeypatch.setattr(writer, "datetime", FixedDatetime) - - -class TestBuildPath: - def test_creates_the_record_type_directory(self, tmp_path): - build_path("likes", tmp_path) - - assert (tmp_path / "likes").is_dir() - - @pytest.mark.usefixtures("frozen_clock") - def test_partitions_by_record_type(self, tmp_path): - path = build_path("likes", tmp_path) - - assert path.parent == tmp_path / "likes" - assert path.name == "2026_07_23-06:48:11.parquet" - - def test_creates_parent_directories(self, tmp_path): - nested = tmp_path / "a" / "b" / "c" - path = build_path("posts", nested) - - assert path.parent.is_dir() - - def test_names_sort_chronologically(self, tmp_path): - """The repo format is zero-padded, so lexical order is time order.""" - - from bluesky_ingestion_jetstream import writer - - names = [] - for moment in [ - datetime(2026, 7, 23, 11, 4, 9, tzinfo=UTC), - datetime(2026, 7, 23, 6, 48, 11, tzinfo=UTC), - ]: - - class Fixed(datetime): - fixed = moment - - @classmethod - def now(cls, tz=None): - return cls.fixed - - original = writer.datetime - writer.datetime = Fixed - names.append(build_path("posts", tmp_path).name) - writer.datetime = original - - assert sorted(names) == [names[1], names[0]] - - @pytest.mark.usefixtures("frozen_clock") - def test_collision_gets_a_suffix(self, tmp_path): - """The format resolves to the second, so two flushes can collide.""" - - first = build_path("likes", tmp_path) - first.touch() - second = build_path("likes", tmp_path) - - assert second.name == "2026_07_23-06:48:11-1.parquet" - - @pytest.mark.usefixtures("frozen_clock") - def test_repeated_collisions_keep_incrementing(self, tmp_path): - names = [] - for _ in range(3): - path = build_path("likes", tmp_path) - path.touch() - names.append(path.name) - - assert names == [ - "2026_07_23-06:48:11.parquet", - "2026_07_23-06:48:11-1.parquet", - "2026_07_23-06:48:11-2.parquet", - ] - - @pytest.mark.usefixtures("frozen_clock") - def test_never_returns_an_existing_path(self, tmp_path): - """Returning one would silently overwrite a written flush.""" - - for _ in range(3): - path = build_path("likes", tmp_path) - assert not path.exists() - path.touch() - - -class TestWrite: - @pytest.mark.parametrize("record_type", RECORD_TYPES) - def test_writes_a_readable_file(self, record_type, rows_factory, tmp_path): - rows = rows_factory(record_type, 5) - path = write(record_type, rows, tmp_path) - - table = pq.read_table(path) - - assert path.is_file() - assert table.num_rows == 5 - - @pytest.mark.parametrize("record_type", RECORD_TYPES) - def test_file_matches_the_declared_schema(self, record_type, rows_factory, tmp_path): - path = write(record_type, rows_factory(record_type, 3), tmp_path) - - table = pq.read_table(path) - - assert table.schema.equals(RECORD_TYPE_TO_SCHEMA[record_type], check_metadata=False) - - def test_values_survive_the_round_trip(self, rows_factory, tmp_path): - rows = rows_factory("follows", 3) - path = write("follows", rows, tmp_path) - - table = pq.read_table(path) - - assert table.column("uri").to_pylist() == [row["uri"] for row in rows] - assert table.column("subject_did").to_pylist() == [row["subject_did"] for row in rows] - assert table.column("created_at").to_pylist() == [row["created_at"] for row in rows] - - def test_langs_round_trips_as_a_list(self, rows_factory, tmp_path): - path = write("posts", rows_factory("posts", 2), tmp_path) - - assert pq.read_table(path).column("langs").to_pylist() == [["en"], ["en"]] - - def test_null_optional_columns_are_written(self, tmp_path): - """A row whose optional columns are all null must still persist.""" - - from bluesky_ingestion_jetstream.network.connection import process_commit_event - from tests.bluesky_ingestion_jetstream.conftest import POST_COLLECTION, make_event - - parsed = process_commit_event( - make_event(POST_COLLECTION, {"createdAt": "2026-07-23T06:48:11Z"}) - ) - assert parsed is not None - - table = pq.read_table(write("posts", [parsed[1]], tmp_path)) - - assert table.num_rows == 1 - assert table.column("text").to_pylist() == [None] - assert table.column("langs").to_pylist() == [None] - - def test_returns_the_written_path(self, rows_factory, tmp_path): - path = write("likes", rows_factory("likes", 1), tmp_path) - - assert path.parent == tmp_path / "likes" - assert path.suffix == ".parquet" - - def test_successive_writes_do_not_overwrite(self, rows_factory, tmp_path): - first = write("likes", rows_factory("likes", 2), tmp_path) - second = write("likes", rows_factory("likes", 3), tmp_path) - - assert first != second - assert pq.read_table(first).num_rows == 2 - assert pq.read_table(second).num_rows == 3 - - def test_empty_rows_write_an_empty_file(self, tmp_path): - """flush() guards against this, but the writer must not raise on it.""" - - table = pq.read_table(write("likes", [], tmp_path)) - - assert table.num_rows == 0 - - def test_unknown_record_type_raises(self, rows_factory, tmp_path): - with pytest.raises(KeyError): - write("blocks", rows_factory("likes", 1), tmp_path) diff --git a/uv.lock b/uv.lock index 20393e10..3466f2eb 100644 --- a/uv.lock +++ b/uv.lock @@ -8,7 +8,7 @@ resolution-markers = [ ] [options] -exclude-newer = "2026-07-16T02:50:59.165364Z" +exclude-newer = "2026-07-21T09:35:18.413802Z" exclude-newer-span = "P7D" [[package]] @@ -326,11 +326,11 @@ wheels = [ [[package]] name = "cachetools" -version = "7.1.4" +version = "6.2.6" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/f4/8b/0d3945a13955303b81272f759a0331e54c5c793da455e6f5706b89d2639c/cachetools-7.1.4.tar.gz", hash = "sha256:437f55a4e0c1b01a4f3077cc470e6991d47430970e36fbcb77e2be0df4fc1cd6", size = 40085, upload-time = "2026-05-21T22:40:43.376Z" } +sdist = { url = "https://files.pythonhosted.org/packages/39/91/d9ae9a66b01102a18cd16db0cf4cd54187ffe10f0865cc80071a4104fbb3/cachetools-6.2.6.tar.gz", hash = "sha256:16c33e1f276b9a9c0b49ab5782d901e3ad3de0dd6da9bf9bcd29ac5672f2f9e6", size = 32363, upload-time = "2026-01-27T20:32:59.956Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/8c/7b/1fc1c09cc0756cf25861a3be10565915953876da48bb228fb9a672b20a42/cachetools-7.1.4-py3-none-any.whl", hash = "sha256:323dc4127934744db5b54eb4924482d7edafbf9554e820d1531c2e08c0e4ef54", size = 16761, upload-time = "2026-05-21T22:40:41.845Z" }, + { url = "https://files.pythonhosted.org/packages/90/45/f458fa2c388e79dd9d8b9b0c99f1d31b568f27388f2fdba7bb66bbc0c6ed/cachetools-6.2.6-py3-none-any.whl", hash = "sha256:8c9717235b3c651603fff0076db52d6acbfd1b338b8ed50256092f7ce9c85bda", size = 11668, upload-time = "2026-01-27T20:32:58.527Z" }, ] [[package]] @@ -1153,6 +1153,7 @@ dependencies = [ { name = "prefect" }, { name = "pyarrow" }, { name = "pydantic" }, + { name = "pyiceberg", extra = ["glue", "pyiceberg-core"] }, { name = "pyjwt" }, { name = "python-dotenv" }, { name = "pyyaml" }, @@ -1219,11 +1220,12 @@ requires-dist = [ { name = "psycopg", extras = ["binary", "pool"], marker = "extra == 'db-experiments'", specifier = ">=3.2.0" }, { name = "pyarrow", specifier = ">=14.0.0" }, { name = "pydantic", specifier = ">=2.13.3" }, + { name = "pyiceberg", extras = ["glue", "pyiceberg-core"], specifier = ">=0.10.0" }, { name = "pyjwt", specifier = ">=2.13.0" }, { name = "pytest", marker = "extra == 'testing'", specifier = ">=8.0.0" }, { name = "python-dotenv", specifier = ">=1.2.2" }, { name = "pyyaml", specifier = ">=6.0.0" }, - { name = "rich", specifier = ">=15.0.0" }, + { name = "rich", specifier = ">=13.9.4,<15" }, { name = "scikit-learn", specifier = ">=1.8.0" }, { name = "spacy", specifier = ">=3.7.0" }, { name = "tenacity", specifier = ">=9.0.0" }, @@ -1501,6 +1503,30 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, ] +[[package]] +name = "mmh3" +version = "5.2.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/91/1a/edb23803a168f070ded7a3014c6d706f63b90c84ccc024f89d794a3b7a6d/mmh3-5.2.1.tar.gz", hash = "sha256:bbea5b775f0ac84945191fb83f845a6fd9a21a03ea7f2e187defac7e401616ad", size = 33775, upload-time = "2026-03-05T15:55:57.716Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/65/d7/3312a59df3c1cdd783f4cf0c4ee8e9decff9c5466937182e4cc7dbbfe6c5/mmh3-5.2.1-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:dae0f0bd7d30c0ad61b9a504e8e272cb8391eed3f1587edf933f4f6b33437450", size = 56082, upload-time = "2026-03-05T15:53:59.702Z" }, + { url = "https://files.pythonhosted.org/packages/61/96/6f617baa098ca0d2989bfec6d28b5719532cd8d8848782662f5b755f657f/mmh3-5.2.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:9aeaf53eaa075dd63e81512522fd180097312fb2c9f476333309184285c49ce0", size = 40458, upload-time = "2026-03-05T15:54:01.548Z" }, + { url = "https://files.pythonhosted.org/packages/c1/b4/9cd284bd6062d711e13d26c04d4778ab3f690c1c38a4563e3c767ec8802e/mmh3-5.2.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:0634581290e6714c068f4aa24020acf7880927d1f0084fa753d9799ae9610082", size = 40079, upload-time = "2026-03-05T15:54:02.743Z" }, + { url = "https://files.pythonhosted.org/packages/f6/09/a806334ce1d3d50bf782b95fcee8b3648e1e170327d4bb7b4bad2ad7d956/mmh3-5.2.1-cp311-cp311-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:e080c0637aea036f35507e803a4778f119a9b436617694ae1c5c366805f1e997", size = 97242, upload-time = "2026-03-05T15:54:04.536Z" }, + { url = "https://files.pythonhosted.org/packages/ee/93/723e317dd9e041c4dc4566a2eb53b01ad94de31750e0b834f1643905e97c/mmh3-5.2.1-cp311-cp311-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:db0562c5f71d18596dcd45e854cf2eeba27d7543e1a3acdafb7eef728f7fe85d", size = 103082, upload-time = "2026-03-05T15:54:06.387Z" }, + { url = "https://files.pythonhosted.org/packages/61/b5/f96121e69cc48696075071531cf574f112e1ffd08059f4bffb41210e6fc5/mmh3-5.2.1-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1d9f9a3ce559a5267014b04b82956993270f63ec91765e13e9fd73daf2d2738e", size = 106054, upload-time = "2026-03-05T15:54:07.506Z" }, + { url = "https://files.pythonhosted.org/packages/82/49/192b987ec48d0b2aecf8ac285a9b11fbc00030f6b9c694664ae923458dde/mmh3-5.2.1-cp311-cp311-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:960b1b3efa39872ac8b6cc3a556edd6fb90ed74f08c9c45e028f1005b26aa55d", size = 112910, upload-time = "2026-03-05T15:54:09.403Z" }, + { url = "https://files.pythonhosted.org/packages/cf/a1/03e91fd334ed0144b83343a76eb11f17434cd08f746401488cfeafb2d241/mmh3-5.2.1-cp311-cp311-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:d30b650595fdbe32366b94cb14f30bb2b625e512bd4e1df00611f99dc5c27fd4", size = 120551, upload-time = "2026-03-05T15:54:10.587Z" }, + { url = "https://files.pythonhosted.org/packages/93/b9/b89a71d2ff35c3a764d1c066c7313fc62c7cc48fa48a4b3b0304a4a0146f/mmh3-5.2.1-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:82f3802bfc4751f420d591c5c864de538b71cea117fce67e4595c2afede08a15", size = 99096, upload-time = "2026-03-05T15:54:11.76Z" }, + { url = "https://files.pythonhosted.org/packages/36/b5/613772c1c6ed5f7b63df55eb131e887cc43720fec392777b95a79d34e640/mmh3-5.2.1-cp311-cp311-musllinux_1_2_i686.whl", hash = "sha256:915e7a2418f10bd1151b1953df06d896db9783c9cfdb9a8ee1f9b3a4331ab503", size = 98524, upload-time = "2026-03-05T15:54:13.122Z" }, + { url = "https://files.pythonhosted.org/packages/5e/0e/1524566fe8eaf871e4f7bc44095929fcd2620488f402822d848df19d679c/mmh3-5.2.1-cp311-cp311-musllinux_1_2_ppc64le.whl", hash = "sha256:fc78739b5ec6e4fb02301984a3d442a91406e7700efbe305071e7fd1c78278f2", size = 106239, upload-time = "2026-03-05T15:54:14.601Z" }, + { url = "https://files.pythonhosted.org/packages/04/94/21adfa7d90a7a697137ad6de33eeff6445420ca55e433a5d4919c79bc3b5/mmh3-5.2.1-cp311-cp311-musllinux_1_2_s390x.whl", hash = "sha256:41aac7002a749f08727cb91babff1daf8deac317c0b1f317adc69be0e6c375d1", size = 109797, upload-time = "2026-03-05T15:54:15.819Z" }, + { url = "https://files.pythonhosted.org/packages/b5/e6/1aacc3a219e1aa62fa65669995d4a3562b35be5200ec03680c7e4bec9676/mmh3-5.2.1-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:9d8089d853c7963a8ce87fff93e2a67075c0bc08684a08ea6ad13577c38ffc38", size = 97228, upload-time = "2026-03-05T15:54:16.992Z" }, + { url = "https://files.pythonhosted.org/packages/f1/b9/5e4cca8dcccf298add0a27f3c357bc8cf8baf821d35cdc6165e4bd5a48b0/mmh3-5.2.1-cp311-cp311-win32.whl", hash = "sha256:baeb47635cb33375dee4924cd93d7f5dcaa786c740b08423b0209b824a1ee728", size = 40751, upload-time = "2026-03-05T15:54:18.714Z" }, + { url = "https://files.pythonhosted.org/packages/72/fc/5b11d49247f499bcda591171e9cf3b6ee422b19e70aa2cef2e0ae65ca3b9/mmh3-5.2.1-cp311-cp311-win_amd64.whl", hash = "sha256:1e4ecee40ba19e6975e1120829796770325841c2f153c0e9aecca927194c6a2a", size = 41517, upload-time = "2026-03-05T15:54:19.764Z" }, + { url = "https://files.pythonhosted.org/packages/8a/5f/2a511ee8a1c2a527c77726d5231685b72312c5a1a1b7639ad66a9652aa84/mmh3-5.2.1-cp311-cp311-win_arm64.whl", hash = "sha256:c302245fd6c33d96bd169c7ccf2513c20f4c1e417c07ce9dce107c8bc3f8411f", size = 39287, upload-time = "2026-03-05T15:54:20.904Z" }, +] + [[package]] name = "multidict" version = "6.7.1" @@ -2403,6 +2429,56 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" }, ] +[[package]] +name = "pyiceberg" +version = "0.11.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "cachetools" }, + { name = "click" }, + { name = "fsspec" }, + { name = "mmh3" }, + { name = "pydantic" }, + { name = "pyparsing" }, + { name = "pyroaring" }, + { name = "requests" }, + { name = "rich" }, + { name = "strictyaml" }, + { name = "tenacity" }, + { name = "zstandard" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ce/f0/7616676603fdbd05ab97816337a9b31be08a5f9e1ffd636260812b217e0f/pyiceberg-0.11.1.tar.gz", hash = "sha256:366fe0d5a74e3cf1d4e7cbf3c49e308da60e7835ea268667be9185388f05d7a5", size = 1076075, upload-time = "2026-03-03T00:10:27.61Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/62/f7/3b7fee2ecc021f0526f23ef4ae5dcc8e0ed26062c35890ad25d39c53fb3b/pyiceberg-0.11.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:ba98d6a41ec0b7c81dd85d764f15653d6abbbbd69d92630677c43f92dd50d924", size = 532406, upload-time = "2026-03-03T00:09:59.647Z" }, + { url = "https://files.pythonhosted.org/packages/94/25/324030b13d91b7b564fb7342bf3fcdbf76eed2672964b273b156bf84d6e5/pyiceberg-0.11.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:6400774e6820760eb6c322f6feace43fe7267deb9f8d508f10bf258887a9c4d5", size = 533368, upload-time = "2026-03-03T00:10:01.264Z" }, + { url = "https://files.pythonhosted.org/packages/1c/83/6a43d06a079292c4fc7815b4de3e90a05ded90031c35d0a1b037659f722b/pyiceberg-0.11.1-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1663d79fc8400903992c63f79b3908b9298c623138e8929bf36c559231e082d3", size = 722886, upload-time = "2026-03-03T00:10:02.558Z" }, + { url = "https://files.pythonhosted.org/packages/ce/5c/53807036b63bb810f2c56b4b5576e79b721ba93f1c16bd0dd49ecfe41055/pyiceberg-0.11.1-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:856c7fca5ed780ed44f60bceb92d6b311ebc008a2249415b8f6045201d4f5530", size = 721212, upload-time = "2026-03-03T00:10:03.694Z" }, + { url = "https://files.pythonhosted.org/packages/db/11/65e25d6016e3844c516c9f04041853115711f64af1ee184a2320c9ceab4c/pyiceberg-0.11.1-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:6cf94756fb6a822d20a5a64f44840e6633ebf8b1deb3ce01057bff1cc03b01c2", size = 717978, upload-time = "2026-03-03T00:10:05.011Z" }, + { url = "https://files.pythonhosted.org/packages/52/0b/33b4ea9dc7f0c496900f5fc6da79e8587e94b88e2244ff02b786016cc649/pyiceberg-0.11.1-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:8bd52c1891ae74cee21a4ebe8325953310a2e0af5352d70f47f5461422fcce2d", size = 718747, upload-time = "2026-03-03T00:10:06.269Z" }, + { url = "https://files.pythonhosted.org/packages/c4/b6/5e133c435efc577afea5be303d7123264a5176a3ce1e2d3dc3a691049eaf/pyiceberg-0.11.1-cp311-cp311-win_amd64.whl", hash = "sha256:65a7ad892a570045b0de2db6af17119162880aebc05a0c125ce2db7dab36f17e", size = 531081, upload-time = "2026-03-03T00:10:07.352Z" }, +] + +[package.optional-dependencies] +glue = [ + { name = "boto3" }, +] +pyiceberg-core = [ + { name = "pyiceberg-core" }, +] + +[[package]] +name = "pyiceberg-core" +version = "0.8.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/9c/a0/0bcedbbe901484aacb6c605505f8574fd65954826e592fdb163e1cfb09f2/pyiceberg_core-0.8.0.tar.gz", hash = "sha256:59021ca5bc7ca95f2b06fb0730280fb3f60ed898060bcd874c156d093853b5f3", size = 618882, upload-time = "2026-01-20T00:50:40.076Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/77/e0/9a8fa537d29d34e3265682056d6517b926975107b5b1af6057d1713557d6/pyiceberg_core-0.8.0-cp310-abi3-macosx_10_12_x86_64.macosx_11_0_arm64.macosx_10_12_universal2.whl", hash = "sha256:d60c75a741a1d9199277a9e50fc3adbc84ab286a881f9b1f721fa120e7197912", size = 24733948, upload-time = "2026-01-20T00:50:20.566Z" }, + { url = "https://files.pythonhosted.org/packages/6d/3e/f5522c1e9c20c3e89bfd76b2f54ba38e57389e5a2872233e49e60a131e04/pyiceberg_core-0.8.0-cp310-abi3-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:1d71e566b2d56141760ff8734667eede5a5d60963dfbcdce80c2dd3cf2edb39d", size = 11682041, upload-time = "2026-01-20T00:50:22.843Z" }, + { url = "https://files.pythonhosted.org/packages/95/4b/f799e5c7a2b2ede75514e64901503358a7a134ca1ea217fd86535af533b6/pyiceberg_core-0.8.0-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:82782d1b974200c5526d069391ba2bc235a868b5d0d6ac17ca406df735ab89a3", size = 13835428, upload-time = "2026-01-20T00:50:25.021Z" }, + { url = "https://files.pythonhosted.org/packages/0b/ff/2dbd6f7c99a2f782f908be2cc997371de45cc1df61abeeff1fc0165c05b6/pyiceberg_core-0.8.0-cp310-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:0e14b2aea26293ba5878c398adc880fff0f1ce5d989e00d4b1a930c143541114", size = 14580807, upload-time = "2026-01-20T00:50:27.158Z" }, + { url = "https://files.pythonhosted.org/packages/bc/13/176c2b00a9b804af79d8b697ba1a2525f4390e959be777076972071ca069/pyiceberg_core-0.8.0-cp310-abi3-win_amd64.whl", hash = "sha256:a5726cc62f9ac2582a0d5dde92e4140b711b5e29ec0c6c636d6d2782d984031b", size = 13354110, upload-time = "2026-01-20T00:50:29.785Z" }, +] + [[package]] name = "pyjwt" version = "2.13.0" @@ -2412,6 +2488,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a3/5e/ecf12fdb62546d64385c158514e9b2b671f7832108ef2ecd2020ce0af2d1/pyjwt-2.13.0-py3-none-any.whl", hash = "sha256:66adcc2aff09b3f1bbd95fc1e1577df8ac8723c978552fd43304c8a290ac5728", size = 31274, upload-time = "2026-05-21T19:54:35.362Z" }, ] +[[package]] +name = "pyparsing" +version = "3.3.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f3/91/9c6ee907786a473bf81c5f53cf703ba0957b23ab84c264080fb5a450416f/pyparsing-3.3.2.tar.gz", hash = "sha256:c777f4d763f140633dcb6d8a3eda953bf7a214dc4eff598413c070bcdc117cbc", size = 6851574, upload-time = "2026-01-21T03:57:59.36Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/10/bd/c038d7cc38edc1aa5bf91ab8068b63d4308c66c4c8bb3cbba7dfbc049f9c/pyparsing-3.3.2-py3-none-any.whl", hash = "sha256:850ba148bd908d7e2411587e247a1e4f0327839c40e2e5e6d05a007ecc69911d", size = 122781, upload-time = "2026-01-21T03:57:55.912Z" }, +] + [[package]] name = "pyright" version = "1.1.409" @@ -2425,6 +2510,26 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/16/6b/330d8ebae582b30c2959a1ef4c3bc344ebde48c2ff0c3f113c4710735e11/pyright-1.1.409-py3-none-any.whl", hash = "sha256:aa3ea228cab90c845c7a60d28db7a844c04315356392aa09fafcee98c8c22fb3", size = 6438161, upload-time = "2026-04-23T11:02:01.309Z" }, ] +[[package]] +name = "pyroaring" +version = "1.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d7/46/a50510d080f8cb089303ec0f7cd80736b2949ca3d148f48f1cc90c49e345/pyroaring-1.1.0.tar.gz", hash = "sha256:f02e4021397ae02a139defdc6813b9942ab163de90affddd4ce4efbac299f619", size = 200298, upload-time = "2026-04-24T21:29:25.212Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/4e/aa/574f153feb89856010092c33cafe4a52f4da8cea19d441cbc4cbcb8ccb1f/pyroaring-1.1.0-cp311-cp311-macosx_14_0_arm64.whl", hash = "sha256:462b0952277d3100a90ae890ae641d3fb3561b10cfea542e02468f0bef7700a5", size = 332142, upload-time = "2026-04-24T21:27:51.258Z" }, + { url = "https://files.pythonhosted.org/packages/fd/a1/938a9fbdd41699ac9419ca922fcf80eb58c0c62aa34539aa540abaee9a63/pyroaring-1.1.0-cp311-cp311-macosx_14_0_universal2.whl", hash = "sha256:9cf8608c9d6cb6bff9c624744f7a2ba8ab12276f097a07930490fe0e2219e9be", size = 706441, upload-time = "2026-04-24T21:27:52.226Z" }, + { url = "https://files.pythonhosted.org/packages/06/06/5120acc9918223ccb647dfadb430da1ec08d7ada818469ad3a2f1574b8c7/pyroaring-1.1.0-cp311-cp311-macosx_14_0_x86_64.whl", hash = "sha256:44a9f9719ed18e86627286f90057f9b7e22f6ba1d952c9793f9600b5c14e8680", size = 382003, upload-time = "2026-04-24T21:27:53.322Z" }, + { url = "https://files.pythonhosted.org/packages/88/73/adc7951e0d4e65ab7a84fa5562bc28ba2dcc6f461a9156edc0d5aefda7f6/pyroaring-1.1.0-cp311-cp311-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c14fc6bd65e5624f76b90297b081222261476978f795f60d48745553617ddceb", size = 2034493, upload-time = "2026-04-24T21:27:54.739Z" }, + { url = "https://files.pythonhosted.org/packages/c7/28/992f2f2ef573b54559ed4bc62b5cf0ea095a59173521cfbe78dfefcc08ad/pyroaring-1.1.0-cp311-cp311-manylinux_2_24_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:01212da3752d6486adcca98c9d353f8fb8e36513e05062cdd0feebc4211dbe70", size = 1901479, upload-time = "2026-04-24T21:27:57.115Z" }, + { url = "https://files.pythonhosted.org/packages/10/4b/38bded4ca17af6359c1766c25e942435332d26e2cb3321d9d081546b75f8/pyroaring-1.1.0-cp311-cp311-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:100585f438b293112e2c52e45a442835837c8a0267dd1e513bafec35628f8ecb", size = 2235338, upload-time = "2026-04-24T21:27:58.66Z" }, + { url = "https://files.pythonhosted.org/packages/ac/ff/8e7da41468a33fc9d0fbee949e8f3f734930b5173020a478686558388f31/pyroaring-1.1.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:532e53191d8dd29dedfc5202cbb45632f7df751b207a7f6d6860fb7067c7fe11", size = 2951300, upload-time = "2026-04-24T21:28:00.12Z" }, + { url = "https://files.pythonhosted.org/packages/23/a8/f8a4b55b0817aac3caf6b3b51f2e93cf152f32784949919512bd38feb0f6/pyroaring-1.1.0-cp311-cp311-musllinux_1_2_armv7l.whl", hash = "sha256:731a7a9e050758986d5757eea10f9ccb08f9c3ef514ce0335f4a90e126f81131", size = 2752747, upload-time = "2026-04-24T21:28:01.281Z" }, + { url = "https://files.pythonhosted.org/packages/b7/26/b5a29c0f38c581ac290245150d4d14b627f110eac208f3569d5d8739c78f/pyroaring-1.1.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:e9864e19109e76111befc75d799d334e7365eb4189607aa734053c12e7840fa5", size = 3203933, upload-time = "2026-04-24T21:28:02.595Z" }, + { url = "https://files.pythonhosted.org/packages/99/60/2b3661d9a9e12dac02d2c3ef4545bd2236fcd964ba8fdd96e308ca621153/pyroaring-1.1.0-cp311-cp311-win32.whl", hash = "sha256:7caf95de39ce869ea0978068521cf6faa7350574fd1734ad6c63e5ed8cd06baa", size = 209205, upload-time = "2026-04-24T21:28:03.811Z" }, + { url = "https://files.pythonhosted.org/packages/be/1f/416a8cf29738d2e8c552fcaff7951c2a9a3bac0faffbb88b888287665834/pyroaring-1.1.0-cp311-cp311-win_amd64.whl", hash = "sha256:a23cd023985b5f2ba23e84e1fadaeacde3c8a59e1d2adb3fe782e99db1e22387", size = 260338, upload-time = "2026-04-24T21:28:04.752Z" }, + { url = "https://files.pythonhosted.org/packages/a9/0d/fd3f57014e98ffd20a3db7fc07157be8abb6cc5a356ccb20e1ea2493c397/pyroaring-1.1.0-cp311-cp311-win_arm64.whl", hash = "sha256:a98d1147fe1d3195053b67b474bccc0be5021506765d27f613a943c8c99f9e4c", size = 217570, upload-time = "2026-04-24T21:28:06.126Z" }, +] + [[package]] name = "pytest" version = "9.0.3" @@ -2669,15 +2774,15 @@ wheels = [ [[package]] name = "rich" -version = "15.0.0" +version = "14.3.4" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "markdown-it-py" }, { name = "pygments" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/c0/8f/0722ca900cc807c13a6a0c696dacf35430f72e0ec571c4275d2371fca3e9/rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36", size = 230680, upload-time = "2026-04-12T08:24:00.75Z" } +sdist = { url = "https://files.pythonhosted.org/packages/e9/67/cae617f1351490c25a4b8ac3b8b63a4dda609295d8222bad12242dfdc629/rich-14.3.4.tar.gz", hash = "sha256:817e02727f2b25b40ef56f5aa2217f400c8489f79ca8f46ea2b70dd5e14558a9", size = 230524, upload-time = "2026-04-11T02:57:45.419Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" }, + { url = "https://files.pythonhosted.org/packages/b3/76/6d163cfac87b632216f71879e6b2cf17163f773ff59c00b5ff4900a80fa3/rich-14.3.4-py3-none-any.whl", hash = "sha256:07e7adb4690f68864777b1450859253bed81a99a31ac321ac1817b2313558952", size = 310480, upload-time = "2026-04-11T02:57:47.484Z" }, ] [[package]] @@ -3014,6 +3119,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0b/c9/584bc9651441b4ba60cc4d557d8a547b5aff901af35bda3a4ee30c819b82/starlette-1.0.0-py3-none-any.whl", hash = "sha256:d3ec55e0bb321692d275455ddfd3df75fff145d009685eb40dc91fc66b03d38b", size = 72651, upload-time = "2026-03-22T18:29:45.111Z" }, ] +[[package]] +name = "strictyaml" +version = "1.7.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "python-dateutil" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/b3/08/efd28d49162ce89c2ad61a88bd80e11fb77bc9f6c145402589112d38f8af/strictyaml-1.7.3.tar.gz", hash = "sha256:22f854a5fcab42b5ddba8030a0e4be51ca89af0267961c8d6cfa86395586c407", size = 115206, upload-time = "2023-03-10T12:50:27.062Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/96/7c/a81ef5ef10978dd073a854e0fa93b5d8021d0594b639cc8f6453c3c78a1d/strictyaml-1.7.3-py3-none-any.whl", hash = "sha256:fb5c8a4edb43bebb765959e420f9b3978d7f1af88c80606c03fb420888f5d1c7", size = 123917, upload-time = "2023-03-10T12:50:17.242Z" }, +] + [[package]] name = "tenacity" version = "9.1.4"