Load a schema by ID. Returns cached version if available. The cache is shared across all clones of this `SchemaManager`, so loading a schema in one stream makes it available to all other streams reading from the same table. Reference: [SchemaManager.schema(long)](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java)
(&self, schema_id: i64)
| 127 | /// |
| 128 | /// Reference: [SchemaManager.schema(long)](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java) |
| 129 | pub async fn schema(&self, schema_id: i64) -> crate::Result<Arc<TableSchema>> { |
| 130 | // Fast path: check cache under a short lock. |
| 131 | { |
| 132 | let cache = self.cache.lock().unwrap(); |
| 133 | if let Some(schema) = cache.get(&schema_id) { |
| 134 | return Ok(schema.clone()); |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | // Cache miss — load from file (no lock held during I/O). |
| 139 | let path = self.schema_path(schema_id); |
| 140 | let input = self.file_io.new_input(&path)?; |
| 141 | let bytes = input.read().await?; |
| 142 | let schema: TableSchema = |
| 143 | serde_json::from_slice(&bytes).map_err(|e| crate::Error::DataInvalid { |
| 144 | message: format!("Failed to parse schema file: {path}"), |
| 145 | source: Some(Box::new(e)), |
| 146 | })?; |
| 147 | let schema = Arc::new(schema); |
| 148 | |
| 149 | // Insert into shared cache (short lock). |
| 150 | { |
| 151 | let mut cache = self.cache.lock().unwrap(); |
| 152 | cache.entry(schema_id).or_insert_with(|| schema.clone()); |
| 153 | } |
| 154 | |
| 155 | Ok(schema) |
| 156 | } |
| 157 | } |
| 158 | |
| 159 | #[cfg(test)] |