Support reporting statistics and currently written byte size from writers in jni

java/vortex-jni/src/main/java/dev/vortex/api/VortexColumnStatistics.java

@@ -0,0 +1,82 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+package dev.vortex.api;
+
+import java.util.Optional;
+import java.util.OptionalLong;
+
+/** Statistics and physical size reported for one top-level column after a Vortex write. */
+public final class VortexColumnStatistics {
+    private final int columnIndex;
+    private final long compressedSize;
+    private final long valueCount;
+    private final long nullValueCount;
+    private final long nanValueCount;
+    private final Object lowerBound;
+    private final Object upperBound;
+
+    /**
+     * Construct native write statistics.
+     *
+     * <p>This constructor is public so the JNI implementation can instantiate the value without reflective access.
+     * Applications should obtain instances from {@link VortexWriteSummary#columnStatistics()}.
+     */
+    public VortexColumnStatistics(
+            int columnIndex,
+            long compressedSize,
+            long valueCount,
+            long nullValueCount,
+            long nanValueCount,
+            Object lowerBound,
+            Object upperBound) {
+        this.columnIndex = columnIndex;
+        this.compressedSize = compressedSize;
+        this.valueCount = valueCount;
+        this.nullValueCount = nullValueCount;
+        this.nanValueCount = nanValueCount;
+        this.lowerBound = lowerBound;
+        this.upperBound = upperBound;
+    }
+
+    /** Zero-based position of this column in the writer's Arrow schema. */
+    public int columnIndex() {
+        return columnIndex;
+    }
+
+    /** Compressed bytes referenced by this column's physical layout. */
+    public long compressedSize() {
+        return compressedSize;
+    }
+
+    /** Number of values in this top-level column. */
+    public long valueCount() {
+        return valueCount;
+    }
+
+    /** Exact null count, or empty when Vortex did not compute one for this data type. */
+    public OptionalLong nullValueCount() {
+        return nullValueCount >= 0 ? OptionalLong.of(nullValueCount) : OptionalLong.empty();
+    }
+
+    /** Exact NaN count, or empty for non-floating-point columns. */
+    public OptionalLong nanValueCount() {
+        return nanValueCount >= 0 ? OptionalLong.of(nanValueCount) : OptionalLong.empty();
+    }
+
+    /**
+     * Lower bound represented using the corresponding Arrow scalar's Java type.
+     *
+     * <p>Integers use {@link Integer}, {@link Long}, or {@link java.math.BigInteger}; floating-point values use
+     * {@link Float} or {@link Double}; decimal values use {@link java.math.BigDecimal}; strings use {@link String}; and
+     * binary values use {@code byte[]}.
+     */
+    public Optional<Object> lowerBound() {
+        return Optional.ofNullable(lowerBound);
+    }
+
+    /** Upper bound represented using the corresponding Arrow scalar's Java type. */
+    public Optional<Object> upperBound() {
+        return Optional.ofNullable(upperBound);
+    }
+}

java/vortex-jni/src/main/java/dev/vortex/api/VortexWriteSummary.java

@@ -0,0 +1,40 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+package dev.vortex.api;
+
+import java.util.List;
+
+/** Immutable metadata returned after a Vortex writer has finalized its file. */
+public final class VortexWriteSummary {
+    private final long fileSize;
+    private final long rowCount;
+    private final List<VortexColumnStatistics> columnStatistics;
+
+    /**
+     * Construct a native write summary.
+     *
+     * <p>This constructor is public so the JNI implementation can instantiate the value without reflective access.
+     * Applications should obtain summaries from {@link VortexWriter#finish()}.
+     */
+    public VortexWriteSummary(long fileSize, long rowCount, VortexColumnStatistics[] columnStatistics) {
+        this.fileSize = fileSize;
+        this.rowCount = rowCount;
+        this.columnStatistics = List.of(columnStatistics.clone());
+    }
+
+    /** Exact size of the completed file in bytes. */
+    public long fileSize() {
+        return fileSize;
+    }
+
+    /** Exact number of rows written to the file. */
+    public long rowCount() {
+        return rowCount;
+    }
+
+    /** Per-column statistics in Arrow schema order. */
+    public List<VortexColumnStatistics> columnStatistics() {
+        return columnStatistics;
+    }
+}

java/vortex-jni/src/main/java/dev/vortex/api/VortexWriter.java

@@ -29,6 +29,7 @@
 public final class VortexWriter implements AutoCloseable {
     private final long pointer;
     private final AtomicBoolean closed = new AtomicBoolean(false);
+    private volatile VortexWriteSummary summary;
 
     private VortexWriter(long pointer) {
         Preconditions.checkArgument(pointer != 0, "invalid writer pointer");
@@ -79,15 +80,45 @@ public void writeBatch(long arrowArrayAddr, long arrowSchemaAddr) throws IOExcep
         }
     }
 
-    /** Flush any pending batches and finalize the file. Idempotent. */
-    @Override
-    public void close() throws IOException {
+    /**
+     * Return the number of bytes successfully written to the underlying sink so far.
+     *
+     * <p>This count does not include queued batches or data still buffered by layout strategies, so it may lag the
+     * amount of input accepted by {@link #writeBatch(long, long)}. After {@link #finish()}, it is the exact completed
+     * file size and is equal to {@link VortexWriteSummary#fileSize()}.
+     */
+    public synchronized long bytesWritten() {
+        if (summary != null) {
+            return summary.fileSize();
+        }
+        Preconditions.checkState(!closed.get(), "writer closed without a write summary");
+        long bytesWritten = NativeWriter.bytesWritten(pointer);
+        Preconditions.checkState(bytesWritten >= 0, "native writer returned an invalid byte count");
+        return bytesWritten;
+    }
+
+    /**
+     * Flush pending batches, finalize the file, and return its statistics and physical sizes.
+     *
+     * <p>This method is idempotent. Later calls return the same immutable summary.
+     */
+    public synchronized VortexWriteSummary finish() throws IOException {
         if (closed.compareAndSet(false, true)) {
             try {
-                NativeWriter.close(pointer);
+                summary = NativeWriter.finish(pointer);
             } catch (RuntimeException e) {
                 throw new IOException("failed to close writer", e);
             }
         }
+        if (summary == null) {
+            throw new IOException("writer was closed without retaining its write summary");
+        }
+        return summary;
+    }
+
+    /** Flush any pending batches and finalize the file. Idempotent. */
+    @Override
+    public void close() throws IOException {
+        finish();
     }
 }

