(&self, statement: CopyToStatement)
| 1664 | } |
| 1665 | |
| 1666 | fn copy_to_plan(&self, statement: CopyToStatement) -> Result<LogicalPlan> { |
| 1667 | // Determine if source is table or query and handle accordingly |
| 1668 | let copy_source = statement.source; |
| 1669 | let (input, input_schema, table_ref) = match copy_source { |
| 1670 | CopyToSource::Relation(object_name) => { |
| 1671 | let table_name = object_name_to_string(&object_name); |
| 1672 | let table_ref = self.object_name_to_table_reference(object_name)?; |
| 1673 | let table_source = |
| 1674 | self.context_provider.get_table_source(table_ref.clone())?; |
| 1675 | let plan = |
| 1676 | LogicalPlanBuilder::scan(table_name, table_source, None)?.build()?; |
| 1677 | let input_schema = Arc::clone(plan.schema()); |
| 1678 | (plan, input_schema, Some(table_ref)) |
| 1679 | } |
| 1680 | CopyToSource::Query(query) => { |
| 1681 | let plan = self.query_to_plan(*query, &mut PlannerContext::new())?; |
| 1682 | let input_schema = Arc::clone(plan.schema()); |
| 1683 | (plan, input_schema, None) |
| 1684 | } |
| 1685 | }; |
| 1686 | |
| 1687 | let options_map = self.parse_options_map(statement.options, true)?; |
| 1688 | |
| 1689 | let maybe_file_type = if let Some(stored_as) = &statement.stored_as { |
| 1690 | self.context_provider.get_file_type(stored_as).ok() |
| 1691 | } else { |
| 1692 | None |
| 1693 | }; |
| 1694 | |
| 1695 | let file_type = match maybe_file_type { |
| 1696 | Some(ft) => ft, |
| 1697 | None => { |
| 1698 | let e = || { |
| 1699 | DataFusionError::Configuration( |
| 1700 | "Format not explicitly set and unable to get file extension! Use STORED AS to define file format." |
| 1701 | .to_string(), |
| 1702 | ) |
| 1703 | }; |
| 1704 | // Try to infer file format from file extension |
| 1705 | let extension: &str = &Path::new(&statement.target) |
| 1706 | .extension() |
| 1707 | .ok_or_else(e)? |
| 1708 | .to_str() |
| 1709 | .ok_or_else(e)? |
| 1710 | .to_lowercase(); |
| 1711 | |
| 1712 | self.context_provider.get_file_type(extension)? |
| 1713 | } |
| 1714 | }; |
| 1715 | |
| 1716 | let partition_by = statement |
| 1717 | .partitioned_by |
| 1718 | .iter() |
| 1719 | .map(|col| input_schema.field_with_name(table_ref.as_ref(), col)) |
| 1720 | .collect::<Result<Vec<_>>>()? |
| 1721 | .into_iter() |
| 1722 | .map(|f| f.name().to_owned()) |
| 1723 | .collect(); |
no test coverage detected