MCPcopy Create free account
hub / github.com/apache/datafusion / insert_to_plan

Method insert_to_plan

datafusion/sql/src/statement.rs:2358–2512  ·  view source on GitHub ↗
(
        &self,
        table_name: ObjectName,
        columns: Vec<ObjectName>,
        source: Box<Query>,
        overwrite: bool,
        replace_into: bool,
    )

Source from the content-addressed store, hash-verified

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() {

Calls 15

newFunction · 0.85
projectFunction · 0.85
as_identMethod · 0.80
lastMethod · 0.80
collectMethod · 0.80
set_table_schemaMethod · 0.80
query_to_planMethod · 0.80
qualified_fieldMethod · 0.80
ColumnClass · 0.50

Tested by

no test coverage detected