fix(table): resolve index files by external path and bucket layout

bindings/c/src/tests.rs

@@ -2461,6 +2461,7 @@ fn build_pk_vector_table(path: &str, vectors: &[[f32; PK_DIM]]) -> Table {
             file_size: i64::try_from(index_file_size).unwrap(),
             row_count,
             deletion_vectors_ranges: None,
+            external_path: None,
             global_index_meta: Some(GlobalIndexMeta {
                 row_range_start: 0,
                 row_range_end: row_count - 1,

crates/integrations/datafusion/src/system_tables/table_indexes.rs

@@ -289,6 +289,7 @@ mod tests {
                 file_size: 1,
                 row_count: 1,
                 deletion_vectors_ranges: None,
+                external_path: None,
                 global_index_meta: None,
             },
             version: 1,

crates/integrations/datafusion/tests/read_tables.rs

@@ -1568,6 +1568,7 @@ mod fulltext_tests {
             file_size: i64::try_from(index_bytes.len()).unwrap(),
             row_count: 5,
             deletion_vectors_ranges: None,
+            external_path: None,
             global_index_meta: Some(GlobalIndexMeta {
                 row_range_start: 0,
                 row_range_end: 4,

crates/paimon/src/catalog/filesystem.rs

@@ -26,7 +26,10 @@ use crate::common::{CatalogOptions, Options};
 use crate::error::{ConfigInvalidSnafu, Error, Result};
 use crate::io::cache::{create_local_cache, LocalCache};
 use crate::io::FileIO;
-use crate::spec::{CoreOptions, Schema, TableSchema, TableType, TABLE_TYPE_OPTION};
+use crate::spec::{
+    CoreOptions, Schema, TableSchema, TableType, INDEX_FILE_IN_DATA_FILE_DIR_OPTION,
+    TABLE_TYPE_OPTION,
+};
 use crate::table::{ObjectTable, SchemaManager, Table};
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -539,7 +542,7 @@ impl Catalog for FileSystemCatalog {
                 full_name: identifier.full_name(),
             })?;
 
-        reject_table_type_changes(current.options(), &changes)?;
+        reject_immutable_option_changes(current.options(), &changes)?;
 
         let new_schema = current
             .apply_changes(changes)
@@ -548,10 +551,19 @@ impl Catalog for FileSystemCatalog {
     }
 }
 
-/// The declared type picks the reader, so it is fixed at creation: flipping
-/// it strands a populated table behind a reader that cannot see its data.
-/// Only case-insensitive no-ops pass, matching Java `SchemaManager`.
-fn reject_table_type_changes(
+/// Options whose value is baked into the on-disk layout, so changing one on a
+/// populated table strands the files already written under the old value.
+///
+/// Java rejects any alteration of an option annotated `@Immutable`
+/// (`SchemaManager.checkAlterTableOption` against `CoreOptions.IMMUTABLE_OPTIONS`);
+/// this mirrors the subset this crate acts on:
+///
+/// * `type` picks the reader, so flipping it strands a populated table behind a
+///   reader that cannot see its data. Only case-insensitive no-ops pass.
+/// * `index-file-in-data-file-dir` picks the directory every bucket-local index
+///   file is written to and read from, so flipping it hides every index file the
+///   table already has.
+fn reject_immutable_option_changes(
     current_options: &HashMap<String, String>,
     changes: &[crate::spec::SchemaChange],
 ) -> Result<()> {
@@ -575,6 +587,18 @@ fn reject_table_type_changes(
                     message: format!("removing '{TABLE_TYPE_OPTION}' is not supported"),
                 });
             }
+            crate::spec::SchemaChange::SetOption { key, .. }
+            | crate::spec::SchemaChange::RemoveOption { key }
+                if key == INDEX_FILE_IN_DATA_FILE_DIR_OPTION =>
+            {
+                return Err(Error::Unsupported {
+                    message: format!(
+                        "changing '{INDEX_FILE_IN_DATA_FILE_DIR_OPTION}' is not supported: \
+                         it selects the directory index files are written to, so the files \
+                         already written would no longer be found"
+                    ),
+                });
+            }
             _ => {}
         }
     }
@@ -828,6 +852,51 @@ mod tests {
         catalog.get_table(&identifier).await.unwrap();
     }
 
+    #[tokio::test]
+    async fn test_alter_table_cannot_change_where_index_files_live() {
+        use crate::spec::SchemaChange;
+
+        // The option selects the directory every bucket-local index file is written
+        // to and read from. Flipping it on a populated table would hide every index
+        // file already written, so it is fixed at creation, as in Java where it is
+        // annotated `@Immutable`.
+        let (_temp_dir, catalog) = create_test_catalog();
+        catalog
+            .create_database("db1", false, HashMap::new())
+            .await
+            .unwrap();
+        let schema = Schema::builder()
+            .column(
+                "id",
+                crate::spec::DataType::Int(crate::spec::IntType::new()),
+            )
+            .build()
+            .unwrap();
+        let identifier = Identifier::new("db1", "t");
+        catalog
+            .create_table(&identifier, schema, false)
+            .await
+            .unwrap();
+
+        for change in [
+            SchemaChange::set_option(
+                "index-file-in-data-file-dir".to_string(),
+                "true".to_string(),
+            ),
+            SchemaChange::set_option(
+                "index-file-in-data-file-dir".to_string(),
+                "false".to_string(),
+            ),
+            SchemaChange::remove_option("index-file-in-data-file-dir".to_string()),
+        ] {
+            let err = catalog
+                .alter_table(&identifier, vec![change], false)
+                .await
+                .unwrap_err();
+            assert!(matches!(err, Error::Unsupported { .. }), "{err:?}");
+        }
+    }
+
     #[tokio::test]
     async fn test_create_table_rejects_an_unknown_type() {
         let (_temp_dir, catalog) = create_test_catalog();

crates/paimon/src/spec/avro/decode_helpers.rs

@@ -58,13 +58,23 @@ pub(crate) fn read_bytes_field(cursor: &mut AvroCursor, nullable: bool) -> crate
 }
 
 pub(crate) fn read_string_field(cursor: &mut AvroCursor, nullable: bool) -> crate::Result<String> {
+    Ok(read_nullable_string_field(cursor, nullable)?.unwrap_or_default())
+}
+
+/// Reads a nullable string field, preserving the null/present distinction.
+/// Returns `None` for the null branch of a `["null", "string"]` union (a
+/// non-nullable field is always `Some`).
+pub(crate) fn read_nullable_string_field(
+    cursor: &mut AvroCursor,
+    nullable: bool,
+) -> crate::Result<Option<String>> {
     if nullable {
         let idx = cursor.read_union_index()?;
         if idx == 0 {
-            return Ok(String::new());
+            return Ok(None);
         }
     }
-    Ok(cursor.read_string()?.to_string())
+    Ok(Some(cursor.read_string()?.to_string()))
 }
 
 const EMPTY_PARTITION: [u8; 4] = [0, 0, 0, 0];

crates/paimon/src/spec/avro/index_manifest_entry_decode.rs

@@ -19,7 +19,7 @@ use super::cursor::AvroCursor;
 use super::decode::{neg_count_to_usize, AvroRecordDecode};
 use super::decode_helpers::{
     extract_record_schema, normalize_partition, read_bytes_field, read_int_field, read_long_field,
-    read_string_field,
+    read_nullable_string_field, read_string_field,
 };
 use super::schema::{skip_nullable_field, WriterSchema};
 use crate::spec::index_manifest::IndexManifestEntry;
@@ -38,6 +38,7 @@ impl AvroRecordDecode for IndexManifestEntry {
         let mut file_size: Option<i64> = None;
         let mut row_count: Option<i64> = None;
         let mut deletion_vectors_ranges: Option<IndexMap<String, DeletionVectorMeta>> = None;
+        let mut external_path: Option<String> = None;
         let mut global_index_meta: Option<GlobalIndexMeta> = None;
 
         for field in &writer_schema.fields {
@@ -65,6 +66,9 @@ impl AvroRecordDecode for IndexManifestEntry {
                 "_DELETIONS_VECTORS_RANGES" | "_DELETION_VECTORS_RANGES" => {
                     deletion_vectors_ranges = decode_nullable_dv_ranges(cursor, field.nullable)?;
                 }
+                "_EXTERNAL_PATH" => {
+                    external_path = read_nullable_string_field(cursor, field.nullable)?;
+                }
                 "_GLOBAL_INDEX" => {
                     global_index_meta =
                         decode_nullable_global_index(cursor, field.nullable, &field.schema)?;
@@ -84,6 +88,7 @@ impl AvroRecordDecode for IndexManifestEntry {
                 file_size: file_size.unwrap_or(0),
                 row_count: row_count.unwrap_or(0),
                 deletion_vectors_ranges,
+                external_path,
                 global_index_meta,
             },
         })

crates/paimon/src/spec/core_options.rs

@@ -32,6 +32,7 @@ const GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION: &str = "global-index.row-count-pe
 const GLOBAL_INDEX_THREAD_NUM_OPTION: &str = "global-index.thread-num";
 const GLOBAL_INDEX_VINDEX_READ_THREAD_NUM_OPTION: &str = "global-index.vindex.read-thread-num";
 const GLOBAL_INDEX_COLUMN_UPDATE_ACTION_OPTION: &str = "global-index.column-update-action";
+pub(crate) const INDEX_FILE_IN_DATA_FILE_DIR_OPTION: &str = "index-file-in-data-file-dir";
 const SORTED_INDEX_RECORDS_PER_RANGE_OPTION: &str = "sorted-index.records-per-range";
 const BTREE_INDEX_RECORDS_PER_RANGE_OPTION: &str = "btree-index.records-per-range";
 const BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION: &str = "btree-index.fallback-scan-max-size";
@@ -686,6 +687,16 @@ impl<'a> CoreOptions<'a> {
             .unwrap_or(true)
     }
 
+    /// Whether index files are stored in the bucket data-file directory rather
+    /// than the table `index/` directory (option `index-file-in-data-file-dir`,
+    /// default false).
+    pub fn index_file_in_data_file_dir(&self) -> bool {
+        self.options
+            .get(INDEX_FILE_IN_DATA_FILE_DIR_OPTION)
+            .map(|value| value.eq_ignore_ascii_case("true"))
+            .unwrap_or(false)
+    }
+
     pub fn global_index_search_mode(&self) -> crate::Result<GlobalIndexSearchMode> {
         self.index_search_mode(GLOBAL_INDEX_SEARCH_MODE_OPTION)
     }

crates/paimon/src/spec/index_file_meta.rs

@@ -77,6 +77,15 @@ pub struct IndexFileMeta {
     )]
     pub deletion_vectors_ranges: Option<IndexMap<String, DeletionVectorMeta>>,
 
+    /// Absolute path of an externally-stored index file. `None` when the file
+    /// lives under the table's index directory (or bucket data-file directory).
+    #[serde(
+        default,
+        rename = "_EXTERNAL_PATH",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub external_path: Option<String>,
+
     #[serde(
         default,
         rename = "_GLOBAL_INDEX",