mbutrovich commented on code in PR #3433:
URL: https://github.com/apache/iceberg-rust/pull/3433#discussion_r4260422271
##########
crates/iceberg/src/arrow/delete_file_loader.rs:
##########
@@ -328,4 +352,68 @@ mod tests {
assert_eq!(batches.len(), 1);
assert_eq!(batches[0].num_rows(), 3);
}
+
+ #[tokio::test]
+ async fn unknown_delete_size_still_reports_missing_and_empty_objects() {
+ let temp = TempDir::new().unwrap();
+ let loader = BasicDeleteFileLoader::new(FileIO::new_with_fs(),
ScanMetrics::new());
+ let missing = temp.path().join("missing.parquet");
+ let error = loader
+ .parquet_to_batch_stream(missing.to_str().unwrap(), None, None)
+ .await
+ .err()
+ .unwrap();
+ assert!(error.to_string().contains("Failed to stat delete file"));
+ // The storage error's kind and retryability are kept exactly.
+ let storage_error = FileIO::new_with_fs()
+ .new_input(missing.to_str().unwrap())
+ .unwrap()
+ .metadata()
+ .await
+ .err()
+ .unwrap();
+ assert_eq!(error.kind(), storage_error.kind());
+ assert_eq!(error.retryable(), storage_error.retryable());
+
+ // A size too small for a Parquet footer fails the read, whether it was
+ // resolved from storage or supplied by the task.
+ let empty = temp.path().join("empty.parquet");
+ std::fs::write(&empty, []).unwrap();
+ for size in [None, Some(0), Some(4)] {
+ let error = loader
+ .parquet_to_batch_stream(empty.to_str().unwrap(), size, None)
+ .await
+ .err()
+ .unwrap();
+ assert!(error.to_string().contains("is less than footer"),
"{error}");
+ }
+
+ // A real file with an unknown size is sized lazily and read in full.
+ let present = temp.path().join("present.parquet");
+ let schema =
Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+ "pos",
+ arrow_schema::DataType::Int64,
+ false,
+ )]));
+ let batch = arrow_array::RecordBatch::try_new(Arc::clone(&schema),
vec![Arc::new(
+ arrow_array::Int64Array::from(vec![1, 2, 3]),
+ )])
+ .unwrap();
+ let mut writer = parquet::arrow::ArrowWriter::try_new(
+ std::fs::File::create(&present).unwrap(),
+ schema,
+ None,
+ )
+ .unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ let batches: Vec<_> = loader
+ .parquet_to_batch_stream(present.to_str().unwrap(), None, None)
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 3);
+ }
Review Comment:
Could this test keep only the footer cases? As I read it, the other two
parts are covered elsewhere in this PR. The missing-file case is in
[`test_load_deletes_reports_unknown_size_failures`](https://github.com/apache/iceberg-rust/blob/b2981e2ead534ade399992edfd88d4cb27df7fd5/crates/iceberg/src/arrow/caching_delete_file_loader.rs#L1392-L1434),
which also checks that a retryable storage error stays retryable. A local
missing file can't show that, since both sides of the `kind()` and
`retryable()` comparison come from the same `LocalFsStorage` error. The read of
a present file with an unknown size is the `None` iteration in
[`test_read_encrypted_positional_delete_file`](https://github.com/apache/iceberg-rust/blob/b2981e2ead534ade399992edfd88d4cb27df7fd5/crates/iceberg/src/arrow/delete_file_loader.rs#L256-L277),
which goes through the same `parquet_to_batch_stream` call. Dropping the copy
also removes the inline `arrow_schema::`, `arrow_array::` and
`parquet::arrow::` paths. The sugg
estion also adds the `test_` prefix the other tests in this module use.
```suggestion
#[tokio::test]
async fn test_delete_file_too_small_for_footer_fails() {
let temp = TempDir::new().unwrap();
let loader = BasicDeleteFileLoader::new(FileIO::new_with_fs(),
ScanMetrics::new());
// A size too small for a Parquet footer fails the read, whether it
was
// resolved from storage or supplied by the task.
let empty = temp.path().join("empty.parquet");
std::fs::write(&empty, []).unwrap();
for size in [None, Some(0), Some(4)] {
let error = loader
.parquet_to_batch_stream(empty.to_str().unwrap(), size, None)
.await
.err()
.unwrap();
assert!(error.to_string().contains("is less than footer"),
"{error}");
}
}
```
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -1129,6 +1154,320 @@ 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)]
+ inner: LocalFsStorage,
+ #[serde(skip)]
+ calls: Arc<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 Storage for MetadataCountingStorage {
+ async fn metadata(&self, path: &str) -> Result<FileMetadata> {
+ self.calls.fetch_add(1, Ordering::SeqCst);
+ if self.fail_metadata {
+ return Err(
+ Error::new(ErrorKind::Unexpected, "metadata temporarily
unavailable")
+ .with_retryable(true),
+ );
+ }
+ self.inner.metadata(path).await
+ }
+
+ fn new_input(&self, path: &str) -> Result<InputFile> {
+ Ok(InputFile::new(Arc::new(self.clone()), path.to_string()))
+ }
+
+ fn new_output(&self, path: &str) -> Result<OutputFile> {
+ Ok(OutputFile::new(Arc::new(self.clone()), path.to_string()))
+ }
+
+ async fn exists(&self, path: &str) -> Result<bool> {
+ self.inner.exists(path).await
+ }
+
+ async fn read(&self, path: &str) -> Result<Bytes> {
+ self.inner.read(path).await
+ }
+
+ async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> {
+ self.inner.reader(path).await
+ }
+
+ async fn write(&self, path: &str, bs: Bytes) -> Result<()> {
+ self.inner.write(path, bs).await
+ }
+
+ async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> {
+ self.inner.writer(path).await
+ }
+
+ async fn delete(&self, path: &str) -> Result<()> {
+ self.inner.delete(path).await
+ }
+
+ async fn delete_prefix(&self, path: &str) -> Result<()> {
+ self.inner.delete_prefix(path).await
+ }
+
+ async fn delete_stream(&self, paths: BoxStream<'static, String>) ->
Result<()> {
+ self.inner.delete_stream(paths).await
+ }
+ }
+
+ #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+ struct MetadataCountingStorageFactory {
+ #[serde(skip)]
+ calls: Arc<AtomicUsize>,
+ #[serde(skip)]
+ fail_metadata: bool,
+ }
+
+ #[typetag::serde(name = "test_metadata_counting_storage_factory")]
+ impl StorageFactory for MetadataCountingStorageFactory {
+ fn build(&self, _config: &StorageConfig) -> Result<Arc<dyn Storage>> {
+ Ok(Arc::new(MetadataCountingStorage {
+ inner: LocalFsStorage,
+ calls: Arc::clone(&self.calls),
+ fail_metadata: self.fail_metadata,
+ }))
+ }
+ }
+
+ /// Rebuilds `delete` at `path` without a file size.
+ fn without_size(delete: &FileScanTaskDeleteFile, path: &str) ->
FileScanTaskDeleteFile {
+ FileScanTaskDeleteFile::builder()
+ .with_file_path(path.to_string())
+ .with_file_type(delete.file_type())
+ .with_file_format(delete.file_format())
+ .with_partition_spec_id(delete.partition_spec_id())
+ .with_equality_ids(delete.equality_ids().map(<[i32]>::to_vec))
+
.with_referenced_data_file(delete.referenced_data_file().map(str::to_string))
+ .with_content_offset(delete.content_offset())
+ .with_content_size_in_bytes(delete.content_size_in_bytes())
+ .with_record_count(delete.record_count())
+ .with_key_metadata(delete.key_metadata().map(Box::from))
+ .build()
+ .unwrap()
+ }
+
+ #[tokio::test]
+ async fn test_unknown_delete_sizes_are_resolved_once_per_file() {
+ let tmp_dir = TempDir::new().unwrap();
+ let file_scan_tasks = setup(tmp_dir.path());
+ let factory = Arc::new(MetadataCountingStorageFactory::default());
+ let calls = Arc::clone(&factory.calls);
+ let file_io = FileIOBuilder::new(factory).build();
+ let task = &file_scan_tasks[0];
+ let unknown: Vec<_> = task
+ .deletes()
+ .iter()
+ .map(|delete| without_size(delete, delete.file_path()))
+ .collect();
+ let positions = |filter: &DeleteFilter| -> Vec<u64> {
+ filter
+ .get_delete_vector(task)
+ .unwrap()
+ .lock()
+ .unwrap()
+ .iter()
+ .collect()
+ };
+
+ // Known sizes never request metadata.
+ let known = CachingDeleteFileLoader::new(file_io.clone(), 10,
Runtime::current())
+ .load_deletes(task.deletes(), task.schema_ref())
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(calls.load(Ordering::SeqCst), 0);
+
+ // Unknown sizes: one request per delete file, and none again when
+ // another task references an already loaded delete file.
+ let loader = CachingDeleteFileLoader::new(file_io, 10,
Runtime::current());
+ let lazy = loader
+ .load_deletes(&unknown, task.schema_ref())
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(calls.load(Ordering::SeqCst), unknown.len());
+ loader
+ .load_deletes(&unknown[1..], task.schema_ref())
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(calls.load(Ordering::SeqCst), unknown.len());
+
+ // The same positions are deleted either way.
+ let expected = positions(&known);
+ assert_eq!(expected.len(), 12);
+ assert_eq!(positions(&lazy), expected);
+ }
+
+ #[tokio::test]
+ async fn test_unknown_size_equality_deletes_are_resolved_once_and_cached()
{
+ use crate::scan::FileScanTask;
+ use crate::spec::DataFileFormat;
+
+ let tmp_dir = TempDir::new().unwrap();
+ let table_location = tmp_dir.path().to_str().unwrap();
+ let eq_delete_path =
setup_write_equality_delete_file_1(table_location);
+ let schema = Arc::new(
+ Schema::builder()
+ .with_fields(vec![
+ NestedField::optional(2, "y",
Type::Primitive(PrimitiveType::Long)).into(),
+ NestedField::optional(3, "z",
Type::Primitive(PrimitiveType::Long)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+ let equality_delete = |file_size_in_bytes| {
+ FileScanTaskDeleteFile::builder()
+ .with_file_path(eq_delete_path.clone())
+ .with_file_size_in_bytes_opt(file_size_in_bytes)
+ .with_file_type(DataContentType::EqualityDeletes)
+ .with_file_format(DataFileFormat::Parquet)
+ .with_partition_spec_id(0)
+ .with_equality_ids(Some(vec![2, 3]))
+ .build()
+ .unwrap()
+ };
+ let data_task = |name: &str, delete| {
+ FileScanTask::builder()
+ .with_file_size_in_bytes(0)
+ .with_start(0)
+ .with_length(0)
+ .with_data_file_path(format!("{table_location}/{name}"))
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(Arc::clone(&schema))
+ .with_project_field_ids(vec![2, 3])
+ .with_deletes(vec![delete])
+ .with_case_sensitive(false)
+ .build()
+ .unwrap()
+ };
+ let known_size = std::fs::metadata(&eq_delete_path).unwrap().len();
+
+ let factory = Arc::new(MetadataCountingStorageFactory::default());
+ let calls = Arc::clone(&factory.calls);
+ let file_io = FileIOBuilder::new(factory).build();
+
+ // Known size: no metadata request.
+ let known_task = data_task("known.parquet",
equality_delete(Some(known_size)));
+ let known = CachingDeleteFileLoader::new(file_io.clone(), 10,
Runtime::current())
+ .load_deletes(known_task.deletes(), known_task.schema_ref())
+ .await
+ .unwrap()
+ .unwrap()
+ .build_equality_delete_predicate(&known_task)
+ .await
+ .unwrap();
+ assert!(known.is_some());
+ assert_eq!(calls.load(Ordering::SeqCst), 0);
+
+ // Unknown size: one metadata request, and none when a second data
+ // file referencing the same equality delete file is loaded.
+ let loader = CachingDeleteFileLoader::new(file_io, 10,
Runtime::current());
+ for (name, expected_calls) in [("first.parquet", 1),
("second.parquet", 1)] {
+ let task = data_task(name, equality_delete(None));
+ let predicate = loader
+ .load_deletes(task.deletes(), task.schema_ref())
+ .await
+ .unwrap()
+ .unwrap()
+ .build_equality_delete_predicate(&task)
+ .await
+ .unwrap();
+ assert_eq!(predicate, known, "{name}");
+ assert_eq!(calls.load(Ordering::SeqCst), expected_calls, "{name}");
+ }
+ }
+
+ #[tokio::test]
+ async fn test_load_deletes_reports_unknown_size_failures() {
+ let tmp_dir = TempDir::new().unwrap();
+ let file_scan_tasks = setup(tmp_dir.path());
+ let task = &file_scan_tasks[0];
+ let unknown_size = |path: &str| without_size(&task.deletes()[0], path);
+
+ // A missing delete file fails the load instead of dropping its
deletes.
+ let missing = unknown_size(&format!(
+ "{}/missing-pos-del.parquet",
+ tmp_dir.path().to_str().unwrap()
+ ));
+ let error = CachingDeleteFileLoader::new(FileIO::new_with_fs(), 10,
Runtime::current())
+ .load_deletes(&[missing], task.schema_ref())
+ .await
+ .unwrap()
+ .err()
+ .unwrap();
+ assert!(
+ error.to_string().contains("Failed to stat delete file"),
+ "{error}"
+ );
+
+ // A retryable storage failure stays retryable through the loader.
+ let factory = Arc::new(MetadataCountingStorageFactory {
+ fail_metadata: true,
+ ..Default::default()
+ });
+ let file_io = FileIOBuilder::new(factory).build();
+ let present = unknown_size(task.deletes()[0].file_path());
+ let error = CachingDeleteFileLoader::new(file_io, 10,
Runtime::current())
+ .load_deletes(&[present], task.schema_ref())
+ .await
+ .unwrap()
+ .err()
+ .unwrap();
+ assert!(
+ error.to_string().contains("Failed to stat delete file"),
+ "{error}"
+ );
+ assert_eq!(error.kind(), ErrorKind::Unexpected);
+ assert!(error.retryable());
+ }
+
+ #[tokio::test]
+ async fn test_delete_file_loader_parses_equality_deletes_of_unknown_size()
{
+ let tmp_dir = TempDir::new().unwrap();
+ let table_location = tmp_dir.path().as_os_str().to_str().unwrap();
+ let eq_delete_file_path =
setup_write_equality_delete_file_1(table_location);
+
+ let basic_delete_file_loader =
+ BasicDeleteFileLoader::new(FileIO::new_with_fs(),
ScanMetrics::new());
+ let known = basic_delete_file_loader
+ .parquet_to_batch_stream(
+ &eq_delete_file_path,
+ Some(std::fs::metadata(&eq_delete_file_path).unwrap().len()),
+ None,
+ )
+ .await
+ .unwrap();
+ let unknown = basic_delete_file_loader
+ .parquet_to_batch_stream(&eq_delete_file_path, None, None)
+ .await
+ .unwrap();
+
+ let eq_ids = HashSet::from_iter(vec![2, 3, 4, 6, 8]);
+ let known =
CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream(
+ known,
+ eq_ids.clone(),
+ )
+ .await
+ .unwrap();
+ let unknown =
+
CachingDeleteFileLoader::parse_equality_deletes_record_batch_stream(unknown,
eq_ids)
+ .await
+ .unwrap();
+ assert_eq!(unknown, known);
+ }
Review Comment:
Is this test needed alongside
`test_unknown_size_equality_deletes_are_resolved_once_and_cached`? That test
already [compares the
predicate](https://github.com/apache/iceberg-rust/blob/b2981e2ead534ade399992edfd88d4cb27df7fd5/crates/iceberg/src/arrow/caching_delete_file_loader.rs#L1387)
from an equality delete file of unknown size with the one from the known size,
through `load_deletes`. This one checks the same thing one layer down, with the
same file. The wider `eq_ids` set here doesn't matter for this change, because
the size only affects how the file is opened.
--
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]
