fix(mem_wal): support wal flush interval on durable writes

java/lance-jni/src/mem_wal.rs

@@ -1265,9 +1265,6 @@ fn build_writer_config(env: &mut JNIEnv, config: &JObject) -> Result<ShardWriter
     if let Some(v) = read_optional_bool(env, config, "durableWrite")? {
         writer_config = writer_config.with_durable_write(v);
     }
-    if let Some(v) = read_optional_bool(env, config, "syncIndexedWrite")? {
-        writer_config = writer_config.with_sync_indexed_write(v);
-    }
     if let Some(v) = read_optional_u64(env, config, "maxWalBufferSize")? {
         writer_config = writer_config.with_max_wal_buffer_size(v as usize);
     }
@@ -1289,12 +1286,6 @@ fn build_writer_config(env: &mut JNIEnv, config: &JObject) -> Result<ShardWriter
     if let Some(v) = read_optional_u64(env, config, "manifestScanBatchSize")? {
         writer_config = writer_config.with_manifest_scan_batch_size(v as usize);
     }
-    if let Some(v) = read_optional_u64(env, config, "asyncIndexBufferRows")? {
-        writer_config = writer_config.with_async_index_buffer_rows(v as usize);
-    }
-    if let Some(v) = read_optional_u64(env, config, "asyncIndexIntervalMs")? {
-        writer_config = writer_config.with_async_index_interval(Duration::from_millis(v));
-    }
     if let Some(v) = read_optional_u64(env, config, "backpressureLogIntervalMs")? {
         writer_config = writer_config.with_backpressure_log_interval(Duration::from_millis(v));
     }
@@ -1368,19 +1359,19 @@ fn write_stats_to_java<'a>(
 
 fn memtable_stats_to_java<'a>(env: &mut JNIEnv<'a>, stats: &MemTableStats) -> Result<JObject<'a>> {
     let max_buffered = box_u64_opt(env, stats.max_buffered_batch_position)?;
