feat(java): add seekable ranged object streams - #170
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (7)
🚧 Files skipped from review as they are similar to previous changes (4)
📝 WalkthroughWalkthroughThe Java project moves from Maven to Gradle with Java 25 and Maven Central publishing. Bucket APIs now support POSIX prefixes, seekable reads, streamed commits, validation, batched deletion, and lifecycle ownership. Memory and S3 backends implement the new contracts. Path handling, ranged I/O, Parquet access, functional tests, and publishing workflows are expanded. Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant ParquetReader
participant BucketParquetDataSource
participant SeekableInputStream
participant S3Bucket
participant S3
ParquetReader->>BucketParquetDataSource: request footer or row-group range
BucketParquetDataSource->>SeekableInputStream: seek and read
SeekableInputStream->>S3Bucket: request object range
S3Bucket->>S3: conditional GetObject
S3-->>ParquetReader: return requested bytes
Possibly related PRs
Warning Billing warning: we have not been able to collect payment for this subscription for more than 72 hours. Please update the payment method or pay any pending invoices in Billing to avoid service interruption. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (3)
java/src/main/java/com/esamtrade/bucketbase/IBucket.java (1)
159-167: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win
copyPrefixbuffers each object fully into RAM — stream it instead. Right above this you builtcopyObjectFrom, which streams source→dest without holding the whole object. But the parallel copy path doesgetObject(srcObj)(entire object into heap) thenputObject. Fire that off on N threads with a few large objects and heap usage goes vertical — exactly the failure mode we don't want on a copy that's supposed to scale. Reuse the streaming path you already wrote.As per coding guidelines: "Performance & Big-O Awareness: watch out for ... wasteful object creation in hot paths."♻️ Stream each object rather than materializing it
futures.add(executor.submit(() -> { String name = dstPrefixStr + srcObj.toString().substring(srcPrefixLen); if (name.startsWith("/")) { name = name.substring(1); } - dstBucket.putObject(PurePosixPath.from(name), getObject(srcObj)); + dstBucket.copyObjectFrom(this, srcObj, PurePosixPath.from(name)); return null; }));🤖 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 `@java/src/main/java/com/esamtrade/bucketbase/IBucket.java` around lines 159 - 167, Update the parallel copy task in copyPrefix to reuse the existing copyObjectFrom streaming path instead of calling getObject(srcObj) and passing a fully materialized object to putObject. Preserve the current destination-name construction and parallel executor behavior while ensuring each source object streams directly to its destination.Source: Coding guidelines
java/src/main/java/com/esamtrade/bucketbase/StreamPipe.java (1)
23-37: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winHand-rolled monitor works, but consider a BlockingQueue-based design.
The class documents choosing manual
wait/notifyAlloverjava.util.concurrentprimitives specifically to supportabort()waking both sides with a failure. That's a defensible reason, but per the project guideline to prefer high-level concurrency abstractions over synchronized blocks where they suffice, anArrayBlockingQueue<byte[]>combined with the existingAtomicReference<Throwable>pattern (already used foruploadFailureinObjectWriter) plusThread.interrupt()for wake-on-abort could shrink this to much less custom monitor code. Not blocking — the abort semantics genuinely complicate a drop-in swap — just worth a look if this class needs future maintenance.As per coding guidelines, "Concurrency & performance: use thread-safe collections and immutables; avoid synchronized blocks where high-level abstractions suffice."
🤖 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 `@java/src/main/java/com/esamtrade/bucketbase/StreamPipe.java` around lines 23 - 37, Consider refactoring StreamPipe’s custom lock, queue, and wait/notify coordination to use a bounded ArrayBlockingQueue<byte[]> with an AtomicReference<Throwable> for failure state and thread interruption to wake blocked readers and writers during abort(). Preserve the existing back-pressure, EOF-after-drain, and abort failure semantics; if the abstraction cannot support these semantics cleanly, leave the monitor implementation unchanged.Source: Coding guidelines
java/src/main/java/com/esamtrade/bucketbase/MemoryBucket.java (1)
72-92: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueNice shallow-listing logic, but the
HashSetmakes ordering a roll of the dice.Physics doesn't care about hash bucket order, and neither should your API consumers.
prefixesSetis aHashSet, soshallowListObjectsreturns prefixes in an unspecified order, whileS3BucketSDKv1.shallowListObjectspreserves server-returned order via aList. Not failing today because the tests only check single-element/size expectations, but it's a landmine for anyone diffing behavior across backends.♻️ Deterministic ordering fix
- Set<PurePosixPrefix> prefixesSet = new HashSet<>(); + Set<PurePosixPrefix> prefixesSet = new LinkedHashSet<>();🤖 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 `@java/src/main/java/com/esamtrade/bucketbase/MemoryBucket.java` around lines 72 - 92, Update shallowListObjects so prefixes retain deterministic discovery order instead of using HashSet iteration order. Replace prefixesSet with an insertion-ordered collection while preserving duplicate elimination, and continue constructing the ShallowListing with the resulting ordered prefixes list.
🤖 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 `@java/gradle/wrapper/gradle-wrapper.properties`:
- Line 3: Add the Gradle Wrapper property distributionSha256Sum alongside
distributionUrl in gradle-wrapper.properties, using the official SHA-256
checksum for gradle-9.6.1-bin.zip:
9c0f7faeeb306cb14e4279a3e084ca6b596894089a0638e68a07c945a32c9e14.
In `@java/src/main/java/com/esamtrade/bucketbase/ObjectWriter.java`:
- Around line 86-132: Update awaitUploader() to cancel the uploader when
join(timeoutMillis) expires, interrupting or otherwise stopping the upload on
timeout before throwing its IOException. Preserve the existing timeout and
interruption errors, and ensure commit() and close() use this best-effort
cancellation so the uploader cannot continue normally after either operation has
failed or aborted.
In `@java/src/main/java/com/esamtrade/bucketbase/S3BucketSDKv1.java`:
- Around line 92-97: Update putObjectStream in S3BucketSDKv1 to set
ObjectMetadata content length before calling s3Client.putObject whenever the
stream size is known. For unknown-length streams, stage through a temporary file
or use multipart upload support so the AWS SDK does not buffer the entire stream
in memory.
In `@java/src/main/java/com/esamtrade/bucketbase/StreamPipe.java`:
- Around line 129-153: Update PipeOutputStream.write(byte[] src, int off, int
len) to split large writes into bounded chunks before enqueueing, so each chunk
is limited by the stream’s configured capacity and back-pressure accounts for
all buffered bytes rather than the caller’s write-call count. Preserve range
validation, zero-length behavior, buffer copying, closure/failure checks, and
synchronization while ensuring a large write cannot bypass the queue limit.
---
Nitpick comments:
In `@java/src/main/java/com/esamtrade/bucketbase/IBucket.java`:
- Around line 159-167: Update the parallel copy task in copyPrefix to reuse the
existing copyObjectFrom streaming path instead of calling getObject(srcObj) and
passing a fully materialized object to putObject. Preserve the current
destination-name construction and parallel executor behavior while ensuring each
source object streams directly to its destination.
In `@java/src/main/java/com/esamtrade/bucketbase/MemoryBucket.java`:
- Around line 72-92: Update shallowListObjects so prefixes retain deterministic
discovery order instead of using HashSet iteration order. Replace prefixesSet
with an insertion-ordered collection while preserving duplicate elimination, and
continue constructing the ShallowListing with the resulting ordered prefixes
list.
In `@java/src/main/java/com/esamtrade/bucketbase/StreamPipe.java`:
- Around line 23-37: Consider refactoring StreamPipe’s custom lock, queue, and
wait/notify coordination to use a bounded ArrayBlockingQueue<byte[]> with an
AtomicReference<Throwable> for failure state and thread interruption to wake
blocked readers and writers during abort(). Preserve the existing back-pressure,
EOF-after-drain, and abort failure semantics; if the abstraction cannot support
these semantics cleanly, leave the monitor implementation unchanged.
🪄 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: CHILL
Plan: Pro Plus
Run ID: 26a1c5ee-8eea-484a-86ad-6c0ffe6f350a
⛔ Files ignored due to path filters (1)
java/gradle/wrapper/gradle-wrapper.jaris excluded by!**/*.jar
📒 Files selected for processing (43)
.gitignorejava/PUBLISHING.mdjava/build.gradlejava/gradle/wrapper/gradle-wrapper.propertiesjava/gradlewjava/gradlew.batjava/pom.xmljava/settings.gradlejava/src/main/java/README.mdjava/src/main/java/com/esamtrade/bucketbase/AbstractAppendOnlySynchronizedBucket.javajava/src/main/java/com/esamtrade/bucketbase/BaseBucket.javajava/src/main/java/com/esamtrade/bucketbase/DeleteError.javajava/src/main/java/com/esamtrade/bucketbase/IBucket.javajava/src/main/java/com/esamtrade/bucketbase/MemoryBucket.javajava/src/main/java/com/esamtrade/bucketbase/ObjectStream.javajava/src/main/java/com/esamtrade/bucketbase/ObjectWriter.javajava/src/main/java/com/esamtrade/bucketbase/PurePosixPath.javajava/src/main/java/com/esamtrade/bucketbase/PurePosixPrefix.javajava/src/main/java/com/esamtrade/bucketbase/RangeSeekableInputStream.javajava/src/main/java/com/esamtrade/bucketbase/S3Bucket.javajava/src/main/java/com/esamtrade/bucketbase/S3BucketSDKv1.javajava/src/main/java/com/esamtrade/bucketbase/S3Names.javajava/src/main/java/com/esamtrade/bucketbase/SeekableInputStream.javajava/src/main/java/com/esamtrade/bucketbase/ShallowListing.javajava/src/main/java/com/esamtrade/bucketbase/StreamPipe.javajava/src/test/java/com/esamtrade/bucketbase/BucketHardeningTest.javajava/src/test/java/com/esamtrade/bucketbase/BucketParquetDataSource.javajava/src/test/java/com/esamtrade/bucketbase/IBucketTester.javajava/src/test/java/com/esamtrade/bucketbase/MemoryBucketTest.javajava/src/test/java/com/esamtrade/bucketbase/MinioParquetFunctionalTest.javajava/src/test/java/com/esamtrade/bucketbase/ParquetReadTest.javajava/src/test/java/com/esamtrade/bucketbase/ParquetTestSupport.javajava/src/test/java/com/esamtrade/bucketbase/ParquetWriteTest.javajava/src/test/java/com/esamtrade/bucketbase/PurePosixPathTest.javajava/src/test/java/com/esamtrade/bucketbase/PurePosixPrefixTest.javajava/src/test/java/com/esamtrade/bucketbase/RangeSeekableInputStreamTest.javajava/src/test/java/com/esamtrade/bucketbase/S3BucketRangeTest.javajava/src/test/java/com/esamtrade/bucketbase/S3BucketSDKv1RangeTest.javajava/src/test/java/com/esamtrade/bucketbase/S3BucketSDKv1Test.javajava/src/test/java/com/esamtrade/bucketbase/S3BucketTest.javajava/src/test/java/com/esamtrade/bucketbase/S3RemoveAndCloseTest.javajava/src/test/java/com/esamtrade/bucketbase/TestConfig.javapython/tests/test_pure_posix_path_stdlib_contract.py
💤 Files with no reviewable changes (4)
- java/src/main/java/com/esamtrade/bucketbase/ObjectStream.java
- java/src/main/java/com/esamtrade/bucketbase/BaseBucket.java
- java/src/main/java/README.md
- java/pom.xml
| @@ -0,0 +1,9 @@ | |||
| distributionBase=GRADLE_USER_HOME | |||
| distributionPath=wrapper/dists | |||
| distributionUrl=https\://services.gradle.org/distributions/gradle-9.6.1-bin.zip | |||
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
https://services.gradle.org/distributions/gradle-9.6.1-bin.zip checksum sha256
💡 Result:
The SHA-256 checksum for the file gradle-9.6.1-bin.zip is 9c0f7faeeb306cb14e4279a3e084ca6b596894089a0638e68a07c945a32c9e14 [1][2][3]. This value is officially published by the Gradle team on their release checksums reference page and the distribution release assets [1][2].
Citations:
- 1: https://gradle.org/release-checksums/
- 2: https://github.com/gradle/gradle-distributions/releases/tag/v9.6.1
- 3: https://github.com/gradle/gradle-distributions/releases
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Wrapper properties:"
if [ -f java/gradle/wrapper/gradle-wrapper.properties ]; then
cat -n java/gradle/wrapper/gradle-wrapper.properties
else
echo "missing java/gradle/wrapper/gradle-wrapper.properties"
fiRepository: eSAMTrade/bucketbase
Length of output: 522
Pin the Gradle distribution checksum, no false positives.
validateDistributionUrl=true only checks the URL; add distributionSha256Sum so Wrapper validation can reject tampered bytes during download. The official checksum for gradle-9.6.1-bin.zip is 9c0f7faeeb306cb14e4279a3e084ca6b596894089a0638e68a07c945a32c9e14.
Proposed fix
distributionUrl=https\://services.gradle.org/distributions/gradle-9.6.1-bin.zip
+distributionSha256Sum=9c0f7faeeb306cb14e4279a3e084ca6b596894089a0638e68a07c945a32c9e14📝 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.
| distributionUrl=https\://services.gradle.org/distributions/gradle-9.6.1-bin.zip | |
| distributionUrl=https\://services.gradle.org/distributions/gradle-9.6.1-bin.zip | |
| distributionSha256Sum=9c0f7faeeb306cb14e4279a3e084ca6b596894089a0638e68a07c945a32c9e14 |
🤖 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 `@java/gradle/wrapper/gradle-wrapper.properties` at line 3, Add the Gradle
Wrapper property distributionSha256Sum alongside distributionUrl in
gradle-wrapper.properties, using the official SHA-256 checksum for
gradle-9.6.1-bin.zip:
9c0f7faeeb306cb14e4279a3e084ca6b596894089a0638e68a07c945a32c9e14.
1 participant
Title
<type>:<short summary>Ticket / Task
<URL>What & Why
"Builds X so that Y can Z" – two sentences max.
Scope
Checklist (self-review)
See Section 5.
Risk / Impact
Screenshots / Logs / Benchmarks
(only if relevant)
Author Self-Review Checklist
pytest -q/mvn testclean locally.git revertwithout dependency hell.python/scripts/run_mypy.batlocally with no critical errorsResources: