feat: expose exact V2 versions in bindings

Author: Xuanwo

docs/src/format/table/versioning.md

@@ -36,3 +36,64 @@ they should return an "unsupported" error on any read or write operation.
 </div>
 
 Flags with bit values 512 and above are unknown and will cause implementations to reject the dataset with an "unsupported" error. The paired mixed-version reader and writer bits must either both be set or both be clear; a half-set manifest is invalid.
+
+## Mixed V2 Data File Versions
+
+The manifest data storage version is the default for operations that do not
+select an exact output version. It is a fallback, not a summary, minimum,
+maximum, or profile of the data files referenced by the snapshot. Once
+`FLAG_MIXED_DATA_FILE_VERSIONS` is enabled, each base data file and data overlay
+file is decoded according to its own normalized version identity.
+
+Mixed snapshots have the following invariants:
+
+- Only exact V2.0, V2.1, V2.2, and V2.3 data file versions may be mixed.
+- V1 and V2 data files may not appear in the same snapshot.
+- A commit that first produces a mixed snapshot derives and sets both the reader
+  and writer capability bits from its final manifest. The bits remain set on all
+  later snapshots, even if a later compaction makes the files homogeneous again.
+- A snapshot without the capability may only reference files matching its
+  manifest fallback. The only repair exception is an unambiguous, homogeneous
+  historical V2 snapshot whose legacy manifest metadata is stale.
+- An operation-level `data_storage_version` selects the exact output version
+  for that operation. Omitting it uses the manifest fallback. Neither case
+  changes the fallback.
+
+For example, a dataset whose fallback is V2.1 can append V2.2 files by setting
+`data_storage_version="2.2"`. The same commit adds both mixed-version capability
+bits. Reads then dispatch V2.1 files to the V2.1 decoder and V2.2 files to the
+V2.2 decoder. Compaction can deliberately rewrite selected fragments to any
+supported exact V2 target; binary copy is only valid when every selected input
+file already has that exact target version.
+
+### Compatibility Matrix
+
+| Dataset state | Mixed-aware client | Client without bit 256 support |
+| --- | --- | --- |
+| Historical homogeneous V1 | Reads and writes through legacy paths | Unchanged |
+| Historical homogeneous V2 | Reads and writes; legacy metadata repair remains uniform-only | Unchanged |
+| New homogeneous V2 without bit 256 | Reads and writes using the manifest fallback | Unchanged |
+| Mixed V2.0-V2.3 with both bits set | Reads and writes by exact per-file identity | Rejects before reading or writing |
+| Mixed V2 without both bits | Rejects as a per-file capability mismatch | Not a valid dataset state |
+| V1/V2 mixture | Rejects | Not a valid dataset state |
+
+### Error Categories
+
+Implementations distinguish these failures in their error messages so operators
+can identify the violated boundary:
+
+- unsupported reader or writer feature bit;
+- half-set mixed-version capability corruption;
+- unknown or malformed data file version identity;
+- V1/V2 mixture;
+- a non-fallback file without mixed-version capability; and
+- binary-copy target mismatch, including the target, actual version, and path.
+
+### Rollout Gate
+
+Before the first mixed-version commit, deploy mixed-aware readers and writers
+everywhere that can access the dataset. Then drain or fence writers that opened
+the dataset with an older client. Only after both steps may a writer select a
+different exact V2 output version. The capability bit makes clients that open
+the resulting snapshot fail closed, but it cannot retroactively fence an old
+writer that read an earlier manifest.

java/lance-jni/src/blocking_dataset.rs

@@ -3340,6 +3340,14 @@ fn convert_java_compaction_options_to_rust(
             &[],
         )?
         .l()?;
+    let data_storage_version = env
+        .call_method(
+            &java_options,
+            "getDataStorageVersion",
+            "()Ljava/util/Optional;",
+            &[],
+        )?
+        .l()?;
 
     build_compaction_options(
         env,
@@ -3357,6 +3365,7 @@ fn convert_java_compaction_options_to_rust(
         &max_source_rows,
         &max_source_bytes,
         &excluded_fragment_ids,
+        &data_storage_version,
         config,
     )
 }

java/lance-jni/src/merge_insert.rs

@@ -15,7 +15,9 @@ use lance::dataset::{
     MergeInsertBuilder, MergeStats, WhenMatched, WhenNotMatched, WhenNotMatchedBySource,
 };
 use lance_core::datatypes::Schema;
+use lance_file::version::LanceFileVersion;
 use lance_index::mem_wal::CompactedSsTable;
+use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
 use uuid::Uuid;
@@ -53,6 +55,7 @@ fn inner_merge_insert<'local>(
     let skip_auto_cleanup = extract_skip_auto_cleanup(env, &jparam)?;
     let use_index = extract_use_index(env, &jparam)?;
     let compacted_sstables = extract_compacted_sstables(env, &jparam)?;
+    let data_storage_version = extract_data_storage_version(env, &jparam)?;
 
     let (new_ds, merge_stats) = unsafe {
         let dataset = env.get_rust_field::<_, _, BlockingDataset>(jdataset, NATIVE_DATASET)?;
@@ -63,7 +66,11 @@ fn inner_merge_insert<'local>(
             when_not_matched_by_source_delete_expr,
         )?;
 
-        let merge_insert_job = MergeInsertBuilder::try_new(Arc::new(dataset.clone().inner), on)?
+        let mut builder = MergeInsertBuilder::try_new(Arc::new(dataset.clone().inner), on)?;
+        if let Some(version) = data_storage_version {
+            builder.data_storage_version(LanceFileVersion::from_str(&version)?);
+        }
+        let merge_insert_job = builder
             .when_matched(when_matched)
             .when_not_matched(when_not_matched)
             .when_not_matched_by_source(when_not_matched_by_source)
@@ -241,6 +248,16 @@ fn extract_use_index<'local>(env: &mut JNIEnv<'local>, jparam: &JObject) -> Resu
     Ok(use_index)
 }
 
+fn extract_data_storage_version<'local>(
+    env: &mut JNIEnv<'local>,
+    jparam: &JObject,
+) -> Result<Option<String>> {
+    let version = env
+        .call_method(jparam, "dataStorageVersion", "()Ljava/util/Optional;", &[])?
+        .l()?;
+    env.get_string_opt(&version)
+}
+
 fn extract_compacted_sstables<'local>(
     env: &mut JNIEnv<'local>,
     jparam: &JObject,

java/lance-jni/src/optimize.rs

@@ -17,7 +17,7 @@ use lance::dataset::{
 };
 
 use crate::{
-    block_on,
+    JNIEnvExt, block_on,
     blocking_dataset::{BlockingDataset, NATIVE_DATASET},
     traits::{
         FromJObjectWithEnv, IntoJava, export_vec, import_vec_from_method, import_vec_to_rust,
@@ -49,6 +49,7 @@ pub extern "system" fn Java_org_lance_compaction_Compaction_nativePlanCompaction
     max_source_rows: JObject,                 // Optional<Long>
     max_source_bytes: JObject,                // Optional<Long>
     excluded_fragment_ids: JObject,           // List<Long>
+    data_storage_version: JObject,            // Optional<String>
 ) -> JObject<'local> {
     ok_or_throw_with_return!(
         env,
@@ -68,7 +69,8 @@ pub extern "system" fn Java_org_lance_compaction_Compaction_nativePlanCompaction
             max_source_fragments,
             max_source_rows,
             max_source_bytes,
-            excluded_fragment_ids
+            excluded_fragment_ids,
+            data_storage_version
         ),
         JObject::null()
     )
@@ -92,6 +94,7 @@ fn inner_plan_compaction<'local>(
     max_source_rows: JObject,                 // Optional<Long>
     max_source_bytes: JObject,                // Optional<Long>
     excluded_fragment_ids: JObject,           // List<Long>
+    data_storage_version: JObject,            // Optional<String>
 ) -> Result<JObject<'local>> {
     let config = {
         let dataset =
@@ -114,6 +117,7 @@ fn inner_plan_compaction<'local>(
         &max_source_rows,
         &max_source_bytes,
         &excluded_fragment_ids,
+        &data_storage_version,
         &config,
     )?;
 
@@ -145,6 +149,7 @@ pub extern "system" fn Java_org_lance_compaction_Compaction_nativeCommitCompacti
     max_source_rows: JObject,                 // Optional<Long>
     max_source_bytes: JObject,                // Optional<Long>
     excluded_fragment_ids: JObject,           // List<Long>
+    data_storage_version: JObject,            // Optional<String>
 ) -> JObject<'local> {
     ok_or_throw_with_return!(
         env,
@@ -166,6 +171,7 @@ pub extern "system" fn Java_org_lance_compaction_Compaction_nativeCommitCompacti
             max_source_rows,
             max_source_bytes,
             excluded_fragment_ids,
+            data_storage_version,
         ),
         JObject::null()
     )
