feat: enable the fragment reuse index on stable row id datasets

python/python/tests/test_optimize.py

@@ -542,25 +542,41 @@ def test_index_remapping_multiple_rewrite_tasks(tmp_path: Path):
     assert index_frag_ids[0] in frag_ids
 
 
-def test_defer_index_remap(tmp_path: Path):
+@pytest.mark.parametrize("enable_stable_row_ids", [False, True])
+def test_defer_index_remap(tmp_path: Path, enable_stable_row_ids: bool):
     base_dir = tmp_path / "dataset"
     data = pa.table({"i": range(6_000), "val": range(6_000)})
-    dataset = lance.write_dataset(data, base_dir, max_rows_per_file=1_000)
+    dataset = lance.write_dataset(
+        data,
+        base_dir,
+        max_rows_per_file=1_000,
+        enable_stable_row_ids=enable_stable_row_ids,
+    )
     dataset.create_scalar_index("i", "BTREE")
     options = dict(
         target_rows_per_fragment=2_000, defer_index_remap=True, num_threads=1
     )
 
     dataset.delete("i < 500")
+    row_ids_before = dataset.to_table(with_row_id=True).column("_rowid").to_pylist()
     dataset.optimize.compact_files(**options)
 
     dataset = lance.dataset(base_dir)
     indices = dataset.describe_indices()
     assert any(idx.name == "__lance_frag_reuse" for idx in indices)
+    row_ids_after = dataset.to_table(with_row_id=True).column("_rowid").to_pylist()
+    if enable_stable_row_ids:
+        assert row_ids_after == row_ids_before
+    # Stable ids 500..1000 collide with the rewritten fragment's addresses; the
+    # index must still answer with the ids.
+    assert dataset.to_table(filter="i < 510").num_rows == 10
 
 
 @pytest.mark.parametrize("use_commit_options", [True, False])
-def test_defer_index_remap_via_commit_options(tmp_path: Path, use_commit_options: bool):
+@pytest.mark.parametrize("enable_stable_row_ids", [False, True])
+def test_defer_index_remap_via_commit_options(
+    tmp_path: Path, use_commit_options: bool, enable_stable_row_ids: bool
+):
     """Compaction.commit respects defer_index_remap passed in options.
 
     When options={"defer_index_remap": True} is supplied to Compaction.commit
@@ -569,7 +585,12 @@ def test_defer_index_remap_via_commit_options(tmp_path: Path, use_commit_options
     """
     base_dir = tmp_path / f"dataset_commit_opts_{use_commit_options}"
     data = pa.table({"i": range(6_000), "val": range(6_000)})
-    dataset = lance.write_dataset(data, base_dir, max_rows_per_file=1_000)
+    dataset = lance.write_dataset(
+        data,
+        base_dir,
+        max_rows_per_file=1_000,
+        enable_stable_row_ids=enable_stable_row_ids,
+    )
     dataset.create_scalar_index("i", "BTREE")
     dataset.delete("i < 500")
 

rust/lance-index/src/scalar/json.rs

@@ -123,6 +123,10 @@ impl ScalarIndex for JsonIndex {
             .await
     }
 
+    fn results_are_row_addresses(&self) -> bool {
+        self.target_index.results_are_row_addresses()
+    }
+
     fn can_remap(&self) -> bool {
         self.target_index.can_remap()
     }
