Uh oh!
There was an error while loading. Please reload this page.
Iceberg setup - #141
Conversation
…ons_interface into backfill_design_doc
…ons_interface into backfill_design_doc
… temp -> bluesky_backfill_app
🚅 Deployed to the lab_data_integrations_int-pr-141 environment in bubbly-courtesy
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughWalkthroughThe pipeline now provisions AWS storage, bootstraps Glue-backed Iceberg tables, streams batches through an Iceberg sink, retries commits, writes failed batches to S3 dead-letter storage, and rejects invalid event timestamps. ChangesIceberg ingestion pipeline
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Jetstream
participant Buffer
participant IcebergSink
participant IcebergTable
participant S3DeadLetter
Jetstream->>Buffer: accumulate record rows
Buffer->>IcebergSink: write(record_type, rows)
IcebergSink->>IcebergTable: append with flush_id
IcebergTable-->>IcebergSink: commit or commit error
IcebergSink->>IcebergTable: refresh and check snapshot tag
IcebergSink->>S3DeadLetter: write failed batch after commit failure
Possibly related issues
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Implementation looks good, just a few requests related to code quality and a request for adding an additional runbook. Good naming and conventions is important because code will be read more than it's written, by both ourselves and AI agents. It's OK to use Claude Code, but please do review the docs it provides as too much or incorrect detail can mess up both future users and AI agents (I ask Claude "add numpy-style docs" and that tends to leave informative clean docstrings). Also, adding implementation details within a docstring or comment that pertain to details not handled within that docstring (unless it's a comment about what's explicitly not handled there that we may expect to be handled there) is often at risk of quickly becoming stale documentation.
Great work on the implementation details! The rigorous architecture planning you did means that we've already preemptively dealt with a lot of the race conditions, edge cases, and other architectural gotcha's that tend to trip something like this up.
Slightly related: the code in data_platform that I wrote could've been better quality. I was in a bit of a time crunch for that one for a paper deadline and therefore didn't do as much QA as I would have for a production app (and we'll use that code more for its shape as a skeleton for what the data platform could look like). Given that this work you're doing needs to scale to millions of accounts, we need to establish good foundations for it (hence all the time we spent in design and architecture review, and the discussion here about code quality).
There was a problem hiding this comment.
let's keep this file for now even if it's not being read or used, for reference later. This is a pipeline that, in 3 months, we will likely want to borrow pieces from, so having a reference of the past architecture, even if deprecated, may prove useful at that point.
| from bluesky_ingestion_jetstream.constants import FOLLOWS, LIKES, POSTS, REPOSTS | ||
| # Filled in by the writer rather than the parsers, because the value is fixed for |
There was a problem hiding this comment.
this is unclear to me. It seems like the intention is that this is a shared field, like COMMON_FIELDS, but it's just one that's added at runtime. The comment is confusing if this is the main intention. I'm unsure if I especially follow the "Kept as a named list so tests can state which columns # are expected to be absent from parser output" note.
| ... | ||
| class MemorySink: |
There was a problem hiding this comment.
if this is only used in tests (which it appears to be), this should just be defined in the context of the conftest.py. The only interface that seems like it should be in the base.py is the Sink protocol.
| def new_run_id() -> str: | ||
| """Identify one process lifetime, stamped onto every row it writes. | ||
| A reconnect does not start a new run: the point of the column is to answer |
There was a problem hiding this comment.
I'm unsure I follow this comment. So a reconnect continues from the previous run ID? It seems like we can just cut this comment out altogether, unless there's some edge case we need to consider (in which case, we should document that elsewhere).
| def build_sink(run_id: str) -> IcebergSink: | ||
| """Load the four tables and wrap them in a sink. | ||
| Done before the first event rather than at the first flush, so a catalog that |
There was a problem hiding this comment.
this implementation detail shouldn't live in the docstring as the function doesn't control when this is done. This should be mentioned either in the spot in the code where this is done or in a doc somewhere else. But also it seems like this detail isn't necessary to include in the first place, as how it's used in the CLI below makes it clear that tables are first creating before you need them.
| ) -> Callable[[Callable[P, R]], Callable[P, R]]: | ||
| """Retry a commit on transient failures, with jittered exponential backoff. | ||
| The jitter is not decoration: all four tables flush together, so an unjittered |
There was a problem hiding this comment.
let's remove the jitter comment here, this is a generic pattern in designing systems with race conditions rather than a detail specific to this implementation.
| return None | ||
| def is_created_at_in_range(created_at: datetime, ingested_at: datetime | None) -> bool: |
There was a problem hiding this comment.
the docstring here has to clarify what should already be apparent in the function's name. If we rename it to something like is_created_at_valid, we can immediately tell what this function is for (which is validating that the created_at can be used for partitioning). is_created_at_in_range doesn't really tell us much about what range we're looking for.
| # Nulled rather than flagged, so an out-of-range timestamp leaves by the same | ||
| # door as an unparseable one: `created_at` is a required key, so | ||
| # `validate_non_null_fields` drops the row without a second code path. | ||
| if created_at is not None and not is_created_at_in_range(created_at, ingested_at): |
There was a problem hiding this comment.
to clarify the comment, is the intention here that even if the created_at field exists, if it's not a valid value then we set it to None, and then downstream callers will pick up the None value and filter out the row? If so, that's clearer than what the comment currently says.
| @@ -0,0 +1,34 @@ | |||
| """The contract between the buffers and wherever their rows end up. | |||
| Narrow on purpose. `storage/buffer.py` decides *when* to flush and holds the | |||
There was a problem hiding this comment.
The only thing we need in the docstring is "The contract between the buffers and wherever their rows end up.", as a base.py holding the interfaces and protocols is common enough practice.
| @@ -0,0 +1,97 @@ | |||
| """One-shot creation of the four Iceberg tables. Run by hand, not by the ingester. | |||
There was a problem hiding this comment.
Can you add a HOW_TO_SETUP_ICEBERG_TABLES runbook that mentions running this script so we know what to do for future reference?
@dudu-theman also a few questions about the notes you left (great job leaving these in there!)
Will deduplication be managed in #135?
I'm unsure if I understand this. For the first case, yes, we can check the actual flush IDs that were committed to the Iceberg tables. But then there's the "Cursor-based replay will create them, so that work needs a dedupe story on (uri, cid) at the same time", which seems to contradict "Nothing here can create duplicates on its own". So it is then possible for duplicates to exist, it's just that compaction deduplications? Could you elaborate?
This is OK for now, and probably lower on the backlog unless we start seeing a lot of records in the deadletter queue.
This is OK, though that should likely be in #138 as one of the last steps, since once we turn this on we'll want all these features included.
We should make this higher before we forget, either in this PR or a quick follow-up PR, since we may inadvertently forget to update this constant and then be hit with a surprising amount of writes per table. 12,000 table writes wouldn't be ideal, especially if each write is going to be pretty small. Ideally we would normally hit a max buffer size before we hit a max buffer age, else we're just writing smaller files than we need to.
This is OK. Good call-out. |
dudu-theman
commented
Aug 2, 2026
@mark-torres10 Took a look at the comments, and yep, agree with the docstring comments. In general I just let claude fill them in based off the conversation i have with it about implementation details, and questions I ask it, so it seems like it can get quite verbose in replaying the conversation I had + justification for why it's doing it this way. Will get back with a V2 soon |
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@bluesky_ingestion_jetstream/aws/catalog.py`:
- Around line 75-98: Add mock-backed unit tests for
bluesky_ingestion_jetstream/aws/catalog.py lines 75-98, covering load_tables
when every RECORD_TYPES table loads successfully and when multiple
NoSuchTableError cases are aggregated into MissingTablesError naming all missing
types. Add tests for bluesky_ingestion_jetstream/aws/bootstrap.py lines 71-86,
covering bootstrap creating a new table and handling TableAlreadyExistsError by
loading the existing table; exercise each branch through a mocked GlueCatalog
and Table interaction.
In `@bluesky_ingestion_jetstream/aws/dead_letter.py`:
- Around line 47-58: Update build_filesystem to address Linux environments where
request_timeout does not bound stalled S3 reads: either document the limitation
explicitly or add an explicit timeout around the dead-letter S3 write, while
preserving the existing connection and request timeout configuration.
In `@pyproject.toml`:
- Around line 25-27: Update the pyiceberg dependency specification in
pyproject.toml to include an upper version bound, matching the existing
bounded-dependency policy used for rich, while preserving the current extras and
minimum version.
In `@terraform/bluesky_ingestion_jetstream/main.tf`:
- Around line 48-59: Add an aws_s3_bucket_server_side_encryption_configuration
resource for aws_s3_bucket.warehouse, explicitly configuring server-side
encryption and linking it to the bucket. Use the standard SSE-S3 algorithm
initially, while keeping the configuration structured so it can later be changed
to a customer-managed KMS key.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 1c783776-338d-4641-91e7-0a4b0ce75f59
⛔ Files ignored due to path filters (4)
CHANGELOG.mdis excluded by!**/*.mddocs/runbooks/HOW_TO_SETUP_ICEBERG_TABLES.mdis excluded by!**/*.mdstrategy_planning/2026-07-24_bluesky_event_schemas.mdis excluded by!**/*.mduv.lockis excluded by!**/*.lock
📒 Files selected for processing (31)
bluesky_ingestion_jetstream/aws/__init__.pybluesky_ingestion_jetstream/aws/bootstrap.pybluesky_ingestion_jetstream/aws/catalog.pybluesky_ingestion_jetstream/aws/constants.pybluesky_ingestion_jetstream/aws/dead_letter.pybluesky_ingestion_jetstream/aws/iceberg_writer.pybluesky_ingestion_jetstream/aws/retry.pybluesky_ingestion_jetstream/constants.pybluesky_ingestion_jetstream/event_parsing/shared.pybluesky_ingestion_jetstream/main.pybluesky_ingestion_jetstream/schemas/arrow_schemas.pybluesky_ingestion_jetstream/sinks/__init__.pybluesky_ingestion_jetstream/sinks/base.pybluesky_ingestion_jetstream/sinks/iceberg.pybluesky_ingestion_jetstream/storage/buffer.pybluesky_ingestion_jetstream/writer.pypyproject.tomlterraform/bluesky_ingestion_jetstream/.terraform.lock.hclterraform/bluesky_ingestion_jetstream/main.tftests/bluesky_ingestion_jetstream/aws/__init__.pytests/bluesky_ingestion_jetstream/aws/test_constants.pytests/bluesky_ingestion_jetstream/aws/test_dead_letter.pytests/bluesky_ingestion_jetstream/conftest.pytests/bluesky_ingestion_jetstream/event_parsing/test_shared.pytests/bluesky_ingestion_jetstream/network/test_connection.pytests/bluesky_ingestion_jetstream/schemas/test_arrow_schemas.pytests/bluesky_ingestion_jetstream/sinks/__init__.pytests/bluesky_ingestion_jetstream/sinks/test_iceberg.pytests/bluesky_ingestion_jetstream/storage/test_buffer.pytests/bluesky_ingestion_jetstream/test_main.pytests/bluesky_ingestion_jetstream/test_writer.py
💤 Files with no reviewable changes (2)
- bluesky_ingestion_jetstream/writer.py
- tests/bluesky_ingestion_jetstream/test_writer.py
| 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 |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Add unit tests for the new Glue/Iceberg integration modules. Both catalog.py and bootstrap.py are new modules with nontrivial branching logic (missing-table aggregation, idempotent create-or-load) but ship without dedicated test files in this cohort; the shared root cause is that neither module's GlueCatalog/Table interaction is exercised by a mock-backed test.
bluesky_ingestion_jetstream/aws/catalog.py#L75-L98: add tests forload_tablescovering the all-tables-present path and the missing-table aggregation path (MissingTablesErrornaming every missing record type), using a mockedGlueCatalog.bluesky_ingestion_jetstream/aws/bootstrap.py#L71-L86: add tests forbootstrapcovering the create-new-table branch and theTableAlreadyExistsErrorload-existing-table branch, using a mockedGlueCatalog.
📍 Affects 2 files
bluesky_ingestion_jetstream/aws/catalog.py#L75-L98(this comment)bluesky_ingestion_jetstream/aws/bootstrap.py#L71-L86
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@bluesky_ingestion_jetstream/aws/catalog.py` around lines 75 - 98, Add
mock-backed unit tests for bluesky_ingestion_jetstream/aws/catalog.py lines
75-98, covering load_tables when every RECORD_TYPES table loads successfully and
when multiple NoSuchTableError cases are aggregated into MissingTablesError
naming all missing types. Add tests for
bluesky_ingestion_jetstream/aws/bootstrap.py lines 71-86, covering bootstrap
creating a new table and handling TableAlreadyExistsError by loading the
existing table; exercise each branch through a mocked GlueCatalog and Table
interaction.
| 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, | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
Does PyArrow 14.0.0 S3FileSystem request_timeout apply on Linux, or only on Windows and macOS?
💡 Result:
Yes, the request_timeout parameter in PyArrow 14.0.0 (and other versions) for S3FileSystem is explicitly documented to be ignored on operating systems other than Windows and macOS [1][2]. The official documentation for pyarrow.fs.S3FileSystem states that the request_timeout option is for "Socket read timeouts on Windows and macOS, in seconds" and explicitly notes that "This option is ignored on non-Windows, non-macOS systems" [1][3]. Consequently, it does not apply to Linux [1].
Citations:
- 1: https://arrow.apache.org/docs/14.0/python/generated/pyarrow.fs.S3FileSystem.html
- 2: https://github.com/apache/arrow/blob/main/python/pyarrow/_s3fs.pyx
- 3: https://arrow.apache.org/docs/16.1/python/generated/pyarrow.fs.S3FileSystem.html
Add an explicit read timeout for Linux S3 writes.
request_timeout does not apply on non-Windows/non-macOS systems; if this deployment runs on Linux, only connect_timeout limits the connection phase, not a stalled S3 read during the dead-letter write. Document this gap or add an explicit timeout around the write instead.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@bluesky_ingestion_jetstream/aws/dead_letter.py` around lines 47 - 58, Update
build_filesystem to address Linux environments where request_timeout does not
bound stalled S3 reads: either document the limitation explicitly or add an
explicit timeout around the dead-letter S3 write, while preserving the existing
connection and request timeout configuration.
| # pyiceberg-core supplies the Rust partition transforms; partitioned writes | ||
| # are unsupported without it. | ||
| "pyiceberg[glue,pyiceberg-core]>=0.10.0", |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Cap the pyiceberg upper bound, matching the rich pin above.
"pyiceberg[glue,pyiceberg-core]>=0.10.0" has no upper bound. PyIceberg's tracked package quality data lists "Follows Semver · No", so a future release can introduce breaking changes without a major version bump. The rich dependency two lines above is deliberately capped (>=13.9.4,<15) for exactly this reason. Apply the same discipline to pyiceberg to avoid an unplanned pipeline break on an unattended dependency bump.
♻️ Proposed fix
- "pyiceberg[glue,pyiceberg-core]>=0.10.0",+ "pyiceberg[glue,pyiceberg-core]>=0.10.0,<0.12",📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # pyiceberg-core supplies the Rust partition transforms; partitioned writes | |
| # are unsupported without it. | |
| "pyiceberg[glue,pyiceberg-core]>=0.10.0", | |
| # pyiceberg-core supplies the Rust partition transforms; partitioned writes | |
| # are unsupported without it. | |
| "pyiceberg[glue,pyiceberg-core]>=0.10.0,<0.12", |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pyproject.toml` around lines 25 - 27, Update the pyiceberg dependency
specification in pyproject.toml to include an upper version bound, matching the
existing bounded-dependency policy used for rich, while preserving the current
extras and minimum version.
| 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 | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🔵 Trivial | ⚡ Quick win
Add an explicit server-side encryption configuration for the warehouse bucket.
aws_s3_bucket.warehouse has no aws_s3_bucket_server_side_encryption_configuration resource. AWS applies default SSE-S3 encryption automatically for buckets created after January 2023, so objects are not stored unencrypted, but the bucket has no explicit, auditable encryption control (no KMS/CMK option, no policy-enforceable minimum). Add an explicit encryption resource so the control is visible in Terraform and can be upgraded to a customer-managed key later if this bucket's compliance requirements change.
🔒️ Proposed fix
resource "aws_s3_bucket" "warehouse" {
bucket = var.s3_bucket
}
++resource "aws_s3_bucket_server_side_encryption_configuration" "warehouse" {+ bucket = aws_s3_bucket.warehouse.id++ rule {+ apply_server_side_encryption_by_default {+ sse_algorithm = "AES256"+ }+ }+}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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 | |
| } | |
| resource"aws_s3_bucket""warehouse" { | |
| bucket=var.s3_bucket | |
| } | |
| resource"aws_s3_bucket_server_side_encryption_configuration""warehouse" { | |
| bucket=aws_s3_bucket.warehouse.id | |
| rule { | |
| apply_server_side_encryption_by_default { | |
| sse_algorithm="AES256" | |
| } | |
| } | |
| } | |
| 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 | |
| } |
🧰 Tools
🪛 Checkov (3.3.8)
[low] 48-50: Ensure S3 buckets should have event notifications enabled
(CKV2_AWS_62)
[medium] 48-50: Ensure that an S3 bucket has a lifecycle configuration
(CKV2_AWS_61)
[low] 48-50: Ensure that S3 bucket has cross-region replication enabled
(CKV_AWS_144)
[low] 48-50: Ensure all data stored in the S3 bucket have versioning enabled
(CKV_AWS_21)
[low] 48-50: Ensure that S3 buckets are encrypted with KMS by default
(CKV_AWS_145)
🪛 Trivy (0.72.0)
[info] 48-50: S3 Bucket Logging
Bucket has logging disabled
Rule: AWS-0089
Resource: aws_s3_bucket.warehouse
(IaC/AWS)
[warning] 48-50: S3 Data should be versioned
Bucket does not have versioning enabled
Rule: AWS-0090
Resource: aws_s3_bucket.warehouse
(IaC/AWS)
[error] 48-50: S3 encryption should use Customer Managed Keys
Bucket does not encrypt data with a customer managed key.
Rule: AWS-0132
Resource: aws_s3_bucket.warehouse
(IaC/AWS)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@terraform/bluesky_ingestion_jetstream/main.tf` around lines 48 - 59, Add an
aws_s3_bucket_server_side_encryption_configuration resource for
aws_s3_bucket.warehouse, explicitly configuring server-side encryption and
linking it to the bucket. Use the standard SSE-S3 algorithm initially, while
keeping the configuration structured so it can later be changed to a
customer-managed KMS key.
Source: Linters/SAST tools
mark-torres10
commented
Aug 3, 2026
Good work! LGTM. |
Overview
This PR moves the Jetstream ingestion output off local disk and into Iceberg tables registered in the AWS Glue Data Catalog. The buffers still decide when to flush; what changed is where the rows go — a flush is now an Iceberg commit (Parquet data files to S3, manifests,
metadata.json, then a GlueUpdateTable), wrapped in retry with exponential backoff and jitter, and falling back to a dead-letter prefix in S3 if the commit cannot be made.The old
terraform/data_platformstack was torn down and replaced byterraform/bluesky_ingestion_jetstream, which provisions only what Iceberg and Glue need: the S3 bucket, its public access block, and thebluesky_rawGlue database. The four Iceberg tables are deliberately not Terraform resources — Iceberg rewrites a table's schema, partition spec, and snapshot pointer on every commit, which anaws_glue_catalog_tablewould read as drift and revert on the next apply. They are created once bybootstrap.py.#132#133
Changes
created_attimestmap validationrun_idfor provenance in observability.Manual Verification
Terraform is already applied and the four Iceberg tables already exist, so this isa single end-to-end run. Let it flush a few times, then check the tables in the AWS console.
State (After)
S3 after several flush iterations (
bluesky/raw/<record_type>/data/created_at_day=.../):Notes
merge-on-readdoes not change that. Merge-on-read governs how aDELETE/UPDATE/MERGEis recorded — a small delete file that readers merge in at query time, rather than rewriting whole data files — but something has to issue that statement first. An append is just an append; there is no primary key and no upsert. Verified empirically: two appends of identical rows on a merge-on-read table gave 2 files and 2 rows for 1 distinct URI. The properties are set now because retracting duplicates under copy-on-write would rewrite entire 256 MB data files to remove a few rows each. Note PyIceberg cannot write delete files, so the eventual dedupe has to run from Athena or Spark.flush_idin the snapshot summary before repeating a commit, which covers the case where a Glue update lands but the response is lost. Cursor-based replay will create them, so that work needs a dedupe story on(uri, cid)at the same time.ERROR-level log line and a non-emptydead_letter/prefix are the only signals that the tables are incomplete.OPTIMIZE/VACUUMare not available without adding one.MAX_BUFFER_AGE_SECONDSis still 30s, which is ~2,880 commits per table per day and squarely in the small-files regime. Left as-is intentionally for testing; to be raised before this runs for real.Summary by CodeRabbit
New Features
Bug Fixes
Tests