(
&self,
table_name: ObjectName,
columns: Vec<ObjectName>,
source: Box<Query>,
overwrite: bool,
replace_into: bool,
)
| 2356 | } |
| 2357 | |
| 2358 | fn insert_to_plan( |
| 2359 | &self, |
| 2360 | table_name: ObjectName, |
| 2361 | columns: Vec<ObjectName>, |
| 2362 | source: Box<Query>, |
| 2363 | overwrite: bool, |
| 2364 | replace_into: bool, |
| 2365 | ) -> Result<LogicalPlan> { |
| 2366 | // Do a table lookup to verify the table exists |
| 2367 | let table_name = self.object_name_to_table_reference(table_name)?; |
| 2368 | let table_source = self.context_provider.get_table_source(table_name.clone())?; |
| 2369 | let table_schema = DFSchema::try_from(table_source.schema())?; |
| 2370 | |
| 2371 | let columns: Vec<Ident> = columns |
| 2372 | .into_iter() |
| 2373 | .map(|name| { |
| 2374 | if name.0.len() != 1 { |
| 2375 | return not_impl_err!( |
| 2376 | "Multi-part column names in INSERT not supported: {name}" |
| 2377 | ); |
| 2378 | } |
| 2379 | let part = &name.0[0]; |
| 2380 | let Some(ident) = part.as_ident() else { |
| 2381 | return not_impl_err!( |
| 2382 | "Non-identifier column name part in INSERT not supported: {part}" |
| 2383 | ); |
| 2384 | }; |
| 2385 | Ok(ident.clone()) |
| 2386 | }) |
| 2387 | .collect::<Result<Vec<_>>>()?; |
| 2388 | |
| 2389 | // Get insert fields and target table's value indices |
| 2390 | // |
| 2391 | // If value_indices[i] = Some(j), it means that the value of the i-th target table's column is |
| 2392 | // derived from the j-th output of the source. |
| 2393 | // |
| 2394 | // If value_indices[i] = None, it means that the value of the i-th target table's column is |
| 2395 | // not provided, and should be filled with a default value later. |
| 2396 | let (fields, value_indices) = if columns.is_empty() { |
| 2397 | // Empty means we're inserting into all columns of the table |
| 2398 | ( |
| 2399 | table_schema.fields().clone(), |
| 2400 | (0..table_schema.fields().len()) |
| 2401 | .map(Some) |
| 2402 | .collect::<Vec<_>>(), |
| 2403 | ) |
| 2404 | } else { |
| 2405 | let mut value_indices = vec![None; table_schema.fields().len()]; |
| 2406 | let fields = columns |
| 2407 | .into_iter() |
| 2408 | .enumerate() |
| 2409 | .map(|(i, c)| { |
| 2410 | let c = self.ident_normalizer.normalize(c); |
| 2411 | let column_index = table_schema |
| 2412 | .index_of_column_by_name(None, &c) |
| 2413 | .ok_or_else(|| unqualified_field_not_found(&c, &table_schema))?; |
| 2414 | |
| 2415 | if value_indices[column_index].is_some() { |
no test coverage detected