(db: &Connection, name: &str, schema: Arc<Schema>)
| 34 | } |
| 35 | |
| 36 | pub async fn ensure_table(db: &Connection, name: &str, schema: Arc<Schema>) -> Result<Table> { |
| 37 | let table = match db.open_table(name).execute().await { |
| 38 | Ok(table) => table, |
| 39 | Err(_) => { |
| 40 | let batch = RecordBatch::new_empty(schema.clone()); |
| 41 | let batches = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema.clone()); |
| 42 | let mut builder = db |
| 43 | .create_table(name, Box::new(batches) as Box<dyn RecordBatchReader + Send>) |
| 44 | .storage_option("new_table_enable_stable_row_ids", "true") |
| 45 | .storage_option("new_table_enable_v2_manifest_paths", "true"); |
| 46 | if schema_has_blob_v2_field(schema.as_ref()) { |
| 47 | builder = builder.storage_option("new_table_data_storage_version", "2.2"); |
| 48 | } |
| 49 | builder.execute().await?; |
| 50 | db.open_table(name).execute().await? |
| 51 | }, |
| 52 | }; |
| 53 | |
| 54 | ensure_table_columns(&table, schema.as_ref()).await?; |
| 55 | Ok(table) |
| 56 | } |
| 57 | |
| 58 | async fn ensure_table_columns(table: &Table, expected_schema: &Schema) -> Result<()> { |
| 59 | let existing_schema = table.schema().await?; |
no test coverage detected