refactor: streaming load no longer support StreamingWithFormat.
@@ -123,57 +123,6 @@ pub async fn streaming_load(
let schema = plan.schema();
match &mut plan {
Plan::Insert(insert) => match &mut insert.source {
- InsertInputSource::StreamingWithFormat(format, start, input_context_ref) => {
- let sql_rest = &insert_sql[*start..].trim();
- if !sql_rest.is_empty() {
- return Err(poem::Error::from_string(
- "should NOT have data after `Format` in streaming load.",
- StatusCode::BAD_REQUEST,
- ));
- };
- let to_table = context
- .get_table(&insert.catalog, &insert.database, &insert.table)
- .await
- .map_err(InternalServerError)?;
- let (tx, rx) = tokio::sync::mpsc::channel(2);
-
- let input_context = Arc::new(
- InputContext::try_create_from_insert_clickhouse(
- format.as_str(),
- rx,
- context.get_settings(),
- schema,
- context.get_scan_progress(),
- true,
- to_table.get_block_compact_thresholds(),
- )
- .await
- .map_err(InternalServerError)?,
- );
- *input_context_ref = Some(input_context.clone());
- tracing::info!("streaming load {:?}", input_context);
-
- let handler = context.spawn(execute_query(context.clone(), plan));
- let files = read_multi_part(multipart, tx, &input_context).await?;
-
- match handler.await {
- Ok(Ok(_)) => Ok(Json(LoadResponse {
- error: None,
- state: "SUCCESS".to_string(),
- id: uuid::Uuid::new_v4().to_string(),
- stats: context.get_scan_progress_value(),
- files,
- })),
- Ok(Err(cause)) => Err(poem::Error::from_string(
- format!("execute fail: {}", cause.message()),
- StatusCode::BAD_REQUEST,
- )),
- Err(_) => Err(poem::Error::from_string(
- "Maybe panic.",
- StatusCode::INTERNAL_SERVER_ERROR,
- )),
- }
- }
InsertInputSource::StreamingWithFileFormat(
option_settings,
start,
