JeonDaehong commented on code in PR #3433:
URL: https://github.com/apache/iceberg-rust/pull/3433#discussion_r4259896580
##########
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:
Done in b2981e2ea: the `crate::io` types are imported at the top and the
storage holds a `LocalFsStorage`.
##########
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:
Reproduced on `main` at b154603a7 with a known size as well: the first load
fails, a second load on the same loader after the file appears doesn't complete
within 3s, and a new loader reads it fine. Filed #NNNN for the positional half.
I'm happy to take the fix there.
--
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]