@@ -190,6 +196,7 @@ fn inner_commit_compaction<'local>(
     max_source_rows: JObject,                 // Optional<Long>
     max_source_bytes: JObject,                // Optional<Long>
     excluded_fragment_ids: JObject,           // List<Long>
+    data_storage_version: JObject,            // Optional<String>
 ) -> Result<JObject<'local>> {
     let config = {
         let dataset =
@@ -212,6 +219,7 @@ fn inner_commit_compaction<'local>(
         &max_source_rows,
         &max_source_bytes,
         &excluded_fragment_ids,
+        &data_storage_version,
         &config,
     )?;
     let completed_tasks = import_vec_to_rust(env, &rewrite_results, |env, rewrite_result| {
@@ -252,6 +260,7 @@ pub extern "system" fn Java_org_lance_compaction_CompactionTask_nativeExecute<'l
     max_source_rows: JObject,                 // Optional<Long>
     max_source_bytes: JObject,                // Optional<Long>
     excluded_fragment_ids: JObject,           // List<Long>
+    data_storage_version: JObject,            // Optional<String>
 ) -> JObject<'local> {
     ok_or_throw_with_return!(
         env,
@@ -273,7 +282,8 @@ pub extern "system" fn Java_org_lance_compaction_CompactionTask_nativeExecute<'l
             max_source_fragments,
             max_source_rows,
             max_source_bytes,
-            excluded_fragment_ids
+            excluded_fragment_ids,
+            data_storage_version
         ),
         JObject::null()
     )
@@ -299,6 +309,7 @@ fn inner_execute_task<'local>(
     max_source_rows: JObject,                 // Optional<Long>
     max_source_bytes: JObject,                // Optional<Long>
     excluded_fragment_ids: JObject,           // List<Long>
+    data_storage_version: JObject,            // Optional<String>
 ) -> Result<JObject<'local>> {
     let task_data: TaskData = task_data.extract_object(env)?;
     let config = {
@@ -322,6 +333,7 @@ fn inner_execute_task<'local>(
         &max_source_rows,
         &max_source_bytes,
         &excluded_fragment_ids,
+        &data_storage_version,
         &config,
     )?;
     let compaction_task = CompactionTask {
@@ -345,11 +357,10 @@ const COMPACTION_PLAN_CLASS: &str = "org/lance/compaction/CompactionPlan";
 const COMPACTION_PLAN_CONSTRUCTOR_SIG: &str =
     "(Ljava/util/List;JLorg/lance/compaction/CompactionOptions;)V";
 const REWRITE_RESULT_CLASS: &str = "org/lance/compaction/RewriteResult";
-const REWRITE_RESULT_CONSTRUCTOR_SIG: &str =
-    "(Lorg/lance/compaction/CompactionMetrics;Ljava/util/List;Ljava/util/List;J[B)V";
+const REWRITE_RESULT_CONSTRUCTOR_SIG: &str = "(Lorg/lance/compaction/CompactionMetrics;Ljava/util/List;Ljava/util/List;J[BLjava/lang/String;)V";
 const COMPACTION_OPTIONS_CLASS: &str = "org/lance/compaction/CompactionOptions";
 const COMPACTION_MODE_CLASS: &str = "org/lance/compaction/CompactionMode";
-const COMPACTION_OPTIONS_CONSTRUCTOR_SIG: &str = "(Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/List;)V";
+const COMPACTION_OPTIONS_CONSTRUCTOR_SIG: &str = "(Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/Optional;Ljava/util/List;Ljava/util/Optional;)V";
 
 impl IntoJava for &TaskData {
     fn into_java<'a>(self, env: &mut JNIEnv<'a>) -> Result<JObject<'a>> {
@@ -431,6 +442,11 @@ impl IntoJava for &CompactionOptions {
             .map(|fragment_id| to_java_long_obj(env, Some(*fragment_id as i64)))
             .collect::<Result<Vec<_>>>()?;
         let excluded_fragment_ids = to_java_list(env, &excluded_fragment_ids)?;
+        let data_storage_version = match self.data_storage_version {
+            Some(version) => env.new_string(version.to_string())?.into(),
+            None => JObject::null(),
+        };
+        let data_storage_version_opt = to_java_optional(env, data_storage_version)?;
 
         Ok(env.new_object(
             COMPACTION_OPTIONS_CLASS,
@@ -450,6 +466,7 @@ impl IntoJava for &CompactionOptions {
                 JValueGen::Object(&max_source_rows_opt),
                 JValueGen::Object(&max_source_bytes_opt),
                 JValueGen::Object(&excluded_fragment_ids),
+                JValueGen::Object(&data_storage_version_opt),
             ],
         )?)
     }
@@ -481,6 +498,7 @@ impl IntoJava for &RewriteResult {
         } else {
             JObject::null()
         };
+        let write_version: JObject<'_> = env.new_string(&self.write_version)?.into();
         Ok(env.new_object(
             REWRITE_RESULT_CLASS,
             REWRITE_RESULT_CONSTRUCTOR_SIG,
@@ -490,6 +508,7 @@ impl IntoJava for &RewriteResult {
                 JValueGen::Object(&original_fragments),
                 JValueGen::Long(self.read_version as i64),
                 JValueGen::Object(&row_addrs),
+                JValueGen::Object(&write_version),
             ],
         )?)
     }
