Bump datafusion from 50.2.0 to 51.0.0 in /rust

java/compaction/compaction-datafusion/src/main/java/sleeper/compaction/datafusion/DataFusionCompactionRunner.java

@@ -17,6 +17,7 @@
 
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.parquet.hadoop.ParquetWriter;
 import org.slf4j.Logger;
@@ -66,14 +67,17 @@ public DataFusionCompactionRunner(DataFusionAwsConfig awsConfig, Configuration h
     @Override
     public RowsProcessed compact(CompactionJob job, TableProperties tableProperties, Region region) throws IOException {
         jnr.ffi.Runtime runtime = jnr.ffi.Runtime.getRuntime(DataFusionCompactionFunctions.INSTANCE);
-
         FFICommonConfig params = createCompactionParams(job, tableProperties, region, awsConfig, runtime);
 
         RowsProcessed result = invokeDataFusion(job, params, runtime);
 
-        if (result.getRowsWritten() < 1) {
+        // Get the filesystem object
+        FileSystem fs = FileSystem.get(hadoopConf);
+        Path outputPath = new Path(job.getOutputFile());
+
+        if (result.getRowsWritten() < 1 && !fs.exists(outputPath)) {
             try (ParquetWriter<Row> writer = ParquetRowWriterFactory.createParquetRowWriter(
-                    new Path(job.getOutputFile()), tableProperties, hadoopConf)) {
+                    outputPath, tableProperties, hadoopConf)) {
                 // Write an empty file. This should be temporary, as we expect DataFusion to add support for this.
                 // See the test should_merge_empty_files in compaction_test.rs
             }

java/compaction/compaction-datafusion/src/test/java/sleeper/compaction/datafusion/DataFusionCompactionRunnerIT.java

@@ -261,6 +261,11 @@ void shouldMergeTwoEmptyFiles() throws Exception {
             // When
             runTask(job);
 
+            // The should_merge_empty_files test in rust/sleeper_core/tests/compaction_test.rs asserts that no result file
+            // is written when empty files are compacted. This differs here as it appears DataFusion will not always
+            // write an empty results file. See comment in that file. We ensure a result file is always written on the Java
+            // side in DataFusionCompactionRunner.
+
             // Then
             assertThat(getRowsProcessed(job)).isEqualTo(new RowsProcessed(0, 0));
             assertThat(readDataFile(job.getOutputFile())).isEmpty();