laskoviymishka commented on code in PR #3236:
URL: https://github.com/apache/iceberg-rust/pull/3236#discussion_r4024352299
##########
crates/iceberg/src/encryption/io.rs:
##########
@@ -138,8 +153,8 @@ impl EncryptedOutputFile {
)))
}
- /// Write bytes to file (transparently encrypted).
- pub async fn write(&self, bs: Bytes) -> Result<()> {
+ /// Write bytes to the file and return its encrypted size.
+ pub async fn write(&self, bs: Bytes) -> Result<FileMetadata> {
Review Comment:
`write()` returns the ciphertext size, but `self.key_metadata` is never
updated with it. So the caller has to remember to
`.clone().with_file_length(file_metadata.size)` before encoding — all four
current call sites do, but the pre-PR idiom `output.key_metadata().encode()`
still compiles and silently produces key metadata with `file_length = None`,
which then fails at read time with a `DataInvalid` that gives no hint the write
path forgot the length.
That's a sharp edge for a public API. I'd either have `write()` hand back
the updated `StandardKeyMetadata` alongside the `FileMetadata` (a small
`WrittenFile { file_metadata, key_metadata }`), or update `self.key_metadata`
in place so `key_metadata()` reflects the length after the write. Either
removes the boilerplate and the footgun.
wdyt?
##########
crates/iceberg/src/encryption/io.rs:
##########
@@ -74,14 +73,30 @@ impl EncryptedInputFile {
/// Creates a reader that transparently decrypts on each read.
pub async fn reader(&self) -> Result<Box<dyn FileRead>> {
- let raw_meta = self.inner.metadata().await?;
+ let encrypted_length = self.encrypted_length()?;
let raw_reader = self.inner.reader().await?;
let cipher = build_cipher(&self.key_metadata)?;
let aad_prefix: Box<[u8]> =
self.key_metadata.aad_prefix().unwrap_or_default().into();
- let decrypting = AesGcmFileRead::new(raw_reader, cipher, aad_prefix,
raw_meta.size)?;
+ let decrypting = AesGcmFileRead::new(raw_reader, cipher, aad_prefix,
encrypted_length)?;
Ok(Box::new(decrypting))
}
+ fn encrypted_length(&self) -> Result<u64> {
Review Comment:
This makes `file_length` mandatory on every read path — `metadata()`,
`reader()`, and `read()` all route through here now, and there's no stat
fallback. So every encrypted manifest, manifest list, and puffin file the
earlier PRs in this series already wrote (without `file_length` in the key
metadata) becomes permanently unreadable by iceberg-rust, not just by Java.
I get that trusting the declared length is the whole point of truncation
proofing, so I don't think we can fall back silently. But I'd want us to pick
one explicitly: either fall back to `self.inner.metadata()` when `file_length`
is `None` and warn that truncation protection is degraded (matches PyIceberg,
keeps old tables readable), or accept the hard break and call it out
prominently in the changelog as a one-time migration for the encryption feature.
If the feature is explicitly no-stability-yet, the second is fine — I just
don't want it to be a silent break. wdyt?
##########
crates/iceberg/src/spec/manifest/writer.rs:
##########
@@ -508,14 +508,22 @@ impl ManifestWriter {
}
let content = avro_writer.into_inner()?;
- let length = content.len();
let mut writer = self.writer_future.await?;
writer.write(Bytes::from(content)).await?;
- writer.close().await?;
+ let file_metadata = writer.close().await?;
+ let key_metadata = self
+ .key_metadata
+ .map(|metadata| {
+ metadata
+ .with_file_length(file_metadata.size)
+ .encode()
+ .map(|bytes| bytes.into_vec())
+ })
+ .transpose()?;
Ok(ManifestFile {
manifest_path: self.location,
- manifest_length: length as i64,
+ manifest_length: file_metadata.size.try_into()?,
Review Comment:
Good catch switching this to the on-disk size — for encrypted manifests the
old `content.len()` was the plaintext Avro length, which is actually a spec
violation (field 501 is the on-disk file length), so this is more correct, not
just different.
Since it's a deliberate behavior change that's invisible for unencrypted
files and only shows up encrypted, I'd add a one-line comment noting the value
is now the ciphertext/on-disk length, plus a small regression test for an
unencrypted manifest asserting `manifest_length == plaintext_avro_len` so
nobody "fixes" it back later.
##########
crates/iceberg/src/encryption/io.rs:
##########
@@ -215,8 +236,134 @@ mod tests {
"encrypted file should be larger than plaintext (header + nonce +
tag)"
);
- let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(),
key_metadata());
+ // A missing path proves the size comes from the key metadata rather
than a stat call.
+ let input = EncryptedInputFile::new(
+ fileio.new_input("memory:///does-not-exist").unwrap(),
+ key_metadata().with_file_length(file_metadata.size),
+ );
let meta = input.metadata().await.unwrap();
assert_eq!(meta.size, plaintext.len() as u64);
}
+
+ #[tokio::test]
+ async fn test_missing_file_length_is_rejected() {
+ let fileio = FileIO::new_with_memory();
+ let path = "memory:///test/missing_length.bin";
+ let output =
EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata());
+ output.write(Bytes::from_static(b"data")).await.unwrap();
+ let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(),
key_metadata());
+
+ for err in [
+ input.metadata().await.err().unwrap(),
+ input.reader().await.err().unwrap(),
+ input.read().await.unwrap_err(),
+ ] {
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(
+ err.to_string()
+ .contains("missing the encrypted file length")
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn test_invalid_file_length_is_rejected() {
+ let fileio = FileIO::new_with_memory();
+ for length in [
+ 0,
+ u64::from(GCM_STREAM_HEADER_LENGTH),
+ u64::from(MIN_STREAM_LENGTH - 1),
+ ] {
+ let input = EncryptedInputFile::new(
+ fileio
+ .new_input("memory:///test/invalid_length.bin")
+ .unwrap(),
+ key_metadata().with_file_length(length),
+ );
+ assert_eq!(
+ input.metadata().await.err().unwrap().kind(),
+ ErrorKind::DataInvalid
+ );
+ assert_eq!(
+ input.reader().await.err().unwrap().kind(),
+ ErrorKind::DataInvalid
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn test_truncated_file_is_rejected() {
Review Comment:
These truncation tests all cover the file-shorter-than-declared case with
correct key metadata. The other half of the tamper check — key metadata
claiming a *larger* `file_length` than what's actually on disk — isn't
exercised anywhere.
Since catching an inflated declared length is exactly what the new
short-read check in `AesGcmFileRead::read` is for, I'd add a case with
`with_file_length(file_metadata.size + CIPHER_BLOCK_SIZE)` asserting a
`DataInvalid`. That locks the contract in both directions.
##########
crates/iceberg/src/io/file_io.rs:
##########
@@ -375,10 +375,10 @@ pub trait FileWrite: Send + Unpin + 'static {
/// TODO: we can support writing non-contiguous bytes in the future.
async fn write(&mut self, bs: Bytes) -> Result<()>;
- /// Close file.
+ /// Close the file and return its stored size.
///
/// Calling close on closed file will generate an error.
- async fn close(&mut self) -> Result<()>;
+ async fn close(&mut self) -> Result<FileMetadata>;
Review Comment:
Flipping `close()` from `Result<()>` to `Result<FileMetadata>` on the public
`FileWrite` trait (plus `EncryptedOutputFile::write`,
`ManifestListWriter::close`, `PuffinWriter::close`) is a breaking change for
any downstream implementor or `Ok(())` matcher. That's fine for this series —
I'd just make sure the changelog lists all four signature changes so it isn't a
surprise.
While we're here, "stored size" is ambiguous for the encrypting wrappers,
where it's the ciphertext size (larger than what the caller wrote). Worth one
line saying that explicitly, since that distinction is the whole point of the
PR.
##########
crates/iceberg/src/transaction/snapshot.rs:
##########
@@ -492,7 +492,20 @@ impl<'a> SnapshotProducer<'a> {
manifest_list_writer.add_manifests(new_manifests.into_iter())?;
let writer_next_row_id = manifest_list_writer.next_row_id();
- manifest_list_writer.close().await?;
+ let file_metadata = manifest_list_writer.close().await?;
+ let encryption_key_id = if let Some(key_metadata) = key_metadata {
+ Some(
+ self.table
+ .encryption_manager()
+ .expect("Encryption manager must be present when key
metadata exists")
Review Comment:
This `.expect()` panics if the invariant ever breaks, and it's only sound
because `key_metadata.is_some()` implies the manager is present — an invariant
enforced by construction up in the match, not by the types. A future reorder
here turns into an unrecoverable panic in library code.
Since we already matched on `encryption_manager()` above, I'd capture the
`Arc` in that first arm and carry it down alongside `key_metadata`, so there's
no second lookup and no `expect`. wdyt?
##########
crates/iceberg/src/encryption/stream.rs:
##########
@@ -1002,8 +1036,10 @@ mod tests {
Ok(())
}
- async fn close(&mut self) -> Result<()> {
- Ok(())
+ async fn close(&mut self) -> Result<FileMetadata> {
Review Comment:
`SharedMemoryWrite::close` now returns the buffer length, but the
`write_through_ags1` helper still does `writer.close().await.unwrap()` and
drops it, so none of the ~dozen tests using it assert that
`AesGcmFileWrite::close`'s reported size matches the actual ciphertext length.
That's the primary AGS1 close path, and it's the value the whole PR hangs on.
I'd have the helper return `(Vec<u8>, FileMetadata)` and assert
`metadata.size == encrypted.len()` in at least one caller. Cheap, and it pins
the streaming path directly.
##########
crates/iceberg/src/encryption/stream.rs:
##########
@@ -990,8 +1024,8 @@ mod tests {
Ok(())
}
- async fn close(&mut self) -> Result<()> {
- Ok(())
+ async fn close(&mut self) -> Result<FileMetadata> {
+ unreachable!()
Review Comment:
`unreachable!()` compiles here (the never type coerces to `FileMetadata`),
but it's only correct as long as `AesGcmFileWrite::close` returns early from
the poisoned-state guard before it ever reaches `self.inner.close()`. Nothing
asserts that ordering, and if it changes this panics with no message (swallowed
without `--nocapture`).
I'd return a real `Err(Error::new(ErrorKind::Unexpected,
"FailingFileWrite::close called unexpectedly"))` instead — same effect for the
test, but it fails loudly and stays a proper `FileWrite` impl.
##########
crates/iceberg/src/puffin/writer.rs:
##########
@@ -117,12 +117,11 @@ impl PuffinWriter {
Ok(())
}
- /// Finalizes the Puffin file
- pub async fn close(mut self) -> Result<()> {
+ /// Finalizes the Puffin file and returns its stored size.
+ pub async fn close(mut self) -> Result<io::FileMetadata> {
Review Comment:
Small readability thing: `puffin::metadata::FileMetadata` and
`io::FileMetadata` are both in scope here, and the `io::` prefix is the only
thing keeping them apart. I'd alias the import — `use crate::io::FileMetadata
as IoFileMetadata;` — so a future edit can't quietly grab the wrong one.
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -635,21 +635,56 @@ impl FileRead for OpenDalReader {
}
/// Wrapper around `opendal::Writer` that implements `FileWrite`.
-pub(crate) struct OpenDalWriter(pub(crate) opendal::Writer);
+pub(crate) struct OpenDalWriter {
+ inner: opendal::Writer,
+ bytes_written: u64,
+}
+
+impl OpenDalWriter {
+ pub(crate) fn new(inner: opendal::Writer) -> Self {
+ Self {
+ inner,
+ bytes_written: 0,
+ }
+ }
+}
#[async_trait]
impl FileWrite for OpenDalWriter {
async fn write(&mut self, bs: Bytes) -> Result<()> {
- Ok(opendal::Writer::write(&mut self.0, bs)
+ let len = bs.len() as u64;
+ opendal::Writer::write(&mut self.inner, bs)
.await
- .map_err(from_opendal_error)?)
+ .map_err(from_opendal_error)?;
+ self.bytes_written += len;
+ Ok(())
}
- async fn close(&mut self) -> Result<()> {
- let _ = opendal::Writer::close(&mut self.0)
+ async fn close(&mut self) -> Result<FileMetadata> {
+ let metadata = opendal::Writer::close(&mut self.inner)
.await
.map_err(from_opendal_error)?;
- Ok(())
+
+ // `Metadata::content_length()` silently returns 0 when the service
did not report a
+ // size, and most object stores don't: S3 only populates it from the
`x-amz-object-size`
+ // response header, which general-purpose buckets never send. A bogus
0 here would be
+ // written into `manifest_length` and into the AGS1 `file_length` used
for truncation
+ // protection, making the file permanently unreadable, so trust our
own byte count and
+ // only use the service value to detect a genuine mismatch.
+ let reported_size = metadata.content_length();
+ if reported_size != 0 && reported_size != self.bytes_written {
Review Comment:
The comment above this is great — it explains exactly why we trust
`bytes_written` over `content_length()`. Given that reasoning, though, this
mismatch branch never actually fires on the common production path:
S3/GCS/Azure return 0, so `reported_size != 0` is false and we skip it. It only
triggers for in-memory operators, where the two are equal by construction.
So it reads like a cross-validation guarantee but protects nothing where
it'd matter. I'd either demote it to a `debug_assert!` or reword the intent to
"only validates when the store reports a size." Not blocking — I just don't
want a future reader to trust it as a real integrity check. wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