-    let max_flushed = box_u64_opt(env, stats.max_flushed_batch_position)?;
     let pending_start = box_u64_opt(env, stats.pending_wal_start_batch_position)?;
     let pending_end = box_u64_opt(env, stats.pending_wal_end_batch_position)?;
     Ok(env.new_object(
         "org/lance/memwal/MemTableStats",
-        "(JJJJLjava/lang/Long;Ljava/lang/Long;Ljava/lang/Long;Ljava/lang/Long;JJJ)V",
+        "(JJJJLjava/lang/Long;JJLjava/lang/Long;Ljava/lang/Long;JJJ)V",
         &[
             JValueGen::Long(stats.row_count as i64),
             JValueGen::Long(stats.batch_count as i64),
             JValueGen::Long(stats.estimated_size as i64),
             JValueGen::Long(stats.generation as i64),
             JValueGen::Object(&max_buffered),
-            JValueGen::Object(&max_flushed),
+            JValueGen::Long(stats.durable_batch_count as i64),
+            JValueGen::Long(stats.global_offset as i64),
             JValueGen::Object(&pending_start),
             JValueGen::Object(&pending_end),
             JValueGen::Long(stats.pending_wal_batch_count as i64),

java/src/main/java/org/lance/memwal/MemTableStats.java

@@ -24,7 +24,8 @@ public class MemTableStats {
   private final long estimatedSizeBytes;
   private final long generation;
   private final Optional<Long> maxBufferedBatchPosition;
-  private final Optional<Long> maxFlushedBatchPosition;
+  private final long durableBatchCount;
+  private final long globalOffset;
   private final Optional<Long> pendingWalStartBatchPosition;
   private final Optional<Long> pendingWalEndBatchPosition;
   private final long pendingWalBatchCount;
@@ -37,7 +38,8 @@ public MemTableStats(
       long estimatedSizeBytes,
       long generation,
       Long maxBufferedBatchPosition,
-      Long maxFlushedBatchPosition,
+      long durableBatchCount,
+      long globalOffset,
       Long pendingWalStartBatchPosition,
       Long pendingWalEndBatchPosition,
       long pendingWalBatchCount,
@@ -48,7 +50,8 @@ public MemTableStats(
     this.estimatedSizeBytes = estimatedSizeBytes;
     this.generation = generation;
     this.maxBufferedBatchPosition = Optional.ofNullable(maxBufferedBatchPosition);
-    this.maxFlushedBatchPosition = Optional.ofNullable(maxFlushedBatchPosition);
+    this.durableBatchCount = durableBatchCount;
+    this.globalOffset = globalOffset;
     this.pendingWalStartBatchPosition = Optional.ofNullable(pendingWalStartBatchPosition);
     this.pendingWalEndBatchPosition = Optional.ofNullable(pendingWalEndBatchPosition);
     this.pendingWalBatchCount = pendingWalBatchCount;
@@ -81,9 +84,14 @@ public Optional<Long> maxBufferedBatchPosition() {
     return maxBufferedBatchPosition;
   }
 
-  /** Highest WAL batch position flushed from the MemTable, if any. */
-  public Optional<Long> maxFlushedBatchPosition() {
-    return maxFlushedBatchPosition;
+  /** Writer-global count of WAL-durable batches (exclusive). */
+  public long durableBatchCount() {
+    return durableBatchCount;
+  }
+
+  /** Writer-global coordinate of this MemTable's batch 0. */
+  public long globalOffset() {
+    return globalOffset;
   }
 
   /** First WAL batch position pending flush, if any. */
@@ -119,7 +127,8 @@ public String toString() {
         .add("estimatedSizeBytes", estimatedSizeBytes)
         .add("generation", generation)
         .add("maxBufferedBatchPosition", maxBufferedBatchPosition.orElse(null))
-        .add("maxFlushedBatchPosition", maxFlushedBatchPosition.orElse(null))
+        .add("durableBatchCount", durableBatchCount)
+        .add("globalOffset", globalOffset)
         .add("pendingWalStartBatchPosition", pendingWalStartBatchPosition.orElse(null))
         .add("pendingWalEndBatchPosition", pendingWalEndBatchPosition.orElse(null))
         .add("pendingWalBatchCount", pendingWalBatchCount)

java/src/main/java/org/lance/memwal/ShardWriterConfig.java

@@ -28,16 +28,13 @@
  */
 public class ShardWriterConfig {
   private Optional<Boolean> durableWrite = Optional.empty();
-  private Optional<Boolean> syncIndexedWrite = Optional.empty();
   private Optional<Long> maxWalBufferSize = Optional.empty();
   private Optional<Long> maxWalFlushIntervalMs = Optional.empty();
   private Optional<Long> maxMemtableSize = Optional.empty();
   private Optional<Long> maxMemtableRows = Optional.empty();
   private Optional<Long> maxMemtableBatches = Optional.empty();
   private Optional<Long> maxUnflushedMemtableBytes = Optional.empty();
   private Optional<Long> manifestScanBatchSize = Optional.empty();
-  private Optional<Long> asyncIndexBufferRows = Optional.empty();
-  private Optional<Long> asyncIndexIntervalMs = Optional.empty();
   private Optional<Long> backpressureLogIntervalMs = Optional.empty();
   private Optional<Long> statsLogIntervalMs = Optional.empty();
   private List<MemWalHnswParams> hnswParams = Collections.emptyList();
@@ -48,12 +45,6 @@ public ShardWriterConfig withDurableWrite(boolean durableWrite) {
     return this;
   }
 
-  /** Whether indexed writes are applied synchronously. */
-  public ShardWriterConfig withSyncIndexedWrite(boolean syncIndexedWrite) {
-    this.syncIndexedWrite = Optional.of(syncIndexedWrite);
-    return this;
-  }
-
   /** Maximum size of the in-memory WAL buffer, in bytes. */
   public ShardWriterConfig withMaxWalBufferSize(long maxWalBufferSize) {
     Preconditions.checkArgument(
@@ -118,26 +109,6 @@ public ShardWriterConfig withManifestScanBatchSize(long manifestScanBatchSize) {
     return this;
   }
 
-  /** Number of rows buffered before an asynchronous index update is triggered. */
-  public ShardWriterConfig withAsyncIndexBufferRows(long asyncIndexBufferRows) {
-    Preconditions.checkArgument(
-        asyncIndexBufferRows >= 0,
-        "asyncIndexBufferRows must not be negative, got %s",
-        asyncIndexBufferRows);
-    this.asyncIndexBufferRows = Optional.of(asyncIndexBufferRows);
-    return this;
-  }
-
-  /** Interval between asynchronous index updates, in milliseconds. */
-  public ShardWriterConfig withAsyncIndexIntervalMs(long asyncIndexIntervalMs) {
-    Preconditions.checkArgument(
-        asyncIndexIntervalMs >= 0,
-        "asyncIndexIntervalMs must not be negative, got %s",
-        asyncIndexIntervalMs);
-    this.asyncIndexIntervalMs = Optional.of(asyncIndexIntervalMs);
-    return this;
-  }
-
   /** Interval between backpressure log messages, in milliseconds. */
   public ShardWriterConfig withBackpressureLogIntervalMs(long backpressureLogIntervalMs) {
     Preconditions.checkArgument(
@@ -172,10 +143,6 @@ public Optional<Boolean> durableWrite() {
     return durableWrite;
   }
 
-  public Optional<Boolean> syncIndexedWrite() {
-    return syncIndexedWrite;
-  }
-
   public Optional<Long> maxWalBufferSize() {
     return maxWalBufferSize;
   }
@@ -204,14 +171,6 @@ public Optional<Long> manifestScanBatchSize() {
     return manifestScanBatchSize;
   }
 
-  public Optional<Long> asyncIndexBufferRows() {
-    return asyncIndexBufferRows;
-  }
-
-  public Optional<Long> asyncIndexIntervalMs() {
-    return asyncIndexIntervalMs;
-  }
-
   public Optional<Long> backpressureLogIntervalMs() {
     return backpressureLogIntervalMs;
   }

java/src/test/java/org/lance/memwal/MemWalTest.java

@@ -409,7 +409,6 @@ void testShardWriterDeleteMasksBaseRow(@TempDir Path tempDir) throws Exception {
       ShardWriterConfig config =
           new ShardWriterConfig()
               .withDurableWrite(true)
-              .withSyncIndexedWrite(true)
               .withMaxWalBufferSize(1)
               .withMaxWalFlushIntervalMs(10);
 

python/python/lance/dataset.py

@@ -5086,16 +5086,13 @@ def initialize_mem_wal(
         identity_column: Optional[str] = None,
         unsharded: bool = False,
         durable_write: Optional[bool] = None,
-        sync_indexed_write: Optional[bool] = None,
         max_wal_buffer_size: Optional[int] = None,
         max_wal_flush_interval_ms: Optional[int] = None,
         max_memtable_size: Optional[int] = None,
         max_memtable_rows: Optional[int] = None,
         max_memtable_batches: Optional[int] = None,
         max_unflushed_memtable_bytes: Optional[int] = None,
         manifest_scan_batch_size: Optional[int] = None,
-        async_index_buffer_rows: Optional[int] = None,
-        async_index_interval_ms: Optional[int] = None,
         backpressure_log_interval_ms: Optional[int] = None,
         stats_log_interval_ms: Optional[int] = None,
         hnsw_params: Optional[Dict[str, Dict[str, int]]] = None,
@@ -5156,16 +5153,13 @@ def initialize_mem_wal(
             identity_column=identity_column,
             unsharded=unsharded,
             durable_write=durable_write,
-            sync_indexed_write=sync_indexed_write,
             max_wal_buffer_size=max_wal_buffer_size,
             max_wal_flush_interval_ms=max_wal_flush_interval_ms,
             max_memtable_size=max_memtable_size,
             max_memtable_rows=max_memtable_rows,
             max_memtable_batches=max_memtable_batches,
             max_unflushed_memtable_bytes=max_unflushed_memtable_bytes,
             manifest_scan_batch_size=manifest_scan_batch_size,
-            async_index_buffer_rows=async_index_buffer_rows,
-            async_index_interval_ms=async_index_interval_ms,
             backpressure_log_interval_ms=backpressure_log_interval_ms,
             stats_log_interval_ms=stats_log_interval_ms,
             hnsw_params=hnsw_params,
@@ -5188,16 +5182,13 @@ def mem_wal_writer(
         shard_id: str,
         *,
         durable_write: Optional[bool] = None,
-        sync_indexed_write: Optional[bool] = None,
         max_wal_buffer_size: Optional[int] = None,
         max_wal_flush_interval_ms: Optional[int] = None,
         max_memtable_size: Optional[int] = None,
         max_memtable_rows: Optional[int] = None,
         max_memtable_batches: Optional[int] = None,
         max_unflushed_memtable_bytes: Optional[int] = None,
         manifest_scan_batch_size: Optional[int] = None,
-        async_index_buffer_rows: Optional[int] = None,
-        async_index_interval_ms: Optional[int] = None,
         backpressure_log_interval_ms: Optional[int] = None,
         stats_log_interval_ms: Optional[int] = None,
         hnsw_params: Optional[Dict[str, Dict[str, int]]] = None,
@@ -5215,8 +5206,6 @@ def mem_wal_writer(
             ``str(uuid.uuid4())``).
         durable_write : bool, optional
             Whether to fsync WAL writes (default: ``True``).
-        sync_indexed_write : bool, optional
-            Whether index updates are synchronous (default: ``True``).
         max_wal_buffer_size : int, optional
             Maximum WAL buffer size in bytes (default: 10 MB).
         max_wal_flush_interval_ms : int, optional
@@ -5231,10 +5220,6 @@ def mem_wal_writer(
             Maximum unflushed bytes before backpressure (default: 1 GB).
         manifest_scan_batch_size : int, optional
             Batch size for manifest scans (default: 2).
-        async_index_buffer_rows : int, optional
-            Buffer rows for async index updates (default: 10 000).
-        async_index_interval_ms : int, optional
-            Interval for async index updates in milliseconds (default: 1000).
         backpressure_log_interval_ms : int, optional
             Interval for backpressure log messages in milliseconds
             (default: 30 000).
@@ -5282,16 +5267,13 @@ def mem_wal_writer(
             name: val
             for name, val in [
                 ("durable_write", durable_write),
-                ("sync_indexed_write", sync_indexed_write),
                 ("max_wal_buffer_size", max_wal_buffer_size),
                 ("max_wal_flush_interval_ms", max_wal_flush_interval_ms),
                 ("max_memtable_size", max_memtable_size),
                 ("max_memtable_rows", max_memtable_rows),
                 ("max_memtable_batches", max_memtable_batches),
                 ("max_unflushed_memtable_bytes", max_unflushed_memtable_bytes),
                 ("manifest_scan_batch_size", manifest_scan_batch_size),
-                ("async_index_buffer_rows", async_index_buffer_rows),
-                ("async_index_interval_ms", async_index_interval_ms),
                 ("backpressure_log_interval_ms", backpressure_log_interval_ms),
                 ("stats_log_interval_ms", stats_log_interval_ms),
                 ("hnsw_params", hnsw_params),

python/python/tests/test_mem_wal.py

@@ -209,7 +209,6 @@ def test_shard_writer_delete_binding_masks_base_row(tmp_path):
     with ds.mem_wal_writer(
         shard_id,
         durable_write=True,
-        sync_indexed_write=True,
         max_wal_buffer_size=1,
         max_wal_flush_interval_ms=10,
     ) as writer:
@@ -355,7 +354,6 @@ def test_shard_writer_e2e_correctness(tmp_path):
     writer = ds.mem_wal_writer(
         shard_id,
         durable_write=True,
-        sync_indexed_write=True,
         max_wal_buffer_size=10 * 1024,  # 10 KB
         max_wal_flush_interval_ms=50,
         max_memtable_size=80,  # flush after ~80 rows
@@ -405,9 +403,7 @@ def test_shard_writer_e2e_correctness(tmp_path):
     # === New writer: write and read back via active MemTable scanner ===
     ds2 = lance.dataset(ds_path)
     shard_id2 = str(uuid.uuid4())
-    with ds2.mem_wal_writer(
-        shard_id2, durable_write=False, sync_indexed_write=True
-    ) as writer2:
+    with ds2.mem_wal_writer(shard_id2, durable_write=False) as writer2:
         verify_batch = _e2e_batch(schema, start_id=10000, num_rows=10)
         writer2.put(pa.Table.from_batches([verify_batch]))
         result = writer2.lsm_scanner().to_table()
@@ -522,7 +518,6 @@ def test_initialize_mem_wal_writer_config_defaults(tmp_path):
     # Duration knobs are recorded in milliseconds with a `_ms` suffix.
     assert defaults["max_wal_flush_interval_ms"] == "250"
     # Every ShardWriterConfig tunable is recorded once any default is set.
-    assert "sync_indexed_write" in defaults
     assert "enable_memtable" in defaults