feat(spark): let VortexOutputWriter store footer metadata

java/vortex-spark/common/src/main/java/dev/vortex/spark/write/VortexOutputWriter.java

@@ -14,6 +14,7 @@
 import dev.vortex.spark.VortexSparkSession;
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.util.Map;
 import org.apache.spark.TaskContext;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.execution.datasources.OutputWriter;
@@ -64,6 +65,21 @@ public final class VortexOutputWriter extends OutputWriter {
      * @param options additional write options
      */
     public VortexOutputWriter(String filePath, StructType schema, VortexOptions options, NativeWritable writable) {
+        this(filePath, schema, options, writable, Map.of());
+    }
+
+    /**
+     * Creates a writer that also stores {@code metadata} in the file footer, for callers that write files outside
+     * Spark's commit protocol.
+     *
+     * @param metadata footer segments, read back verbatim by {@code NativeFiles.readMetadata}
+     */
+    public VortexOutputWriter(
+            String filePath,
+            StructType schema,
+            VortexOptions options,
+            NativeWritable writable,
+            Map<String, byte[]> metadata) {
         this.filePath = filePath;
         this.options = options;
         this.writable = writable;
@@ -90,6 +106,7 @@ public VortexOutputWriter(String filePath, StructType schema, VortexOptions opti
 
             this.session = VortexSparkSession.get(options);
             this.vortexWriter = VortexWriter.builder(session, writable, arrowSchema, allocator)
+                    .metadata(metadata)
                     .build();
             this.vectorSchemaRoot = VectorSchemaRoot.create(arrowSchema, allocator);
             // The native writer holds several batches at once, so a batch of few rows must not take Arrow's default

java/vortex-spark/common/src/test/java/dev/vortex/spark/write/VortexOutputWriterTest.java

@@ -3,13 +3,15 @@
 
 package dev.vortex.spark.write;
 
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import dev.vortex.api.DataSource;
 import dev.vortex.api.Scan;
 import dev.vortex.api.ScanOptions;
+import dev.vortex.jni.NativeFiles;
 import dev.vortex.relocated.org.apache.arrow.memory.ArrowBuf;
 import dev.vortex.relocated.org.apache.arrow.memory.BufferAllocator;
 import dev.vortex.relocated.org.apache.arrow.memory.RootAllocator;
@@ -20,6 +22,7 @@
 import dev.vortex.spark.VortexSparkSession;
 import dev.vortex.spark.io.HadoopWritable;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -238,6 +241,24 @@ void aLateReleaseDoesNotFailCloseAndIsStillFreed() {
         assertEquals(1, readRows(writer.path()).size());
     }
 
+    @Test
+    void storesFooterMetadata(@TempDir Path tempDir) {
+        StructType schema = new StructType().add("id", DataTypes.LongType, false);
+        String path = tempDir.resolve("metadata.vortex").toUri().toString();
+        byte[] value = "v1".getBytes(StandardCharsets.UTF_8);
+        VortexOutputWriter writer = new VortexOutputWriter(
+                path,
+                schema,
+                VortexOptions.empty(),
+                HadoopWritable.create(new Configuration(), path),
+                Map.of("app.version", value));
+        writer.write(new GenericInternalRow(new Object[] {7L}));
+        writer.close();
+
+        Map<String, byte[]> metadata = NativeFiles.readMetadata(VortexSparkSession.get(), path, Map.of());
+        assertArrayEquals(value, metadata.get("app.version"));
+    }
+
     private VortexOutputWriter writer(String name, StructType schema, Map<String, String> values) {
         String path = tempDir.resolve(name).toUri().toString();
         return new VortexOutputWriter(path, schema, options(values), HadoopWritable.create(new Configuration(), path));