v2.0 wave 1: resource & contract fixes (#201 #202 #203) - #205
v2.0 wave 1: resource & contract fixes (#201 #202 #203)#205tomasvotava wants to merge 5 commits into
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request enhances the Catalog with a spool_threshold_bytes setting for in-memory Avro serialization and hardens the AssetRepartitioner using an LRU cache for open writers to prevent file descriptor leaks. It also updates aiter_any to handle plain callables. Review feedback identifies a memory efficiency regression where large assets are fully buffered before threshold evaluation and points out a missing try...finally block for the BytesIO buffer during serialization.
The aiter_any signature advertises Callable[[], Iterable[T] | AsyncIterable[T]] but the body only handled isasyncgenfunction / isgeneratorfunction, so a plain lambda: [1, 2, 3] raised TypeError. Add a one-hop callable branch and corresponding tests. Closes #203
…nd-trip _serialize_asset_to_tempfile previously wrote to NamedTemporaryFile then store_asset_async reopened it by path - a full disk round-trip for even a 5 KB asset. Add a configurable spool_threshold_bytes (default 8 MiB); small payloads serialize to BytesIO and upload from the in-memory buffer; large payloads keep the disk path. fastavro only needs seek/tell, which BytesIO provides. Closes #201
…oner AssetRepartitioner.__enter__ clobbered self._state on re-entry (leaking any still-open writers) and _get_handler opened one NamedTemporaryFile per partition for the life of the run -- a 5-year hourly repartition means ~43k open fds. _get_handler also lacked the BaseException guard that _serialize_asset_to_tempfile uses. Add a bounded LRU of open writers (default 256, configurable via constructor), make __enter__ raise on re-entry instead of silently clobbering, and mirror the tempfile cleanup pattern around BaseException. On eviction the writer is flushed/closed but its NamedTemporaryFile(delete=False) path is retained; a later write to the same partition reopens it in 'a+b' so fastavro's appendable-container detection resumes the same avro file (verified via _is_appendable + round-trip test). Strategy chosen: streaming-reopen (rather than concat-on-close or in-memory buffering), since fastavro natively supports re-reading an existing container header and appending new blocks with the same sync marker -- no header duplication or partial-container concatenation needed. Closes #202 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Accumulate per-wave entries under "Unreleased — v2.0 (in progress)"; final v2.0 release PR will consolidate them into the dated section.
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces performance and resource management improvements to the data catalog. Key updates include a spooling threshold for Avro serialization to keep small payloads in memory and an LRU cache for open writers in AssetRepartitioner to bound file descriptor usage. Feedback identifies a memory efficiency concern where payloads are fully buffered before thresholding, suggests using standard file modes for consistency, and recommends hardening the partition cleanup logic to ensure all temporary files are removed even if an error occurs during the process.
There was a problem hiding this comment.
Code Review
This pull request introduces performance optimizations and stability fixes for the catalog and asyncutils modules. Key updates include a memory-spooling mechanism for small Avro payloads and an LRU-based writer management system in the AssetRepartitioner to prevent file descriptor leaks. Feedback from the review highlights a potential OOM risk where large assets are fully buffered in memory before size validation, suggesting the use of SpooledTemporaryFile instead. Additionally, a concern was raised regarding the repartitioning logic, which currently produces concatenated Avro containers with duplicate headers when reopening files, potentially impacting file size and compatibility.
…itioner cleanup Two follow-ups on the wave-1 review: - Replace the BytesIO+promote-to-disk pattern in _serialize_asset_to_tempfile with a single tempfile.SpooledTemporaryFile(max_size=spool_threshold_bytes). fastavro now writes into a buffer that stays in memory below the threshold and rolls to disk transparently during write — large payloads no longer get double-buffered in RAM before being copied to a NamedTemporaryFile. Caller path collapses to one branch: always upload from the open handle, close on done. - Isolate per-partition cleanup failures in AssetRepartitioner.__exit__ and _finalize_partitions so one bad writer.dump() or unlink() doesn't abandon the rest of the open writers and tempfiles. Add a test that monkeypatches writer.dump() to raise mid-loop and asserts the surviving partition still gets closed and its tempfile unlinked.
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces several improvements to the hyperion library. It updates asyncutils to support plain callables in aiter_any, optimizes the Catalog by adding a configurable spool_threshold_bytes to keep small Avro payloads in memory, and enhances AssetRepartitioner with a bounded LRU cache for open writers to prevent file descriptor leaks during large repartitioning tasks. The reviewer identified two potential blocking I/O operations in AssetRepartitioner that should be offloaded to asyncio.to_thread to prevent event loop blocking.
| path=entry.path.as_posix(), | ||
| partition_date=partition_date.isoformat(), | ||
| ) | ||
| upload_file = entry.path.open("rb") |
There was a problem hiding this comment.
The entry.path.open("rb") call is a blocking I/O operation. When called from an async function, this can block the event loop, which should be avoided. This operation should be run in a separate thread using asyncio.to_thread.
| upload_file = entry.path.open("rb") | |
| upload_file = await asyncio.to_thread(entry.path.open, "rb") |
| try: | ||
| await self.catalog._resolve_storage(asset).put_async(asset.get_path(), file) | ||
| finally: | ||
| file.close() |
There was a problem hiding this comment.
The file.close() call is a blocking I/O operation. When called from an async function, this can block the event loop, which should be avoided. This operation should be run in a separate thread using asyncio.to_thread.
| file.close() | |
| await asyncio.to_thread(file.close) |
1 participant
First wave of the v2.0 async overhaul (epic #193). Three independent, non-breaking issues bundled into one PR via per-issue worktrees. Targets the long-lived
v2.0branch; master stays on the 1.x patch track.Summary
fix(asyncutils): handle plain callables in aiter_any— the signature advertisedCallable[[], Iterable[T] | AsyncIterable[T]]but the body only handledisasyncgenfunction/isgeneratorfunction, so a plainlambda: [1, 2, 3]raisedTypeError. One-hop callable branch added.fix(catalog): bounded writer LRU + reentrancy guard in AssetRepartitioner—_get_handleropened oneNamedTemporaryFileper partition for the life of the run (5-year hourly = ~43k open fds).__enter__clobbered_stateon re-entry._get_handlerhad noBaseExceptionguard. Replaced unbounded state with boundedOrderedDictLRU (default 256, configurable viamax_open_writers); evicted writers reopen ina+band append (fastavro detects the existing container, reuses the sync marker, no header duplication);__enter__raisesRuntimeErroron re-entry; tempfile creation now mirrors the_serialize_asset_to_tempfileBaseException/cleanup pattern.perf(catalog): keep small avro payloads in memory—_serialize_asset_to_tempfilewrote to aNamedTemporaryFile, closed it, thenstore_asset_asyncreopened it by path. Newspool_threshold_bytesCatalog kwarg (default 8 MiB) — payloads ≤ threshold serialize to aBytesIOand upload from RAM; larger payloads keep the on-disk path. fastavro only needs seek/tell, whichBytesIOprovides.Each issue was implemented in its own worktree on
v2.0-wave-1-iss-{201,202,203}and merged intov2.0-wave-1here.Verification
poetry run pytest -q→ 641 passed, 0 failed on the bundled branch (4 newtest_aiter_callable_*, 4 newtest_serialize_spool_*, 4 newTestAssetRepartitionerHardening::test_repartitioner_*).poetry run pre-commit run --all-files→ ruff, mypy, detect-secrets, all hooks pass.poetry run lint-imports→ 5/5 DDD contracts kept.Test plan
Unreleased — v2.0 (in progress)/Wave 1spool_threshold_bytesis reasonable; profile-tune later if needed (per the issue note).Closes #201
Closes #202
Closes #203