mbutrovich commented on code in PR #3433:
URL: https://github.com/apache/iceberg-rust/pull/3433#discussion_r4259496649
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -1129,6 +1129,330 @@ mod tests {
assert!(result.is_none()); // no pos dels for file 3
}
+ /// Local filesystem storage that counts metadata requests.
+ #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+ struct MetadataCountingStorage {
+ #[serde(skip)]
+ calls: Arc<std::sync::atomic::AtomicUsize>,
+ /// Fail every metadata request with a retryable error.
+ #[serde(skip)]
+ fail_metadata: bool,
+ }
+
+ #[async_trait::async_trait]
+ #[typetag::serde(name = "test_metadata_counting_storage")]
+ impl crate::io::Storage for MetadataCountingStorage {
+ async fn exists(&self, path: &str) -> Result<bool> {
+ crate::io::LocalFsStorage::new().exists(path).await
+ }
+
+ async fn metadata(&self, path: &str) ->
Result<crate::io::FileMetadata> {
+ self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+ if self.fail_metadata {
+ return Err(
+ Error::new(ErrorKind::Unexpected, "metadata temporarily
unavailable")
+ .with_retryable(true),
+ );
+ }
+ crate::io::LocalFsStorage::new().metadata(path).await
+ }
+
+ async fn read(&self, path: &str) -> Result<Bytes> {
+ crate::io::LocalFsStorage::new().read(path).await
+ }
+
+ async fn reader(&self, path: &str) -> Result<Box<dyn
crate::io::FileRead>> {
+ crate::io::LocalFsStorage::new().reader(path).await
+ }
+
+ async fn write(&self, path: &str, bs: Bytes) -> Result<()> {
+ crate::io::LocalFsStorage::new().write(path, bs).await
+ }
+
+ async fn writer(&self, path: &str) -> Result<Box<dyn
crate::io::FileWrite>> {
+ crate::io::LocalFsStorage::new().writer(path).await
+ }
+
+ async fn delete(&self, path: &str) -> Result<()> {
+ crate::io::LocalFsStorage::new().delete(path).await
+ }
+
+ async fn delete_prefix(&self, path: &str) -> Result<()> {
+ crate::io::LocalFsStorage::new().delete_prefix(path).await
+ }
+
+ async fn delete_stream(
+ &self,
+ paths: futures::stream::BoxStream<'static, String>,
+ ) -> Result<()> {
+ crate::io::LocalFsStorage::new().delete_stream(paths).await
+ }
+
+ fn new_input(&self, path: &str) -> Result<crate::io::InputFile> {
+ Ok(crate::io::InputFile::new(
+ Arc::new(self.clone()),
+ path.to_string(),
+ ))
+ }
+
+ fn new_output(&self, path: &str) -> Result<crate::io::OutputFile> {
+ Ok(crate::io::OutputFile::new(
+ Arc::new(self.clone()),
+ path.to_string(),
+ ))
+ }
+ }
+
+ #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+ struct MetadataCountingStorageFactory {
+ #[serde(skip)]
+ calls: Arc<std::sync::atomic::AtomicUsize>,
+ #[serde(skip)]
+ fail_metadata: bool,
+ }
+
+ #[typetag::serde(name = "test_metadata_counting_storage_factory")]
+ impl crate::io::StorageFactory for MetadataCountingStorageFactory {
+ fn build(&self, _config: &crate::io::StorageConfig) -> Result<Arc<dyn
crate::io::Storage>> {
+ Ok(Arc::new(MetadataCountingStorage {
+ calls: Arc::clone(&self.calls),
+ fail_metadata: self.fail_metadata,
+ }))
+ }
+ }
Review Comment:
Could the test storage import `LocalFsStorage`, `AtomicUsize`, `Ordering`
and the other `crate::io` types at the top of the test module, and hold one
`LocalFsStorage` in a field? Each method currently builds a new
`crate::io::LocalFsStorage::new()` behind a fully qualified path, which makes
the 90 lines harder to scan for the two methods that differ (`metadata` and the
constructors).
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -612,7 +618,8 @@ mod _serde {
#[derive(Deserialize)]
pub(super) struct FileScanTaskDeleteFileSerde {
file_path: String,
- file_size_in_bytes: u64,
+ #[serde(default)]
+ file_size_in_bytes: Option<u64>,
Review Comment:
Could we add a serde test for the unknown size? A serialized task without
`file_size_in_bytes` used to fail to deserialize and now produces `None`, and
`skip_serializing_if` drops the field on the way out. A round trip of a task
with `None`, next to
`test_deserialized_deletion_vector_rejects_missing_fields`, would cover both
directions.
##########
crates/iceberg/src/arrow/delete_file_loader.rs:
##########
@@ -66,13 +66,45 @@ impl BasicDeleteFileLoader {
pub(crate) async fn parquet_to_batch_stream(
&self,
data_file_path: &str,
- file_size_in_bytes: u64,
+ file_size_in_bytes: Option<u64>,
key_metadata: Option<&[u8]>,
) -> Result<ArrowRecordBatchStream> {
/*
Essentially a super-cut-down ArrowReader. We can't use ArrowReader
directly
as that introduces a circular dependency.
*/
+ // `None` means the task did not carry a size (see
`FileScanTaskDeleteFile`).
+ // Resolve it when the file is first loaded.
+ let file_size_in_bytes = match file_size_in_bytes {
+ Some(file_size_in_bytes) => file_size_in_bytes,
+ None => {
+ self.file_io
+ .new_input(data_file_path)?
+ .metadata()
+ .await
+ .map_err(|error| {
+ // Keep the storage error's kind and retryability.
+ Error::new(
+ error.kind(),
+ format!("Failed to stat delete file
'{data_file_path}'"),
+ )
+ .with_retryable(error.retryable())
+ .with_source(error)
+ })?
+ .size
+ }
+ };
+ // Reading a Parquet footer needs at least its last eight bytes. Reject
+ // a smaller size rather than read the file as if it had no deletes.
+ if file_size_in_bytes < 8 {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Delete file '{data_file_path}' has a size of
{file_size_in_bytes} bytes, \
+ below the Parquet footer minimum"
+ ),
+ ));
+ }
Review Comment:
> A size below the Parquet footer minimum, including `Some(0)`, is rejected
instead of dropping the file's deletes.
Which path reads a sub-footer size as a file with no deletes? Parquet
already rejects a size below its footer in
[`ParquetMetaDataReader`](https://github.com/apache/arrow-rs/blob/f90e061326bd821a7af09281d9e92de6f3b603d9/parquet/src/file/metadata/reader.rs#L588-L590).
With this check disabled on the head commit, `None` on an empty file,
`Some(0)` and `Some(4)` each fail with `Failed to load Parquet metadata,
source: EOF: file size of N is less than footer`. A 10-byte file passes this
check and then fails in Parquet with `Corrupt footer`, so the check doesn't
define the minimum either.
What do you think about removing it and relying on Parquet's error? The
`footer minimum` assertions in
`unknown_delete_size_still_reports_missing_and_empty_objects` would then check
for Parquet's error instead. If the `DataInvalid` kind matters to callers,
could we compare against `parquet::file::FOOTER_SIZE` rather than the literal
`8`?
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -366,8 +366,14 @@ pub struct FileScanTaskDeleteFile {
/// The delete file path
file_path: String,
- /// The total size of the delete file in bytes, from the manifest entry.
- file_size_in_bytes: u64,
+ /// The total size of the delete file in bytes. Tasks planned from
manifest entries carry
+ /// the recorded size. An externally constructed task may leave it unset
when the size is
+ /// unavailable; the Parquet delete-file loader then resolves it with one
metadata request
+ /// when the file is first loaded. Deletion vectors do not use it.
Review Comment:
Should this doc also cover a recorded size that can't be trusted? A task
planned on the JVM always has `fileSizeInBytes()`, since
[`file_size_in_bytes`](https://github.com/apache/iceberg/blob/db43b88a6c0d470bd4f8acf08a18328750aeed20/format/spec.md?plain=1#L725)
is required in every format version. The case I know of is
apache/iceberg#12554: `rewrite_table_path` before Iceberg Java 1.12.0 left the
old size in the manifest after rewriting position delete files, and an engine
that wants to read those tables has to ignore the recorded size. Saying so here
would tell engine authors when to leave it unset.
##########
crates/iceberg/src/arrow/delete_file_loader.rs:
##########
@@ -66,13 +66,45 @@ impl BasicDeleteFileLoader {
pub(crate) async fn parquet_to_batch_stream(
&self,
data_file_path: &str,
- file_size_in_bytes: u64,
+ file_size_in_bytes: Option<u64>,
key_metadata: Option<&[u8]>,
) -> Result<ArrowRecordBatchStream> {
/*
Essentially a super-cut-down ArrowReader. We can't use ArrowReader
directly
as that introduces a circular dependency.
*/
+ // `None` means the task did not carry a size (see
`FileScanTaskDeleteFile`).
+ // Resolve it when the file is first loaded.
+ let file_size_in_bytes = match file_size_in_bytes {
+ Some(file_size_in_bytes) => file_size_in_bytes,
+ None => {
+ self.file_io
+ .new_input(data_file_path)?
+ .metadata()
+ .await
+ .map_err(|error| {
+ // Keep the storage error's kind and retryability.
+ Error::new(
+ error.kind(),
+ format!("Failed to stat delete file
'{data_file_path}'"),
+ )
+ .with_retryable(error.retryable())
Review Comment:
What should a caller do with a retryable error here? On the head commit, I
loaded a position delete file of unknown size whose path didn't exist yet, so
`load_deletes` failed with `Failed to stat delete file`. After creating the
file, a second `load_deletes` for the same delete file on the same
`CachingDeleteFileLoader` never completed (I stopped it after 3 seconds). A new
loader read it fine.
As I read it, this predates the PR, and #3345's description listed it under
"Not in this PR".
[`try_start_pos_del_load`](https://github.com/apache/iceberg-rust/blob/b154603a710d730d2f43f8af4db257cc137237ef/crates/iceberg/src/arrow/delete_filter.rs#L127-L145)
marks the file `Loading`, and only the success path calls
`finish_pos_del_load`, so a failed load leaves it `Loading` and later callers
wait on a notification that never comes. I see the same hang with a known size
when the open fails. Since
[`ArrowReader`](https://github.com/apache/iceberg-rust/blob/b154603a710d730d2f43f8af4db257cc137237ef/crates/iceberg/src/arrow/reader/mod.rs#L168-L172)
clones share this state, retrying `read` on a clone of the same reader would
hang too. #3274 tracks the equality-delete half, and #3275 fixes only that
half. This PR adds a new retryable failure on the positional path, so could we
file an issue for the positional half and link it here?
--
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]