@@ -554,12 +573,17 @@ impl FromJObjectWithEnv<RewriteResult> for JObject<'_> {
         } else {
             Some(env.convert_byte_array(row_addrs_obj)?)
         };
+        let write_version_obj = env
+            .call_method(self, "getWriteVersion", "()Ljava/util/Optional;", &[])?
+            .l()?;
+        let write_version = env.get_string_opt(&write_version_obj)?.unwrap_or_default();
         Ok(RewriteResult {
             metrics,
             new_fragments,
             read_version,
             original_fragments,
             row_addrs,
+            write_version,
         })
     }
 }

java/lance-jni/src/update.rs

@@ -9,6 +9,8 @@ use crate::{JNIEnvExt, block_on};
 use jni::JNIEnv;
 use jni::objects::{JMap, JObject, JValueGen};
 use lance::dataset::UpdateBuilder;
+use lance_file::version::LanceFileVersion;
+use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -30,6 +32,7 @@ fn inner_update<'local>(
     let where_clause = extract_where(env, &jparam)?;
     let conflict_retries = extract_conflict_retries(env, &jparam)?;
     let retry_timeout_ms = extract_retry_timeout_ms(env, &jparam)?;
+    let data_storage_version = extract_data_storage_version(env, &jparam)?;
 
     // Clone the inner Dataset out of the `get_rust_field` guard and drop the
     // guard before running the long-lived async update. Otherwise the guard
@@ -44,6 +47,10 @@ fn inner_update<'local>(
         .conflict_retries(conflict_retries)
         .retry_timeout(Duration::from_millis(retry_timeout_ms));
 
+    if let Some(version) = data_storage_version {
+        builder = builder.data_storage_version(LanceFileVersion::from_str(&version)?);
+    }
+
     if let Some(predicate) = where_clause {
         builder = builder.update_where(&predicate)?;
     }
@@ -96,6 +103,16 @@ fn extract_retry_timeout_ms<'local>(env: &mut JNIEnv<'local>, jparam: &JObject)
     Ok(timeout_ms)
 }
 
+fn extract_data_storage_version<'local>(
+    env: &mut JNIEnv<'local>,
+    jparam: &JObject,
+) -> Result<Option<String>> {
+    let version = env
+        .call_method(jparam, "dataStorageVersion", "()Ljava/util/Optional;", &[])?
+        .l()?;
+    env.get_string_opt(&version)
+}
+
 const UPDATE_RESULT_CLASS: &str = "org/lance/update/UpdateResult";
 const UPDATE_RESULT_CONSTRUCTOR_SIG: &str = "(Lorg/lance/Dataset;J)V";
 

java/lance-jni/src/utils.rs

@@ -194,6 +194,7 @@ pub fn build_compaction_options(
     max_source_rows: &JObject,                 // Optional<Long>
     max_source_bytes: &JObject,                // Optional<Long>
     excluded_fragment_ids: &JObject,           // List<Long>
+    data_storage_version: &JObject,            // Optional<String>
     config: &std::collections::HashMap<String, String>,
 ) -> Result<CompactionOptions> {
     let mut compaction_options = CompactionOptions::from_dataset_config(config)?;
@@ -256,6 +257,9 @@ pub fn build_compaction_options(
             })
         })
         .collect::<Result<Vec<_>>>()?;
+    if let Some(version) = env.get_string_opt(data_storage_version)? {
+        compaction_options.data_storage_version = Some(LanceFileVersion::from_str(&version)?);
+    }
 
     Ok(compaction_options)
 }