java/vortex-jni/src/main/java/dev/vortex/jni/NativeWriter.java

@@ -3,6 +3,7 @@
 
 package dev.vortex.jni;
 
+import dev.vortex.api.VortexWriteSummary;
 import java.util.Map;
 
 /** JNI boundary for {@link dev.vortex.api.VortexWriter}. */
@@ -27,6 +28,12 @@ public static native long create(
      */
     public static native boolean writeBatch(long writerPointer, long arrowArrayAddress, long arrowSchemaAddress);
 
+    /** Number of bytes successfully written to the underlying sink so far. */
+    public static native long bytesWritten(long writerPointer);
+
     /** Flush and close the writer. Must be called exactly once. */
     public static native void close(long writerPointer);
+
+    /** Flush and close the writer, returning its completed-file summary. Must be called exactly once. */
+    public static native VortexWriteSummary finish(long writerPointer);
 }

java/vortex-jni/src/test/java/dev/vortex/jni/JNIWriterTest.java

@@ -14,6 +14,7 @@
 import dev.vortex.api.Scan;
 import dev.vortex.api.ScanOptions;
 import dev.vortex.api.Session;
+import dev.vortex.api.VortexWriteSummary;
 import dev.vortex.api.VortexWriter;
 import dev.vortex.arrow.ArrowAllocation;
 import java.io.IOException;
@@ -157,6 +158,8 @@ public void testWriteBatch() throws IOException {
         Schema schema = personSchema();
 
         Session session = Session.create();
+        VortexWriteSummary summary;
+        long bytesWhileOpen;
         try (VortexWriter writer = VortexWriter.create(session, writePath, schema, new HashMap<>(), allocator);
                 VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
             VarCharVector nameVec = (VarCharVector) root.getVector("name");
@@ -179,9 +182,25 @@ public void testWriteBatch() throws IOException {
                 Data.exportVectorSchemaRoot(allocator, root, null, arrowArray, arrowSchemaFfi);
                 writer.writeBatch(arrowArray.memoryAddress(), arrowSchemaFfi.memoryAddress());
             }
+            bytesWhileOpen = writer.bytesWritten();
+            summary = writer.finish();
+            assertEquals(summary.fileSize(), writer.bytesWritten());
         }
 
         assertTrue(Files.exists(outputPath), "output file should exist");
+        assertTrue(bytesWhileOpen >= 0);
+        assertTrue(bytesWhileOpen <= summary.fileSize());
+        assertEquals(Files.size(outputPath), summary.fileSize());
+        assertEquals(3L, summary.rowCount());
+        assertEquals(2, summary.columnStatistics().size());
+        assertEquals(0, summary.columnStatistics().get(0).columnIndex());
+        assertTrue(summary.columnStatistics().get(0).compressedSize() > 0);
+        assertEquals(3L, summary.columnStatistics().get(0).valueCount());
+        assertEquals(0L, summary.columnStatistics().get(0).nullValueCount().orElseThrow());
+        assertEquals("Alice", summary.columnStatistics().get(0).lowerBound().orElseThrow());
+        assertEquals("Carol", summary.columnStatistics().get(0).upperBound().orElseThrow());
+        assertEquals(25, summary.columnStatistics().get(1).lowerBound().orElseThrow());
+        assertEquals(40, summary.columnStatistics().get(1).upperBound().orElseThrow());
 
         DataSource ds = DataSource.open(session, writePath);
         assertEquals(new DataSource.RowCount.Exact(3L), ds.rowCount());

vortex-array/src/stats/stats_set.rs

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use std::any::type_name;
 use std::fmt::Debug;
 
 use enum_iterator::all;
@@ -133,12 +134,7 @@ impl StatsSet {
                     .vortex_expect("failed to construct a scalar statistic"),
             )
             .unwrap_or_else(|err| {
-                vortex_panic!(
-                    err,
-                    "Failed to get stat {} as {}",
-                    stat,
-                    std::any::type_name::<T>()
-                )
+                vortex_panic!(err, "Failed to get stat {} as {}", stat, type_name::<T>())
             })
         })
     }

vortex-file/src/counting.rs

@@ -9,20 +9,22 @@ use std::sync::atomic::Ordering;
 use vortex_io::IoBuf;
 use vortex_io::VortexWrite;
 
-/// A wrapper around an `VortexWrite` that counts the number of bytes written.
-pub(crate) struct CountingVortexWrite<W> {
+/// A wrapper around a [`VortexWrite`] that counts the number of bytes written.
+pub struct CountingVortexWrite<W> {
     inner: W,
     bytes_written: Arc<AtomicU64>,
 }
 
 impl<W: VortexWrite> CountingVortexWrite<W> {
+    /// Wrap a writer with a new byte counter.
     pub fn new(inner: W) -> Self {
         Self {
             inner,
             bytes_written: Default::default(),
         }
     }
 
+    /// Returns the shared byte counter updated by this writer.
     pub fn counter(&self) -> Arc<AtomicU64> {
         Arc::clone(&self.bytes_written)
     }