Fix Gzip file source (#4457)

Co-authored-by: François Massot <[email protected]>

quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs

@@ -636,7 +636,8 @@ mod tests {
 
     async fn test_indexing_pipeline_num_fails_before_success(
         mut num_fails: usize,
-    ) -> anyhow::Result<bool> {
+        test_file: &str,
+    ) -> anyhow::Result<()> {
         let universe = Universe::new();
         let mut metastore = MetastoreServiceClient::mock();
         metastore
@@ -696,7 +697,7 @@ mod tests {
             max_num_pipelines_per_indexer: NonZeroUsize::new(1).unwrap(),
             desired_num_pipelines: NonZeroUsize::new(1).unwrap(),
             enabled: true,
-            source_params: SourceParams::file(PathBuf::from("data/test_corpus.json")),
+            source_params: SourceParams::file(PathBuf::from(test_file)),
             transform_config: None,
             input_format: SourceInputFormat::Json,
         };
@@ -728,23 +729,31 @@ mod tests {
         let (pipeline_exit_status, pipeline_statistics) = pipeline_handle.join().await;
         assert_eq!(pipeline_statistics.generation, 1);
         assert_eq!(pipeline_statistics.num_spawn_attempts, 1 + num_fails);
-        Ok(pipeline_exit_status.is_success())
+        assert!(pipeline_exit_status.is_success());
+        Ok(())
     }
 
     #[tokio::test]
     async fn test_indexing_pipeline_retry_0() -> anyhow::Result<()> {
-        test_indexing_pipeline_num_fails_before_success(0).await?;
-        Ok(())
+        test_indexing_pipeline_num_fails_before_success(0, "data/test_corpus.json").await
     }
 
     #[tokio::test]
     async fn test_indexing_pipeline_retry_1() -> anyhow::Result<()> {
-        test_indexing_pipeline_num_fails_before_success(1).await?;
-        Ok(())
+        test_indexing_pipeline_num_fails_before_success(1, "data/test_corpus.json").await
     }
 
     #[tokio::test]
-    async fn test_indexing_pipeline_simple() -> anyhow::Result<()> {
+    async fn test_indexing_pipeline_retry_0_gz() -> anyhow::Result<()> {
+        test_indexing_pipeline_num_fails_before_success(0, "data/test_corpus.json.gz").await
+    }
+
+    #[tokio::test]
+    async fn test_indexing_pipeline_retry_1_gz() -> anyhow::Result<()> {
+        test_indexing_pipeline_num_fails_before_success(1, "data/test_corpus.json.gz").await
+    }
+
+    async fn indexing_pipeline_simple(test_file: &str) -> anyhow::Result<()> {
         let mut metastore = MetastoreServiceClient::mock();
         metastore
             .expect_index_metadata()
@@ -796,7 +805,7 @@ mod tests {
             max_num_pipelines_per_indexer: NonZeroUsize::new(1).unwrap(),
             desired_num_pipelines: NonZeroUsize::new(1).unwrap(),
             enabled: true,
-            source_params: SourceParams::file(PathBuf::from("data/test_corpus.json")),
+            source_params: SourceParams::file(PathBuf::from(test_file)),
             transform_config: None,
             input_format: SourceInputFormat::Json,
         };
@@ -833,6 +842,16 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_indexing_pipeline_simple() -> anyhow::Result<()> {
+        indexing_pipeline_simple("data/test_corpus.json").await
+    }
+
+    #[tokio::test]
+    async fn test_indexing_pipeline_simple_gz() -> anyhow::Result<()> {
+        indexing_pipeline_simple("data/test_corpus.json.gz").await
+    }
+
     #[tokio::test]
     async fn test_merge_pipeline_does_not_stop_on_indexing_pipeline_failure() {
         let mut mock_metastore = MetastoreServiceClient::mock();
@@ -930,8 +949,7 @@ mod tests {
         panic!("Pipeline was apparently not restarted.");
     }
 
-    #[tokio::test]
-    async fn test_indexing_pipeline_all_failures_handling() -> anyhow::Result<()> {
+    async fn indexing_pipeline_all_failures_handling(test_file: &str) -> anyhow::Result<()> {
         let mut metastore = MetastoreServiceClient::mock();
         metastore
             .expect_index_metadata()
@@ -981,7 +999,7 @@ mod tests {
             max_num_pipelines_per_indexer: NonZeroUsize::new(1).unwrap(),
             desired_num_pipelines: NonZeroUsize::new(1).unwrap(),
             enabled: true,
-            source_params: SourceParams::file(PathBuf::from("data/test_corpus.json")),
+            source_params: SourceParams::file(PathBuf::from(test_file)),
             transform_config: None,
             input_format: SourceInputFormat::Json,
         };
@@ -1040,4 +1058,14 @@ mod tests {
         universe.assert_quit().await;
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_indexing_pipeline_all_failures_handling() -> anyhow::Result<()> {
+        indexing_pipeline_all_failures_handling("data/test_corpus.json").await
+    }
+
+    #[tokio::test]
+    async fn test_indexing_pipeline_all_failures_handling_gz() -> anyhow::Result<()> {
+        indexing_pipeline_all_failures_handling("data/test_corpus.json.gz").await
+    }
 }

quickwit/quickwit-indexing/src/source/file_source.rs

@@ -153,13 +153,6 @@ impl TypedSourceFactory for FileSourceFactory {
             let (dir_uri, file_name) = dir_and_filename(filepath)?;
             let storage = ctx.storage_resolver.resolve(&dir_uri).await?;
             let file_size = storage.file_num_bytes(file_name).await?.try_into().unwrap();
-            if offset > file_size {
-                return Err(anyhow::anyhow!(
-                    "offset {} can't be greater than the file size {}",
-                    offset,
-                    file_size
-                ));
-            }
             // If it's a gzip file, we can't seek to a specific offset, we need to start from the
             // beginning of the file, decompress and skip the first `offset` bytes.
             if filepath.extension() == Some(OsStr::new("gz")) {