Uh oh!
There was an error while loading. Please reload this page.
Add Iceberg write-amplification experiment - #127
Conversation
Measures the S3 and Glue cost of each operation in an Iceberg ingest pipeline fed by the Bluesky Jetstream firehose, against a raw-Parquet control. Capture and replay are separate: a 10-minute firehose capture is never reproducible, so it is recorded once and replayed as many times as needed, which keeps every write-path variant measured against byte-identical input. Metering works by pinning the catalog to FsspecFileIO. PyIceberg's default PyArrowFileIO drives a C++ S3 client that Python cannot intercept; fsspec routes through botocore, where every request is countable. Glue commits are counted as their own billing tier, since swapping metadata_location is a Glue UpdateTable, not an S3 call. Isolated venv: pyiceberg pins rich<15 and the root project requires rich>=15, so the two cannot share a resolution. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughWalkthroughAdds a complete Iceberg experiment pipeline covering Jetstream capture, event schemas, deterministic replay, raw and Iceberg writes, AWS metering, maintenance, orchestration, and markdown/JSON reporting. ChangesIceberg experiment pipeline
Estimated code review effort: 5 (Critical) | ~120 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 Checkov (3.3.8)experimentation/iceberg/data/results/20260723_075417-results.jsonTraceback (most recent call last): experimentation/iceberg/requirements.txtTraceback (most recent call last): 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 |
🚅 Deployed to the lab_data_integrations_int-pr-127 environment in bubbly-courtesy
|
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 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 `@experimentation/iceberg/catalog.py`:
- Around line 56-76: Update create_tables to handle TableAlreadyExistsError from
catalog.create_table when a reused run_id collides with an existing table. Fail
cleanly for the entire create_tables operation, or detect existing tables before
creation and skip the run, ensuring partial table creation is not left as an
unrecoverable path.
In `@experimentation/iceberg/constants.py`:
- Around line 43-47: Update COST_PER_GLUE_REQUEST in the constants definition to
represent the Glue API request rate of $1.00 per 1,000,000 requests, replacing
the current metadata storage-rate denominator. Preserve the existing cost model
structure and constant name.
In `@experimentation/iceberg/maintenance.py`:
- Around line 147-157: Update _delete_keys to use each delete_objects response
and count only keys not present in its Errors list, rather than assuming every
requested chunk key was deleted. Preserve batching and Quiet=True, and return
the accumulated successful-delete count for accurate orphans_deleted metrics.
In `@experimentation/iceberg/requirements.txt`:
- Around line 11-15: Update the five dependencies in requirements.txt—pyiceberg,
pyarrow, boto3, websockets, and pytest—to exact pinned versions instead of
lower-bound constraints, preserving the currently specified minimum versions
unless the experiment’s intended baseline defines different exact versions.
In `@experimentation/iceberg/s3_meter.py`:
- Around line 153-177: Update the attempt accounting in _on_before_send and
PhaseStats.cost_usd so each HTTP attempt contributes its request tier cost,
including retries, rather than relying only on by_tier counts from
_on_after_call. Propagate the tier or operation through request.context, using
the context populated by _on_before_call, and preserve logical-call metrics
separately from attempt-based billing.
🪄 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: 5c22b06e-8cc9-4427-b135-74e1e8b2a6e5
⛔ Files ignored due to path filters (2)
experimentation/iceberg/README.mdis excluded by!**/*.mdexperimentation/iceberg/data/results/20260723_075417-report.mdis excluded by!**/*.md
📒 Files selected for processing (19)
experimentation/iceberg/.gitignoreexperimentation/iceberg/capture.pyexperimentation/iceberg/catalog.pyexperimentation/iceberg/constants.pyexperimentation/iceberg/data/results/20260723_075417-results.jsonexperimentation/iceberg/iceberg_writer.pyexperimentation/iceberg/maintenance.pyexperimentation/iceberg/raw_writer.pyexperimentation/iceberg/replay.pyexperimentation/iceberg/report.pyexperimentation/iceberg/requirements.txtexperimentation/iceberg/run_experiment.pyexperimentation/iceberg/s3_meter.pyexperimentation/iceberg/schemas.pyexperimentation/iceberg/tests/conftest.pyexperimentation/iceberg/tests/test_replay.pyexperimentation/iceberg/tests/test_s3_meter.pyexperimentation/iceberg/tests/test_schemas.pypyproject.toml
| def create_tables(catalog: Catalog, run_id: str) -> dict[str, Any]: | ||
| """Create one partitioned table per record type. Returns record_type -> Table.""" | ||
| ensure_namespace(catalog) | ||
| tables: dict[str, Any] = {} | ||
| for record_type in constants.RECORD_TYPES: | ||
| identifier = (constants.GLUE_DATABASE, table_name(record_type, run_id)) | ||
| tables[record_type] = catalog.create_table( | ||
| identifier=identifier, | ||
| schema=schemas.SCHEMAS[record_type], | ||
| partition_spec=schemas.PARTITION_SPEC, | ||
| location=f"{warehouse_uri(run_id)}/{record_type}", | ||
| properties={ | ||
| "format-version": "2", | ||
| "write.parquet.compression-codec": "zstd", | ||
| # Leave stale metadata.json files in place so the expiry phase | ||
| # has real work to measure. | ||
| "write.metadata.delete-after-commit.enabled": "false", | ||
| }, | ||
| ) | ||
| return tables |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bashset -euo pipefail
echo"== locate files =="
git ls-files | rg '(^|/)catalog.py$|run_experiment.py$|constants.py$|schemas.py$|pyproject.toml$|requirements.*\.txt$'||trueecho"== catalog.py relevant section =="if [ -f experimentation/iceberg/catalog.py ];then
nl -ba experimentation/iceberg/catalog.py | sed -n '1,130p'fiecho"== run_experiment --run-id references =="if [ -f experimentation/iceberg/run_experiment.py ];then
rg -n -- "--run-id|run_id|run_id=" experimentation/iceberg/run_experiment.py
nl -ba experimentation/iceberg/run_experiment.py | sed -n '200,235p'fiecho"== exception handling references =="
rg -n "NoSuchTableError|TableAlreadyExistsError|create_tables|drop_tables|load_tables" experimentation/iceberg ||trueecho"== pyiceberg dependency/version refs =="
rg -n "pyiceberg|iceberg|Catalog|TableAlreadyExistsError|NoSuchTableError" pyproject.toml requirements*.txt experimentation 2>/dev/null ||trueRepository: METResearchGroup/lab_data_integrations_interface
Length of output: 844
🏁 Script executed:
#!/bin/bashset -euo pipefail
echo"== catalog.py relevant section =="
python3 - <<'PY'from pathlib import Pathp = Path("experimentation/iceberg/catalog.py")text = p.read_text()for i, line in enumerate(text.splitlines(), 1): if 1 <= i <= 130: print(f"{i:4d}\t{line}")PYecho"== run_experiment --run-id references =="
python3 - <<'PY'from pathlib import Pathp = Path("experimentation/iceberg/run_experiment.py")text = p.read_text()lines = text.splitlines()for start, end in [(1, 280), (200, 235)]: print(f"---- lines {start}-{end} ----") for i in range(start - 1, min(end, len(lines))): print(f"{i+1:4d}\t{lines[i]}")PYecho"== relevant exception/helper references =="
python3 - <<'PY'from pathlib import Pathpatterns = ["NoSuchTableError", "TableAlreadyExistsError", "create_tables", "drop_tables", "load_tables"]for path in [ Path("experimentation/iceberg/catalog.py"), Path("experimentation/iceberg/run_experiment.py")]: text = path.read_text() print(f"---- {path} ----") lines = text.splitlines() for i, line in enumerate(lines, 1): if any(tok in line for tok in patterns): print(f"{i:4d}\t{line}")PYecho"== pyiceberg dependency/version refs =="
python3 - <<'PY'for path in [Path("pyproject.toml"), Path("experimentation/iceberg/requirements.txt"), Path("pyproject.toml")]: if path.exists(): print(f"---- {path} ----") lines = path.read_text().splitlines() for i, line in enumerate(lines, 1): if any(tok in line.lower() for tok in ["pyiceberg", "iceberg", "py-io-impl"]): print(f"{i:4d}\t{line}")PYecho"== behavioral/source evidence from installed PyIceberg, if available =="
python3 - <<'PY'try: import pyiceberg from pyiceberg.exceptions import TableAlreadyExistsError, NoSuchTableError print("PyIceberg", pyiceberg.__version__) print("TableAlreadyExistsError base:", [type(base).__name__ for base in TableAlreadyExistsError.__mro__ if base is not object]) print("NoSuchTableError base:", [type(base).__name__ for base in NoSuchTableError.__mro__ if base is not object])except ImportError as e: print("PyIceberg import unavailable:", e)PYRepository: METResearchGroup/lab_data_integrations_interface
Length of output: 17388
🌐 Web query:
PyIceberg 0.8 exceptions TableAlreadyExistsError create_table raises TableAlreadyExistsError
💡 Result:
In PyIceberg, the create_table method is explicitly designed to raise a TableAlreadyExistsError if a table with the specified identifier already exists in the catalog [1][2]. This behavior is a core part of the API specification and is used to prevent accidental overwrites or conflicts during table creation [1][3]. If you wish to avoid this exception, the recommended approach is to use the create_table_if_not_exists method, which is available in the Catalog interface [1][2]. This method handles the check internally: it attempts to create the table and, if a TableAlreadyExistsError is caught, it proceeds to return the existing table instance instead [1][2]. Using create_table_if_not_exists is generally preferred over a manual "check-then-create" pattern (e.g., calling table_exists followed by create_table) because it performs the operation atomically within the catalog, avoiding potential race conditions where a concurrent process might create the table between your check and your creation call [2][4].
Citations:
- 1: https://py.iceberg.apache.org/reference/pyiceberg/catalog/
- 2: https://github.com/apache/iceberg-python/blob/e7590446/pyiceberg/catalog/__init__.py
- 3: https://github.com/apache/iceberg-python/blob/e7590446/tests/catalog/test_base.py
- 4: check if table exist apache/iceberg-python#406
🌐 Web query:
PyIceberg catalog create_table TableAlreadyExistsError
💡 Result:
In PyIceberg, the TableAlreadyExistsError is an exception raised by catalog operations when you attempt to create a table that already exists in the specified namespace [1][2]. To handle this scenario, you have two primary options: 1. Use create_table_if_not_exists: The most robust and recommended approach is to use the create_table_if_not_exists method provided by the catalog interface [2]. This method handles the existence check atomically, creating the table if it is absent or returning the existing table instance if it is already present [3][2]. from pyiceberg.catalog import load_catalog catalog = load_catalog("default") table = catalog.create_table_if_not_exists( identifier="your_namespace.your_table", schema=your_schema) 2. Exception Handling (Try-Except): Alternatively, you can call create_table and explicitly catch the TableAlreadyExistsError exception [4]. This pattern is a standard Pythonic way to handle potential collisions [4]. from pyiceberg.exceptions import TableAlreadyExistsError try: table = catalog.create_table( identifier="your_namespace.your_table", schema=your_schema) except TableAlreadyExistsError: table = catalog.load_table("your_namespace.your_table") Note that using catalog.table_exists followed by catalog.create_table is generally discouraged for production environments, as it introduces a race condition where the table could be created by a concurrent process between the two calls [4]. Utilizing create_table_if_not_exists is safer because it performs the operation in a single request [4].
Citations:
- 1: https://py.iceberg.apache.org/reference/pyiceberg/exceptions/
- 2: https://py.iceberg.apache.org/reference/pyiceberg/catalog/
- 3: Feat: Implement
create_table_if_not_existsapache/iceberg-python#415 - 4: check if table exist apache/iceberg-python#406
Handle reused --run-id table collisions.
--run-id can be provided explicitly and creates tables named <record_type>_<run_id>; rerunning with a used ID will hit catalog.create_table(...) and fail unrecoverably partway through table creation. Catch TableAlreadyExistsError here and fail cleanly for the whole create_tables path, or skip the run after detecting existing tables.
🤖 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 `@experimentation/iceberg/catalog.py` around lines 56 - 76, Update
create_tables to handle TableAlreadyExistsError from catalog.create_table when a
reused run_id collides with an existing table. Fail cleanly for the entire
create_tables operation, or detect existing tables before creation and skip the
run, ensuring partial table creation is not left as an unrecoverable path.
| COST_PER_PUT_REQUEST = 0.005 / 1_000 # PUT, COPY, POST, LIST | ||
| COST_PER_GET_REQUEST = 0.0004 / 1_000 # GET, SELECT, and all other requests | ||
| COST_PER_DELETE_REQUEST = 0.0 # DELETE and CANCEL are free | ||
| COST_PER_GLUE_REQUEST = 1.00 / 100_000 | ||
| COST_PER_GB_MONTH_STANDARD = 0.023 |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bashset -euo pipefail
echo"## files"
git ls-files | rg 'experimentation/iceberg|s3_meter|reporting|constants\.py|pytest|test'| sed -n '1,200p'echo"## constants.py relevant"if [ -f experimentation/iceberg/constants.py ];then
nl -ba experimentation/iceberg/constants.py | sed -n '1,120p'fiecho"## searches for Glue pricing / COST_PER_GLUE_REQUEST / Glue call metering"
rg -n "COST_PER_GLUE_REQUEST|Glue|GLUE|glue|Data Catalog|metadata object|1\.00 / 100_000|1\.00 / 1_000_000" -S .echo"## diff stat/name only"
git diff --stat
git diff --name-onlyRepository: METResearchGroup/lab_data_integrations_interface
Length of output: 4328
🌐 Web query:
AWS Glue Data Catalog request pricing $1.00 per 1,000,000 requests free tier metadata objects storage $1.00 per 100,000 objects
💡 Result:
The pricing for the AWS Glue Data Catalog as of July 2026 includes a free tier and usage-based charges for metadata storage and access [1][2]. Free Tier: The AWS Glue Data Catalog offers a free tier that includes the first 1 million metadata objects stored and the first 1 million requests (accesses) per month [1][3][4]. Pricing Beyond the Free Tier: - Metadata Storage: For storage exceeding the first 1 million objects, you are charged $1.00 per 100,000 objects per month [1][5]. - Metadata Requests: For requests exceeding the first 1 million per month, you are charged $1.00 per 1 million requests [1][3][5]. A metadata object in the AWS Glue Data Catalog is defined as a table, table version, partition, partition index, statistic, database, or catalog [1][6]. These charges apply to the storage and retrieval of this metadata, which is distinct from the costs associated with the underlying data files themselves (e.g., in Amazon S3) or the compute costs for running ETL jobs or crawlers [1][3].
Citations:
- 1: https://aws.amazon.com/glue/pricing/
- 2: https://awsglossary.org/terms/glue
- 3: https://www.integrate.io/blog/aws-glue-pricing/
- 4: https://aws.amazon.com/free/analytics/
- 5: https://www.wring.co/blog/aws-glue-pricing-guide
- 6: https://repost.aws/questions/QUWg6Xf0aOTxy34WjKgYZT8A/what-is-considered-a-glue-data-catalog-object-for-pricing
🏁 Script executed:
#!/bin/bashset -u
echo"## constants.py relevant"if [ -f experimentation/iceberg/constants.py ];then
awk '{printf "%6d: %s\n", NR, $0}' experimentation/iceberg/constants.py | sed -n '1,120p'fiecho"## searches for Glue pricing / COST_PER_GLUE_REQUEST / Glue call metering"
rg -n "COST_PER_GLUE_REQUEST|Glue|GLUE|glue|Data Catalog|metadata object|1\.00 / 100_000|1\.00 / 1_000_000" -S .||trueecho"## relevant s3_meter/report files"
awk '{printf "%6d: %s\n", NR, $0}' experimentation/iceberg/s3_meter.py | sed -n '1,160p'echo"--- report.py excerpt ---"
awk '{printf "%6d: %s\n", NR, $0}' experimentation/iceberg/report.py | sed -n '1,180p'echo"--- results excerpt ---"
python3 - <<'PY'import json, pathlibp=pathlib.Path('experimentation/iceberg/data/results/20260723_075417-results.json')if p.exists(): data=json.loads(p.read_text()) print(json.dumps(data, indent=2)[:12000])PYRepository: METResearchGroup/lab_data_integrations_interface
Length of output: 50405
Use the Glue request rate for COST_PER_GLUE_REQUEST.
COST_PER_GLUE_REQUEST = 1.00 / 100_000 models the Glue Data Catalog metadata object-storage rate, but this experiment charges per Glue call. The request rate is $1.00 per 1,000,000 requests above the free tier, so the reported Glue contribution is overcounted 10× unless storage costs are intentionally included elsewhere.
Proposed fix
-COST_PER_GLUE_REQUEST = 1.00 / 100_000+COST_PER_GLUE_REQUEST = 1.00 / 1_000_000📝 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.
| COST_PER_PUT_REQUEST=0.005/1_000# PUT, COPY, POST, LIST | |
| COST_PER_GET_REQUEST=0.0004/1_000# GET, SELECT, and all other requests | |
| COST_PER_DELETE_REQUEST=0.0# DELETE and CANCEL are free | |
| COST_PER_GLUE_REQUEST=1.00/100_000 | |
| COST_PER_GB_MONTH_STANDARD=0.023 | |
| COST_PER_PUT_REQUEST=0.005/1_000# PUT, COPY, POST, LIST | |
| COST_PER_GET_REQUEST=0.0004/1_000# GET, SELECT, and all other requests | |
| COST_PER_DELETE_REQUEST=0.0# DELETE and CANCEL are free | |
| COST_PER_GLUE_REQUEST=1.00/1_000_000 | |
| COST_PER_GB_MONTH_STANDARD=0.023 |
🤖 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 `@experimentation/iceberg/constants.py` around lines 43 - 47, Update
COST_PER_GLUE_REQUEST in the constants definition to represent the Glue API
request rate of $1.00 per 1,000,000 requests, replacing the current metadata
storage-rate denominator. Preserve the existing cost model structure and
constant name.
| def _delete_keys(client: Any, keys: list[str]) -> int: | ||
| """Batch-delete keys 1000 at a time (the DeleteObjects limit).""" | ||
| deleted = 0 | ||
| for start in range(0, len(keys), 1000): | ||
| chunk = keys[start : start + 1000] | ||
| client.delete_objects( | ||
| Bucket=constants.S3_BUCKET, | ||
| Delete={"Objects": [{"Key": key} for key in chunk], "Quiet": True}, | ||
| ) | ||
| deleted += len(chunk) | ||
| return deleted |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bashset -euo pipefail
echo"== Locate file =="
git ls-files | rg '(^|/)maintenance\.py$|experimentation/iceberg/maintenance\.py'echo"== Inspect relevant lines =="if [ -f experimentation/iceberg/maintenance.py ];then
wc -l experimentation/iceberg/maintenance.py
sed -n '120,230p' experimentation/iceberg/maintenance.py | cat -n
fiecho"== Search for delete_objects responses/usages =="
rg -n "delete_objects|_delete_keys|def sweep_orphans|_referenced_paths|quiet" experimentation/iceberg -S ||trueecho"== Check boto3/type stub for delete_objects response shape in installed env, if available =="
python3 - <<'PY'import importlib.utilfor name in ("boto3","botocore"): if importlib.util.find_spec(name): print(f"{name} installed") else: print(f"{name} NOT installed")try: import boto3 print("boto3 version:", getattr(boto3, "__version__", "unknown")) import botocore print("botocore version:", getattr(botocore, "__version__", "unknown")) service_model = boto3.Session().get_available_services() print("'s3' available:", bool("s3" in boto3.Session().get_available_services())) import subprocess result = subprocess.run(["python3", "-c", "import boto3; print(boto3.client('s3', region_name='us-east-1')._service_model.operation_model('DeleteObjects').output_shape.members if hasattr(boto3.client('s3', region_name='us-east-1'), '_service_model') else 'no output shape')"], cwd="/tmp", text=True, capture_output=True, timeout=20) print("output probe return:", result.returncode) print(result.stdout.strip()) print(result.stderr.strip())except Exception as e: print("boto3 probe failed:", type(e).__name__, str(e))PYRepository: METResearchGroup/lab_data_integrations_interface
Length of output: 5549
🌐 Web query:
AWS S3 DeleteObjects API response Errors array Quiet delete_objects boto3
💡 Result:
The S3 DeleteObjects API provides two response modes, controlled by the Quiet parameter within the Delete element of your request [1][2][3]. Verbose Mode (Default, Quiet: False) When Quiet is set to False (or omitted), the API operates in verbose mode [1]. The response includes a Deleted array containing details for every object that the operation successfully deleted [1][2][3]. If any deletions fail, the response also includes an Errors array containing details about those specific failures [1][2][3]. Quiet Mode (Quiet: True) When Quiet is set to True, the API minimizes the response body [1][2]. In this mode, the response does not return any information about successfully deleted objects [1][2]. The response will only contain an Errors array if the operation encounters failures for any of the specified objects [1][2][3]. If all deletions are successful, the Errors array will not be present in the response [1][2]. Important Considerations for Boto3 - Response Handling: In Boto3, the delete_objects method returns a dictionary [2][3]. When Quiet=True, you should check for the presence of the Errors key in this dictionary to determine if any operations failed, as the Deleted key will be absent [2][3]. - Non-Existent Keys: Note that if you attempt to delete an object that does not exist, the S3 DeleteObjects API generally treats this as a successful deletion and returns a 200 OK status [4]. Consequently, such operations will not appear in the Errors array even if Quiet=True [4]. - Exceptions: Boto3 typically only raises an exception if the entire API call fails (e.g., a non-200 HTTP status code) [4]. Partial failures, such as permission issues for specific objects, are returned within the Errors array in the response body, not as raised exceptions [1][4].
Citations:
- 1: https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjects.html
- 2: https://docs.aws.amazon.com/boto3/latest/reference/services/s3/bucket/delete_objects.html
- 3: https://docs.aws.amazon.com/boto3/latest/reference/services/s3/client/delete_objects.html
- 4: Boto3 delete_objects response does not match documented response boto/boto3#507
Count actual successful deletes before reporting orphan deletion metrics.
DeleteObjects with Quiet=True omits successful deletes but still returns an Errors list for keys that failed. Since this value is returned directly as orphans_deleted, any partial failure can overstate the cleanup report.
🤖 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 `@experimentation/iceberg/maintenance.py` around lines 147 - 157, Update
_delete_keys to use each delete_objects response and count only keys not present
in its Errors list, rather than assuming every requested chunk key was deleted.
Preserve batching and Quiet=True, and return the accumulated successful-delete
count for accurate orphans_deleted metrics.
| pyiceberg[glue,s3fs,sql-sqlite,pyiceberg-core]>=0.10.0 | ||
| pyarrow>=14.0.0 | ||
| boto3>=1.35.0 | ||
| websockets>=15.0.1 | ||
| pytest>=8.0.0 |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Pin exact versions for reproducible measurements.
All five dependencies use >= with no upper bound. For an experiment whose entire point is precise, comparable request/cost counts, an unpinned resolve could silently change Iceberg's write path (or botocore's retry/behavior) between runs without anyone noticing.
♻️ Proposed fix
-pyiceberg[glue,s3fs,sql-sqlite,pyiceberg-core]>=0.10.0-pyarrow>=14.0.0-boto3>=1.35.0-websockets>=15.0.1-pytest>=8.0.0+pyiceberg[glue,s3fs,sql-sqlite,pyiceberg-core]==0.10.0+pyarrow==14.0.0+boto3==1.35.0+websockets==15.0.1+pytest==8.0.0📝 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[glue,s3fs,sql-sqlite,pyiceberg-core]>=0.10.0 | |
| pyarrow>=14.0.0 | |
| boto3>=1.35.0 | |
| websockets>=15.0.1 | |
| pytest>=8.0.0 | |
| pyiceberg[glue,s3fs,sql-sqlite,pyiceberg-core]==0.10.0 | |
| pyarrow==14.0.0 | |
| boto3==1.35.0 | |
| websockets==15.0.1 | |
| pytest==8.0.0 |
🤖 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 `@experimentation/iceberg/requirements.txt` around lines 11 - 15, Update the
five dependencies in requirements.txt—pyiceberg, pyarrow, boto3, websockets, and
pytest—to exact pinned versions instead of lower-bound constraints, preserving
the currently specified minimum versions unless the experiment’s intended
baseline defines different exact versions.
| @dataclass | ||
| class PhaseStats: | ||
| """Aggregates for a single phase, assembled by :meth:`Meter.summarize`.""" | ||
| phase: str | ||
| wall_seconds: float = 0.0 | ||
| calls: int = 0 | ||
| attempts: int = 0 | ||
| request_bytes: int = 0 | ||
| response_bytes: int = 0 | ||
| by_tier: dict[str, int] = field(default_factory=lambda: defaultdict(int)) | ||
| by_operation: dict[str, int] = field(default_factory=lambda: defaultdict(int)) | ||
| by_key_class: dict[str, int] = field(default_factory=lambda: defaultdict(int)) | ||
| # key_class -> {operation -> count}, the table that actually explains cost. | ||
| by_key_class_operation: dict[str, dict[str, int]] = field(default_factory=dict) | ||
| latencies_ms: list[float] = field(default_factory=list) | ||
| @property | ||
| def cost_usd(self) -> float: | ||
| return ( | ||
| self.by_tier["put"] * constants.COST_PER_PUT_REQUEST | ||
| + self.by_tier["get"] * constants.COST_PER_GET_REQUEST | ||
| + self.by_tier["delete"] * constants.COST_PER_DELETE_REQUEST | ||
| + self.by_tier["glue"] * constants.COST_PER_GLUE_REQUEST | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
cost_usd undercounts retried requests.
AWS bills every HTTP attempt, but cost_usd is computed from by_tier, which is only incremented once per logical call (in _on_after_call). _on_before_send counts attempts per phase but discards model/tier entirely, so a retried PutObject is billed by AWS twice while this meter's cost model counts it once. In the committed sample run attempts == calls everywhere, so today's headline numbers aren't affected — but any run that hits throttling/transient retries will silently understate the real dollar cost this experiment is built to measure.
Consider threading tier/operation through _on_before_send (via request.context, which carries the same dict _on_before_call populated) so retried attempts are costed too, or at minimum note the limitation in the report footer.
Also applies to: 251-255
🤖 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 `@experimentation/iceberg/s3_meter.py` around lines 153 - 177, Update the
attempt accounting in _on_before_send and PhaseStats.cost_usd so each HTTP
attempt contributes its request tier cost, including retries, rather than
relying only on by_tier counts from _on_after_call. Propagate the tier or
operation through request.context, using the context populated by
_on_before_call, and preserve logical-call metrics separately from attempt-based
billing.
| "tombstones_dropped": 1169, | ||
| "tombstone_pct": 1.3476747135182494, | ||
| "redelivered_duplicates": 0, | ||
| "lifecycle_collapses": 623 |
There was a problem hiding this comment.
lifecycle_collapses counts distinct AT events (same uri, different cid) discarded when compaction keeps only the latest row per URI — real lifecycle history (create→update, create→delete), not identical stream redeliveries.
Derived in _collapse_to_latest in experimentation/iceberg/maintenance.py:
Sort by (uri ASC, ingested_at DESC), keep newest per uri
distinct_events = unique (uri, cid) pairs
lifecycle_collapses = distinct_events − unique_uris
Sibling metric: redelivered_duplicates = total_rows − distinct_events (same uri and cid).
| "bytes_after": 10237942, | ||
| "rows_before": 86742, | ||
| "rows_after": 84950, | ||
| "tombstones_dropped": 1169, |
There was a problem hiding this comment.
tombstones_dropped is the count of operation == "delete" rows removed during compaction, after each URI is already collapsed to its latest row.
Derived in _drop_tombstones in experimentation/iceberg/maintenance.py, called from compact_table:
_collapse_to_latest — keep newest row per uri
Filter out operation == "delete"
tombstones_dropped = rows before that filter − rows after
Prices the four S3-facing operations of an Iceberg ingest pipeline fed by the Bluesky Jetstream firehose. Every number here is counted, not estimated — the experiment meters every individual S3 and Glue API call and attributes it to a phase.
How the experiment is set up
Capture and replay are separate
capture.pystreams Jetstream for 10 minutes and writes raw events to a local gzipped JSONL file.run_experiment.pyreplays that file.They're split because a 10-minute firehose capture is never reproducible. Recording once and replaying means every write-path variant (flush interval, partition spec, compaction strategy) is measured against byte-identical input. There's a test asserting batching is deterministic across replays.
The measured capture: 132,192 events in 610s (~217/s) — 19,440 posts, 86,742 likes, 13,666 reposts, 11,977 follows.
What one run does
A run is one invocation against one capture. It produces 10 flush batches.
A batch is a 60s window of stream time, cut on the broker's
time_usrather than wall-clock, so boundaries are a property of the data. Each batch holds rows for all four record types, and all four flush together.Per batch, for each of the 4 tables, one
table.append()— and every append is a full Iceberg commit:So one flush = ~16-18 S3 objects, of which only ~4 carry data. Across 10 batches: 40 commits, 50 data files, 183 warehouse objects.
Each batch is also written to a raw-Parquet baseline under a separate prefix — same rows, same encoder, same partition layout, no table format. That's the control.
After all 10 batches
Compaction (
compact_table) — scan the table, collapse each AT-URI to its latest row, drop delete tombstones, rewrite as one file per partition viaoverwrite(). PyIceberg 0.11 has norewrite_data_files, so this is scan → collapse → overwrite, which is delete-all + append in one commit.Metadata cleaning (
expire_snapshots+sweep_orphans) — expire every snapshot but the current one, then list the table prefix and delete objects no live metadata references. The orphan sweep is separate because expiry alone doesn't remove everything.How metering works
PyIceberg's default
PyArrowFileIOdrives a C++ S3 client Python can't intercept. The catalog pinspy-io-impl=pyiceberg.io.fsspec.FsspecFileIO, routing through s3fs → aiobotocore → botocore, wheres3_meter.Meterhooksbefore-call/after-calland counts every request by phase, object class, and billing tier.Two things this gets right that are easy to miss:
metadata_locationis a GlueUpdateTableat $1/100k, not $5/100k. Metering only S3 misses the commit path entirely.Content-Lengthatbefore-call, and large uploads switch toaws-chunkedwhich carries none at all.Findings
Iceberg costs 4.2x the writes of raw Parquet
Bytes are nearly identical — the metadata tree is small. The cost is entirely in request count. The 11.4x on total calls is dominated by cheap
HeadObject/GetObjecton metadata; the 4.2x on PUTs is what actually shows up on the bill.Per phase:
raw_writeiceberg_appendcompact_dedupexpire_metadataExtrapolated at this event rate: ~$0.48/day, ~$14.43/month, split roughly 2:1 ingest to maintenance.
Posts writes double the data files of every other table
created_atIceberg data files cannot span partitions. A single row with a different partition value forces an entire extra data file in that commit. Posts gets ~7 backdated rows per batch — scheduled posts, offline drafts, skewed client clocks — so 9 of 10 batches write two files instead of one. Post-compaction that partition is a 12 KB file holding 75 rows next to a 1.6 MB file holding 19,365.
This is not a day-boundary effect. The capture ran 07:02–07:13 UTC, seven hours after midnight — under ingest-time partitioning it would produce zero cross-day files. And
created_atis 47.9% out-of-order relative to arrival, versustime_uswhich had 0 inversions in 132,192 events. The backdated values are scattered across all of yesterday (08:31, 12:59, 15:00, 21:50, 23:08), not clustered near a boundary.Takeaway: with time-partitioning, cost scales with how many partitions a batch touches, not row count. A 0.4% long tail produced a 90% increase in data files. Partitioning on
ingested_atwould make one-file-per-table-per-flush structurally guaranteed, at the cost of making "posts authored on day X" a multi-partition scan.There were zero duplicates in the stream
Of 1,052 URIs appearing more than once, 1,042 are
create→deleteand every one has a differentcid. Not a single redelivered event in 132,192.So the thing usually called "deduplication" here is really lifecycle collapse — a record's history being materialised into current state. The two are now counted separately, because only redelivery is a stream defect.
Tombstones are a large fraction of some tables
A
deletecommit carries no record body — you learn that a follow was deleted, not who was unfollowed. It's a pointer, meaningful only if you hold the create it points at.They're dropped at compaction, not ingest. Ordering matters: the tombstone must survive long enough to cancel its create during collapse. Dropping at ingest would leave stale creates in the table forever.
Known limitation: this only works when create and delete land in the same compaction window. A record created six months ago and deleted today puts the tombstone in today's partition while the create sits in January's — collapse never sees them together, and the deletion is silently lost. Production would need equality deletes or
MERGE INTO.The
createdAtskew rule is 91% one bot7,103 rows fall back to ingest-time partitioning, but they split very differently than the raw number suggests:
deleteevents with no record body — structural, not a data problemOf those 445, 406 come from a single account stamping posts with Jan-1 of 2011, 2013, 2016, 2017 — an archive-import bot whose timestamps are genuinely correct. The skew rule discards real authoring dates to stop one bot opening a daily partition per historical date. That's a deliberate correctness-for-file-count trade, now documented as such rather than labelled "malformed data".
Two bugs found while building this
Field-ID renumbering caused silent NULLs.
create_tablerenumbers schema field IDs contiguously from 1. Declared IDs started at 20; the catalog assigned 11. Writes stamped Parquet footers with 20, reads looked for 11, found nothing, returned NULL — no error at any layer.subject_uriandsubject_cidwere 100% NULL in the likes and reposts tables. Fixed by renumbering and by building Arrow fromtable.schema()rather than the declared schema, so the two can't drift. Regression test asserts contiguity.This also explains an earlier bogus "write amplification 0.66x" — Iceberg appeared to write less than the control because two columns were being lost.
Byte metering read 0. botocore sets no
Content-Lengthon the request dict atbefore-call._body_sizenow measures theBytesIOand restores its position.Caveats
Testing
77 offline tests (no AWS, no network): metering accounting, key classification, billing tiers, thread-safety, Jetstream parsing, the
createdAtfallback rule, field-ID contiguity, flush-window batching, lifecycle collapse vs. redelivery, and tombstone dropping. Aconftest.pyskips collection under the root interpreter, which has no pyiceberg.ruff, ruff-format, complexipy, and vulture all pass; pyright is clean against the experiment interpreter.
Note on the isolated venv
pyiceberg pins
rich<15; this project requiresrich>=15. They cannot share a resolution, so the experiment has its own venv viarequirements.txt, andexperimentation/icebergis added to the root pyright exclude (it legitimately type-checks against a different interpreter).🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Documentation
Tests