@@ -1008,14 +1012,13 @@ impl ScalarIndexPlugin for JsonIndexPlugin {
         index_name: String,
         index_details: &prost_types::Any,
     ) -> Option<Box<dyn ScalarQueryParser>> {
-        // TODO: Allow return Result here
-        let registry = self.registry().unwrap();
+        let registry = self.registry().ok()?;
         let json_details =
-            crate::pb::JsonIndexDetails::decode(index_details.value.as_slice()).unwrap();
-        let target_details = json_details.target_details.as_ref().expect_ok().unwrap();
-        let target_plugin = registry.get_plugin_by_details(target_details).unwrap();
+            crate::pb::JsonIndexDetails::decode(index_details.value.as_slice()).ok()?;
+        let target_details = json_details.target_details.as_ref()?;
+        let target_plugin = registry.get_plugin_by_details(target_details).ok()?;
         // TODO: Use something like ${index_name}_${path} for the index name?  Don't have access to path here tho
-        let target_parser = target_plugin.new_query_parser(index_name, index_details)?;
+        let target_parser = target_plugin.new_query_parser(index_name, target_details)?;
         Some(Box::new(JsonQueryParser::new(
             json_details.path.clone(),
             target_parser,
@@ -1063,6 +1066,35 @@ mod tests {
     use std::ops::Bound;
     use std::sync::Arc;
 
+    #[test]
+    fn test_nested_json_query_parser() {
+        use prost::Message;
+        let json_type_url = "type.googleapis.com/lance.index.pb.JsonIndexDetails".to_string();
+        let btree = prost_types::Any {
+            type_url: "type.googleapis.com/lance.table.BTreeIndexDetails".to_string(),
+            value: vec![],
+        };
+        let inner = prost_types::Any {
+            type_url: json_type_url.clone(),
+            value: crate::pb::JsonIndexDetails {
+                path: "a".to_string(),
+                target_details: Some(btree),
+            }
+            .encode_to_vec(),
+        };
+        let outer = prost_types::Any {
+            type_url: json_type_url,
+            value: crate::pb::JsonIndexDetails {
+                path: "b".to_string(),
+                target_details: Some(inner),
+            }
+            .encode_to_vec(),
+        };
+        let registry = crate::registry::IndexPluginRegistry::with_default_plugins();
+        let plugin = registry.get_plugin_by_details(&outer).unwrap();
+        assert!(plugin.new_query_parser("idx".to_string(), &outer).is_some());
+    }
+
     // Note: The old test_detect_json_value_type test has been removed as we now use
     // JSONB's inherent type information instead of string-based type detection
 

rust/lance-table/src/feature_flags.rs

@@ -56,9 +56,10 @@ pub const FLAG_COVERED_INDEX_METADATA: u64 = 1 << 7;
 pub const FLAG_MIXED_DATA_FILE_VERSIONS: u64 = 1 << 8;
 /// The table uses stable row ids and carries a fragment reuse index.
 ///
-/// Reserved ahead of its implementation. This build treats the bit as unknown
-/// (see `supported_flags_when`), so a build that knows the flag but not the
-/// handling behind it cannot open such a table.
+/// Older readers and writers did not expect the combination and could corrupt
+/// such a table, so both feature words carry the bit. Set by `build_manifest`
+/// whenever both hold, and lifted when either stops holding; see
+/// `supported_flags_when`.
 pub const FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS: u64 = 1 << 9;
 /// Tagged FRI requires a reader that interprets its mappings and a writer that
 /// preserves them during maintenance. Legacy-only FRI does not set this bit.
@@ -91,12 +92,12 @@ pub fn apply_feature_flags(
     disable_transaction_file: bool,
 ) -> Result<()> {
     // Carried across the reset: a `Manifest` only points at its index section,
-    // so whether any index declares covering columns is not visible here. `build_manifest` decides it from the index list it is
-    // committing and sets the bit after calling this; without the carry the
-    // second call, from `write_manifest_file`, would clear that decision
-    // immediately before the write.
-    let covered_index_metadata = (manifest.reader_feature_flags | manifest.writer_feature_flags)
-        & FLAG_COVERED_INDEX_METADATA;
+    // so what its indices declare is not visible here. `build_manifest` decides
+    // these from the index list it is committing and sets the bits after calling
+    // this; without the carry the second call, from `write_manifest_file`, would
+    // clear that decision immediately before the write.
+    let index_derived_flags = (manifest.reader_feature_flags | manifest.writer_feature_flags)
+        & (FLAG_COVERED_INDEX_METADATA | FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS);
     let sticky_paired_flags = validated_sticky_paired_flags(manifest)?;
 
     // Reset flags
@@ -157,8 +158,8 @@ pub fn apply_feature_flags(
         manifest.writer_feature_flags |= FLAG_DISABLE_TRANSACTION_FILE;
     }
 
-    manifest.reader_feature_flags |= covered_index_metadata;
-    manifest.writer_feature_flags |= covered_index_metadata;
+    manifest.reader_feature_flags |= index_derived_flags;
+    manifest.writer_feature_flags |= index_derived_flags;
     manifest.reader_feature_flags |= sticky_paired_flags;
     manifest.writer_feature_flags |= sticky_paired_flags;
 
@@ -198,25 +199,30 @@ fn mark_supported(flags: &mut u64, flag: u64, feature_enabled: bool) {
 }
 
 /// The feature-flag bits this build understands, given whether overlay support
-/// is enabled. Split out from [`supported_flags`] so the policy is testable
-/// without toggling the build profile or environment.
-fn supported_flags_when(overlay_enabled: bool) -> u64 {
+/// is enabled and whether a fragment reuse index on a stable-row-id table is
+/// supported. Split out from [`supported_flags`] so the policy is testable
+/// without toggling the build profile or environment, and so the compatibility
+/// test can show what a build without that support does with such a table.
+fn supported_flags_when(overlay_enabled: bool, frag_reuse_with_stable_row_ids: bool) -> u64 {
     let mut supported = FLAG_UNKNOWN - 1;
     mark_supported(
         &mut supported,
         FLAG_UNSTABLE_DATA_OVERLAY_FILES,
         overlay_enabled,
     );
-    // Reserved, not implemented: see the flag's doc comment.
-    mark_supported(&mut supported, FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS, false);
+    mark_supported(
+        &mut supported,
+        FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS,
+        frag_reuse_with_stable_row_ids,
+    );
     // Bit 10 now falls below the unknown boundary, so keep tagged FRI refused
     // until its reader/writer handling lands.
     mark_supported(&mut supported, FLAG_FRAGMENT_REUSE_INDEX, false);
     supported
 }
 
 fn supported_flags() -> u64 {
-    supported_flags_when(data_overlay_files_enabled())
+    supported_flags_when(data_overlay_files_enabled(), true)
 }
 
 pub fn can_read_dataset(reader_flags: u64) -> bool {
@@ -314,18 +320,25 @@ mod tests {
     use super::*;
     use crate::format::BasePath;
 
-    /// Reserved ahead of its implementation: refused for reading and writing
-    /// until the handling lands, so a build from the gap cannot open the table.
+    /// A build without support for the combination must refuse the table for
+    /// reading and writing, or it could corrupt it; this build accepts it.
     #[test]
-    fn test_frag_reuse_with_stable_row_ids_flag_is_reserved_not_supported() {
+    fn test_frag_reuse_with_stable_row_ids_flag_gating() {
         use crate::format::{DataStorageFormat, Manifest};
         use arrow_schema::{Field as ArrowField, Schema as ArrowSchema};
         use lance_core::datatypes::Schema;
         use std::collections::HashMap;
         use std::sync::Arc;
 
-        assert!(!can_read_dataset(FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS));
-        assert!(!can_write_dataset(FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS));
+        let without_support = supported_flags_when(true, false);
+        assert_eq!(
+            without_support & FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS,
+            0,
+            "a build without the implementation must not accept the flag"
+        );
+        assert_ne!(without_support & FLAG_STABLE_ROW_IDS, 0);
+        assert!(can_read_dataset(FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS));
+        assert!(can_write_dataset(FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS));
 
         let arrow_schema = ArrowSchema::new(vec![ArrowField::new(
             "id",
@@ -340,14 +353,13 @@ mod tests {
         );
         manifest.reader_feature_flags = FLAG_STABLE_ROW_IDS | FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS;
         manifest.writer_feature_flags = FLAG_STABLE_ROW_IDS | FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS;
-        assert!(matches!(
-            ensure_can_read_manifest(&manifest).unwrap_err(),
-            Error::NotSupported { .. }
-        ));
-        assert!(matches!(
-            ensure_can_write_manifest(&manifest).unwrap_err(),
-            Error::NotSupported { .. }
-        ));
+        ensure_can_read_manifest(&manifest).unwrap();
+        ensure_can_write_manifest(&manifest).unwrap();
+        assert_ne!(
+            manifest.reader_feature_flags & !without_support,
+            0,
+            "the same manifest is refused without the implementation"
+        );
     }
 
     #[test]
@@ -378,12 +390,12 @@ mod tests {
     fn test_data_overlay_flag_release_gating() {
         // Release default (overlays disabled): the overlay flag is treated as
         // unknown so the dataset is refused, while other known flags still pass.
-        let supported = supported_flags_when(false);
+        let supported = supported_flags_when(false, true);
         assert_eq!(supported & FLAG_UNSTABLE_DATA_OVERLAY_FILES, 0);
         assert_eq!(FLAG_DELETION_FILES & !supported, 0);
         assert_ne!(FLAG_UNSTABLE_DATA_OVERLAY_FILES & !supported, 0);
         // Enabled (debug or env opt-in): the overlay flag is understood.
-        let supported = supported_flags_when(true);
+        let supported = supported_flags_when(true, true);
         assert_eq!(FLAG_UNSTABLE_DATA_OVERLAY_FILES & !supported, 0);
     }
 

rust/lance-table/src/format.rs

@@ -16,7 +16,10 @@ pub use crate::rowids::version::{
     RowDatasetVersionMeta, RowDatasetVersionRun, RowDatasetVersionSequence,
 };
 pub use fragment::*;
-pub use index::{IndexFile, IndexMetadata, index_metadata_codec, list_index_files_with_sizes};
+pub use index::{
+    IdentifierDomain, IndexFile, IndexMetadata, MAX_JSON_INDEX_NESTING, index_metadata_codec,
+    list_index_files_with_sizes,
+};
 
 pub use manifest::{
     BasePath, DETACHED_VERSION_MASK, DataStorageFormat, Manifest, ManifestBuildConfig,