Method create

datafusion/catalog/src/stream.rs:50–81

(
        &self,
        state: &dyn Session,
        cmd: &CreateExternalTable,
    )

#[async_trait]
impl TableProviderFactory for StreamTableFactory {
    async fn create(
        &self,
        state: &dyn Session,
        cmd: &CreateExternalTable,
    ) -> Result<Arc<dyn TableProvider>> {
        let schema: SchemaRef = Arc::clone(cmd.schema.inner());
        let location = cmd.location.clone();
        let encoding = cmd.file_type.parse()?;
        let header = if let Ok(opt) = cmd
            .options
            .get("format.has_header")
            .map(|has_header| bool::from_str(has_header.to_lowercase().as_str()))
            .transpose()
        {
            opt.unwrap_or(false)
        } else {
            return config_err!(
                "Valid values for format.has_header option are 'true' or 'false'"
            );
        };

        let source = FileStreamProvider::new_file(schema, location.into())
            .with_encoding(encoding)
            .with_batch_size(state.config().batch_size())
            .with_header(header);

        let config = StreamConfig::new(Arc::new(source))
            .with_order(cmd.order_exprs.clone())
            .with_constraints(cmd.constraints.clone());

        Ok(Arc::new(StreamTable(Arc::new(config))))
    }
}

/// The data encoding for [`StreamTable`]