feat: propagate file_format_version to CommitBuilder.storageFormat()

lance-spark-3.4_2.12/src/main/java/org/lance/spark/write/SparkPositionDeltaWrite.java

@@ -207,6 +207,10 @@ public void commit(WriterCommitMessage[] messages) {
                 .writeParams(
                     LanceRuntime.mergeStorageOptions(
                         writeOptions.getStorageOptions(), initialStorageOptions));
+        String fileFormatVersion = writeOptions.getFileFormatVersion();
+        if (fileFormatVersion != null) {
+          commitBuilder.storageFormat(fileFormatVersion);
+        }
         if (dataset.hasStableRowIds()
             || Boolean.TRUE.equals(writeOptions.getEnableStableRowIds())) {
           commitBuilder.useStableRowIds(true);

lance-spark-3.5_2.12/src/main/java/org/lance/spark/write/SparkPositionDeltaWrite.java

@@ -228,6 +228,10 @@ public void commit(WriterCommitMessage[] messages) {
                 .writeParams(
                     LanceRuntime.mergeStorageOptions(
                         writeOptions.getStorageOptions(), initialStorageOptions));
+        String fileFormatVersion = writeOptions.getFileFormatVersion();
+        if (fileFormatVersion != null) {
+          commitBuilder.storageFormat(fileFormatVersion);
+        }
         if (useStableRowIds) {
           commitBuilder.useStableRowIds(true);
         }

lance-spark-base_2.12/src/main/java/org/lance/spark/BaseLanceNamespaceSparkCatalog.java

@@ -1125,6 +1125,7 @@ public StagedTable stageCreate(
         StagedCommitOptions.of(
             merged,
             catalogConfig.isEnableStableRowIds(properties),
+            fileFormatVersion,
             namespace,
             tableIdList,
             managedVersioning);
@@ -1161,7 +1162,9 @@ private StagedTable stageCreateAtPath(
     Schema arrowSchema = LanceArrowUtils.toArrowSchema(processedSchema, "UTC", true);
     final StagedCommitOptions commitOptions =
         StagedCommitOptions.pathBased(
-            catalogConfig.getStorageOptions(), catalogConfig.isEnableStableRowIds(properties));
+            catalogConfig.getStorageOptions(),
+            catalogConfig.isEnableStableRowIds(properties),
+            fileFormatVersion);
     StagedCommit stagedCommit = StagedCommit.forNewTable(arrowSchema, datasetUri, commitOptions);
     stagedCommit.setShardingSpec(shardingSpec);
     return createStagedDataset(
@@ -1201,19 +1204,20 @@ public StagedTable stageReplace(
     Dataset ds = Utils.openDatasetBuilder(resolved.readOptions).build();
     Map<String, String> merged =
         LanceRuntime.mergeStorageOptions(catalogConfig.getStorageOptions(), initialStorageOptions);
+    // Use specified file format version, or fall back to existing table's version
+    if (fileFormatVersion == null) {
+      fileFormatVersion = ds.getLanceFileFormatVersion();
+    }
     final StagedCommitOptions commitOptions =
         StagedCommitOptions.of(
             merged,
             catalogConfig.isEnableStableRowIds(properties),
+            fileFormatVersion,
             namespace,
             resolved.tableIdList,
             managedVersioning);
     StagedCommit stagedCommit = StagedCommit.forExistingTable(ds, arrowSchema, commitOptions);
     stagedCommit.setShardingSpec(shardingSpec);
-    // Use specified file format version, or fall back to existing table's version
-    if (fileFormatVersion == null) {
-      fileFormatVersion = ds.getLanceFileFormatVersion();
-    }
     return createStagedDataset(
         resolved.readOptions,
         processedSchema,
@@ -1250,16 +1254,18 @@ private StagedTable stageReplaceAtPath(
       throw new NoSuchTableException(ident);
     }
 
+    // Use specified file format version, or fall back to existing table's version
+    if (fileFormatVersion == null) {
+      fileFormatVersion = ds.getLanceFileFormatVersion();
+    }
     Schema arrowSchema = LanceArrowUtils.toArrowSchema(processedSchema, "UTC", true);
     final StagedCommitOptions commitOptions =
         StagedCommitOptions.pathBased(
-            catalogConfig.getStorageOptions(), catalogConfig.isEnableStableRowIds(properties));
+            catalogConfig.getStorageOptions(),
+            catalogConfig.isEnableStableRowIds(properties),
+            fileFormatVersion);
     StagedCommit stagedCommit = StagedCommit.forExistingTable(ds, arrowSchema, commitOptions);
     stagedCommit.setShardingSpec(shardingSpec);
-    // Use specified file format version, or fall back to existing table's version
-    if (fileFormatVersion == null) {
-      fileFormatVersion = ds.getLanceFileFormatVersion();
-    }
     return createStagedDataset(
         readOptions,
         processedSchema,
@@ -1333,24 +1339,32 @@ public StagedTable stageCreateOrReplace(
             name);
 
     Schema arrowSchema = LanceArrowUtils.toArrowSchema(processedSchema, "UTC", true);
-    // Use specified file format version, or fall back to existing table's version
     Map<String, String> merged =
         LanceRuntime.mergeStorageOptions(catalogConfig.getStorageOptions(), initialStorageOptions);
-    final StagedCommitOptions commitOptions =
-        StagedCommitOptions.of(
-            merged,
-            catalogConfig.isEnableStableRowIds(properties),
-            namespace,
-            tableIdList,
-            managedVersioning);
     StagedCommit stagedCommit;
     if (exists) {
       Dataset ds = Utils.openDatasetBuilder(readOptions).build();
-      stagedCommit = StagedCommit.forExistingTable(ds, arrowSchema, commitOptions);
       if (fileFormatVersion == null) {
         fileFormatVersion = ds.getLanceFileFormatVersion();
       }
+      final StagedCommitOptions commitOptions =
+          StagedCommitOptions.of(
+              merged,
+              catalogConfig.isEnableStableRowIds(properties),
+              fileFormatVersion,
+              namespace,
+              tableIdList,
+              managedVersioning);
+      stagedCommit = StagedCommit.forExistingTable(ds, arrowSchema, commitOptions);
     } else {
+      final StagedCommitOptions commitOptions =
+          StagedCommitOptions.of(
+              merged,
+              catalogConfig.isEnableStableRowIds(properties),
+              fileFormatVersion,
+              namespace,
+              tableIdList,
+              managedVersioning);
       stagedCommit = StagedCommit.forNewTable(arrowSchema, location, commitOptions);
     }
     stagedCommit.setShardingSpec(shardingSpec);
@@ -1384,18 +1398,24 @@ private StagedTable stageCreateOrReplaceAtPath(
 
     boolean exists = tableExistsAtPath(ident);
     Schema arrowSchema = LanceArrowUtils.toArrowSchema(processedSchema, "UTC", true);
-    final StagedCommitOptions commitOptions =
-        StagedCommitOptions.pathBased(
-            catalogConfig.getStorageOptions(), catalogConfig.isEnableStableRowIds(properties));
     StagedCommit stagedCommit;
-    // Use specified file format version, or fall back to existing table's version
     if (exists) {
       Dataset ds = Utils.openDatasetBuilder(readOptions).build();
-      stagedCommit = StagedCommit.forExistingTable(ds, arrowSchema, commitOptions);
       if (fileFormatVersion == null) {
         fileFormatVersion = ds.getLanceFileFormatVersion();
       }
+      final StagedCommitOptions commitOptions =
+          StagedCommitOptions.pathBased(
+              catalogConfig.getStorageOptions(),
+              catalogConfig.isEnableStableRowIds(properties),
+              fileFormatVersion);
+      stagedCommit = StagedCommit.forExistingTable(ds, arrowSchema, commitOptions);
     } else {
+      final StagedCommitOptions commitOptions =
+          StagedCommitOptions.pathBased(
+              catalogConfig.getStorageOptions(),
+              catalogConfig.isEnableStableRowIds(properties),
+              fileFormatVersion);
       stagedCommit = StagedCommit.forNewTable(arrowSchema, datasetUri, commitOptions);
     }
     stagedCommit.setShardingSpec(shardingSpec);

lance-spark-base_2.12/src/main/java/org/lance/spark/write/AddColumnsBackfillBatchWrite.java

@@ -149,14 +149,18 @@ public void commit(WriterCommitMessage[] messages) {
 
       // Commit merge operation using CommitBuilder
       Merge merge = Merge.builder().fragments(fragments).schema(arrowSchema).build();
+      CommitBuilder commitBuilder =
+          new CommitBuilder(dataset)
+              .writeParams(
+                  LanceRuntime.mergeStorageOptions(
+                      writeOptions.getStorageOptions(), initialStorageOptions));
+      String fileFormatVersion = writeOptions.getFileFormatVersion();
+      if (fileFormatVersion != null) {
+        commitBuilder.storageFormat(fileFormatVersion);
+      }
       try (Transaction txn =
               new Transaction.Builder().readVersion(version).operation(merge).build();
-          Dataset committed =
-              new CommitBuilder(dataset)
-                  .writeParams(
-                      LanceRuntime.mergeStorageOptions(
-                          writeOptions.getStorageOptions(), initialStorageOptions))
-                  .execute(txn)) {
+          Dataset committed = commitBuilder.execute(txn)) {
         // auto-close txn and committed dataset
       }
     }

lance-spark-base_2.12/src/main/java/org/lance/spark/write/LanceBatchWrite.java

@@ -178,6 +178,10 @@ public void commit(WriterCommitMessage[] messages) {
       if (enableStableRowIds != null) {
         stagedCommit.setEnableStableRowIds(enableStableRowIds);
       }
+      String fileFormatVersion = writeOptions.getFileFormatVersion();
+      if (fileFormatVersion != null) {
+        stagedCommit.setFileFormatVersion(fileFormatVersion);
+      }
       // For a path-based staged create, StagedCommit only has the catalog's static storage
       // options at this point. Merge in write-time and namespace-vended options now so
       // StagedCommit.commit() uses them. Mirrors the non-staged merge below.
@@ -202,6 +206,10 @@ public void commit(WriterCommitMessage[] messages) {
                 .writeParams(
                     LanceRuntime.mergeStorageOptions(
                         writeOptions.getStorageOptions(), initialStorageOptions));
+        String fileFormatVersion = writeOptions.getFileFormatVersion();
+        if (fileFormatVersion != null) {
+          commitBuilder.storageFormat(fileFormatVersion);
+        }
         // When enableStableRowIds is null (user didn't pass the option),
         // lance-core auto-inherits the flag from the existing manifest.
         // Appending to a table with stable row IDs works without

lance-spark-base_2.12/src/main/java/org/lance/spark/write/StagedCommit.java

@@ -53,6 +53,7 @@ public class StagedCommit {
   // The non-staged path (LanceBatchWrite) uses boxed Boolean because null means
   // "user didn't specify" and lets lance-core inherit the flag from the manifest.
   private boolean enableStableRowIds;
+  private String fileFormatVersion;
   private List<FragmentMetadata> fragments;
   private Schema schema;
   private ShardingSpec shardingSpec;
@@ -95,6 +96,7 @@ private StagedCommit(
     this.storageOptions = new HashMap<>(options.getStorageOptions());
     this.isNewTable = datasetUri != null;
     this.enableStableRowIds = options.isEnableStableRowIds();
+    this.fileFormatVersion = options.getFileFormatVersion();
     this.namespace = options.getNamespace();
     this.tableId = options.getTableId();
     this.managedVersioning = options.isManagedVersioning();
@@ -112,6 +114,10 @@ public void setEnableStableRowIds(boolean enableStableRowIds) {
     this.enableStableRowIds = enableStableRowIds;
   }
 
+  public void setFileFormatVersion(String fileFormatVersion) {
+    this.fileFormatVersion = fileFormatVersion;
+  }
+
   public void setShardingSpec(ShardingSpec shardingSpec) {
     this.shardingSpec = shardingSpec;
   }
@@ -134,6 +140,11 @@ Map<String, String> getStorageOptions() {
     return storageOptions;
   }
 
+  /** Returns the current file format version. Visible for testing. */
+  String getFileFormatVersion() {
+    return fileFormatVersion;
+  }
+
   /** Performs the actual commit using the stored dataset and fragments. */
   public void commit() {
     if (dataset.isEmpty()) {
@@ -150,6 +161,9 @@ private void commitNewTable() {
     if (enableStableRowIds) {
       builder.useStableRowIds(true);
     }
+    if (fileFormatVersion != null) {
+      builder.storageFormat(fileFormatVersion);
+    }
     applyManagedVersioning(builder);
     try (Transaction txn = new Transaction.Builder().operation(operation).build();
         Dataset committed = builder.execute(txn)) {
@@ -167,6 +181,9 @@ private void commitExistingTable() {
     final CommitBuilder builder =
         new CommitBuilder(uri, LanceRuntime.allocator()).writeParams(storageOptions);
     builder.useStableRowIds(enableStableRowIds);
+    if (fileFormatVersion != null) {
+      builder.storageFormat(fileFormatVersion);
+    }
     applyManagedVersioning(builder);
     try (Dataset committed = commitOperation(builder, version, operation)) {
       SparkLanceShardingUtils.initializeMemWal(committed, shardingSpec);

lance-spark-base_2.12/src/main/java/org/lance/spark/write/StagedCommitOptions.java

@@ -27,18 +27,21 @@ public class StagedCommitOptions {
 
   private final Map<String, String> storageOptions;
   private final boolean enableStableRowIds;
+  private final String fileFormatVersion;
   private final LanceNamespace namespace;
   private final List<String> tableId;
   private final boolean managedVersioning;
 
   private StagedCommitOptions(
       final Map<String, String> storageOptions,
       final boolean enableStableRowIds,
+      final String fileFormatVersion,
       final LanceNamespace namespace,
       final List<String> tableId,
       final boolean managedVersioning) {
     this.storageOptions = storageOptions;
     this.enableStableRowIds = enableStableRowIds;
+    this.fileFormatVersion = fileFormatVersion;
     this.namespace = namespace;
     this.tableId = tableId;
     this.managedVersioning = managedVersioning;
@@ -47,17 +50,30 @@ private StagedCommitOptions(
   public static StagedCommitOptions of(
       final Map<String, String> storageOptions,
       final boolean enableStableRowIds,
+      final String fileFormatVersion,
       final LanceNamespace namespace,
       final List<String> tableId,
       final boolean managedVersioning) {
     return new StagedCommitOptions(
-        storageOptions, enableStableRowIds, namespace, tableId, managedVersioning);
+        storageOptions,
+        enableStableRowIds,
+        fileFormatVersion,
+        namespace,
+        tableId,
+        managedVersioning);
   }
 
   public static StagedCommitOptions pathBased(
-      final Map<String, String> storageOptions, final boolean enableStableRowIds) {
+      final Map<String, String> storageOptions,
+      final boolean enableStableRowIds,
+      final String fileFormatVersion) {
     return new StagedCommitOptions(
-        storageOptions, enableStableRowIds, NO_NAMESPACE, NO_TABLE_ID, NO_MANAGED_VERSIONING);
+        storageOptions,
+        enableStableRowIds,
+        fileFormatVersion,
+        NO_NAMESPACE,
+        NO_TABLE_ID,
+        NO_MANAGED_VERSIONING);
   }
 
   public Map<String, String> getStorageOptions() {
@@ -68,6 +84,10 @@ public boolean isEnableStableRowIds() {
     return enableStableRowIds;
   }
 
+  public String getFileFormatVersion() {
+    return fileFormatVersion;
+  }
+
   public LanceNamespace getNamespace() {
     return namespace;
   }

lance-spark-base_2.12/src/main/java/org/lance/spark/write/UpdateColumnsBackfillBatchWrite.java

@@ -156,14 +156,18 @@ public void commit(WriterCommitMessage[] messages) {
           Objects.requireNonNull(
               writeOptions.getVersion(),
               "version must be set (resolved in UpdateColumnsBackfillBatchWrite constructor)");
+      CommitBuilder commitBuilder =
+          new CommitBuilder(dataset)
+              .writeParams(
+                  LanceRuntime.mergeStorageOptions(
+                      writeOptions.getStorageOptions(), initialStorageOptions));
+      String fileFormatVersion = writeOptions.getFileFormatVersion();
+      if (fileFormatVersion != null) {
+        commitBuilder.storageFormat(fileFormatVersion);
+      }
       try (Transaction txn =
               new Transaction.Builder().readVersion(version).operation(update).build();
-          Dataset committed =
-              new CommitBuilder(dataset)
-                  .writeParams(
-                      LanceRuntime.mergeStorageOptions(
-                          writeOptions.getStorageOptions(), initialStorageOptions))
-                  .execute(txn)) {
+          Dataset committed = commitBuilder.execute(txn)) {
         // auto-close txn and committed dataset
       }
     }