fix(csv): unwrap temporal force-text values

crates/dbx-core/src/table_import.rs

@@ -4016,10 +4016,12 @@ fn has_numeric_leading_zero(value: &str) -> bool {
 }
 
 fn is_likely_date(value: &str) -> bool {
+    let value = crate::temporal_format::strip_csv_force_text_wrapper(value);
     ["%Y-%m-%d", "%Y/%m/%d"].iter().any(|format| NaiveDate::parse_from_str(value, format).is_ok())
 }
 
 fn is_likely_timestamp(value: &str) -> bool {
+    let value = crate::temporal_format::strip_csv_force_text_wrapper(value);
     if DateTime::parse_from_rfc3339(value).is_ok() {
         return true;
     }
@@ -9481,6 +9483,10 @@ mod tests {
         assert_eq!(xlsx_cell_value(&duration_cell), serde_json::json!("60:00:00"));
         assert_eq!(infer_value_type(&date_value), Some(ImportInferredType::Timestamp));
         assert_eq!(infer_value_type(&time_value), Some(ImportInferredType::Decimal));
+        assert_eq!(
+            infer_value_type(&serde_json::json!("=\"2026-06-24 02:00:07\"")),
+            Some(ImportInferredType::Timestamp)
+        );
     }
 
     #[test]

crates/dbx-core/src/temporal_format.rs

@@ -148,6 +148,7 @@ fn parse_epoch_temporal(value: &str) -> Option<NaiveDateTime> {
 }
 
 fn parse_temporal(value: &str, preferred_pattern: Option<&str>) -> Option<ParsedTemporal> {
+    let value = strip_csv_force_text_wrapper(value);
     preferred_pattern
         .filter(|pattern| !pattern.trim().is_empty())
         .and_then(|pattern| parse_with_pattern(value, pattern))
@@ -224,6 +225,10 @@ fn active_temporal_pattern(pattern: Option<&str>) -> Option<&str> {
     pattern.filter(|pattern| !pattern.trim().is_empty())
 }
 
+pub(crate) fn strip_csv_force_text_wrapper(value: &str) -> &str {
+    value.strip_prefix("=\"").and_then(|value| value.strip_suffix('"')).unwrap_or(value)
+}
+
 pub fn format_temporal_export_row_cow<'a>(
     row: &'a [Value],
     column_types: &[Option<String>],
@@ -633,6 +638,15 @@ mod tests {
         );
     }
 
+    #[test]
+    fn import_normalization_unwraps_csv_force_text_temporal_values() {
+        assert_eq!(
+            normalize_temporal_import_value(&json!("=\"2026-06-24 02:00:07\""), Some("DATETIME"), None),
+            json!("2026-06-24 02:00:07")
+        );
+        assert_eq!(normalize_temporal_import_value(&json!("=\"2026-06-24\""), Some("DATE"), None), json!("2026-06-24"));
+    }
+
     #[test]
     fn preserves_raw_export_values_when_pattern_requires_missing_fields() {
         assert_eq!(

crates/dbx-core/tests/live_mysql57.rs

@@ -14,7 +14,9 @@ use dbx_core::query_result_export::{export_query_result_core, ExportStatus, Quer
 use dbx_core::sql::{split_sql_statements_for_database, SqlFileRequest};
 use dbx_core::sql_file_import::execute_sql_file_path;
 use dbx_core::storage::Storage;
-use dbx_core::table_import::parse_xlsx_file;
+use dbx_core::table_import::{
+    build_import_insert_batch_from_rows, parse_csv_bytes, parse_xlsx_file, TableImportColumnMapping,
+};
 use mysql_async::prelude::Queryable;
 use tokio_util::sync::CancellationToken;
 
@@ -511,6 +513,118 @@ async fn live_mysql_query_result_export_xlsx_streams_single_query_without_duplic
     assert_eq!(exported_ids, expected_ids);
 }
 
+#[tokio::test]
+#[ignore = "requires a writable MySQL endpoint for issue #8803 regression coverage"]
+async fn live_mysql_csv_temporal_export_round_trip_preserves_dbx_force_text_values() {
+    let host = std::env::var("DBX_LIVE_MYSQL_EXPORT_HOST").expect("DBX_LIVE_MYSQL_EXPORT_HOST");
+    let port = std::env::var("DBX_LIVE_MYSQL_EXPORT_PORT").expect("DBX_LIVE_MYSQL_EXPORT_PORT").parse::<u16>().unwrap();
+    let user = std::env::var("DBX_LIVE_MYSQL_EXPORT_USER").expect("DBX_LIVE_MYSQL_EXPORT_USER");
+    let password = std::env::var("DBX_LIVE_MYSQL_EXPORT_PASSWORD").expect("DBX_LIVE_MYSQL_EXPORT_PASSWORD");
+    let database = std::env::var("DBX_LIVE_MYSQL_EXPORT_DATABASE").expect("DBX_LIVE_MYSQL_EXPORT_DATABASE");
+    let suffix = uuid::Uuid::new_v4().simple().to_string();
+    let table = format!("dbx_issue_8803_{}", &suffix[..8]);
+    let target_table = format!("dbx_issue_8803_target_{}", &suffix[..8]);
+    let connection_id = format!("live-mysql-issue-8803-{suffix}");
+    let config = live_mysql_query_export_config(&connection_id, &host, port, &user, &password, &database);
+    let dir = std::env::temp_dir().join(format!("dbx-live-issue-8803-{suffix}"));
+    std::fs::create_dir_all(&dir).unwrap();
+    let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
+    let state = AppState::new(storage);
+    state.configs.write().await.insert(config.id.clone(), config);
+
+    let cleanup_sql = format!("DROP TABLE IF EXISTS `{table}`, `{target_table}`");
+    let create_sql = format!("CREATE TABLE `{table}` (insert_time DATETIME NOT NULL, id INT NOT NULL)");
+    let create_target_sql = format!("CREATE TABLE `{target_table}` (insert_time DATETIME NOT NULL, id INT NOT NULL)");
+    let insert_sql = format!("INSERT INTO `{table}` (insert_time, id) VALUES ('2026-06-24 02:00:07', 695350)");
+    let _ = execute_sql_statement(&state, &connection_id, &database, &cleanup_sql, None, None).await;
+    execute_sql_statement(&state, &connection_id, &database, &create_sql, None, None).await.expect("create fixture");
+    execute_sql_statement(&state, &connection_id, &database, &insert_sql, None, None).await.expect("insert fixture");
+    execute_sql_statement(&state, &connection_id, &database, &create_target_sql, None, None)
+        .await
+        .expect("create import target");
+
+    let file_path = dir.join("issue-8803.csv");
+    let sql = format!("SELECT insert_time, id FROM `{table}`");
+    let request = QueryResultExportRequest {
+        export_id: format!("live-mysql-issue-8803-{suffix}"),
+        connection_id: connection_id.clone(),
+        database: database.clone(),
+        schema: None,
+        catalog: None,
+        sql: sql.clone(),
+        query_base_sql: sql,
+        setup_sql: Vec::new(),
+        database_type: DatabaseType::Mysql,
+        use_agent_cursor: false,
+        file_path: file_path.to_string_lossy().to_string(),
+        format: "csv".to_string(),
+        insert_mode: Default::default(),
+        include_sql_sheet: false,
+        page_size: 100,
+        row_limit: None,
+        total_rows: Some(1),
+        timeout_secs: Some(30),
+        keyset_optimization_enabled: false,
+        client_session_id: None,
+        execution_id: Some(format!("live-mysql-issue-8803-{suffix}")),
+        date_time_format: None,
+        csv_quote_mode: Default::default(),
+        export_table_name: None,
+        export_column_types: None,
+        column_comments: None,
+        auto_filter: None,
+        identifier_quote: None,
+        numeric_column_right_align: false,
+    };
+    export_query_result_core(&state, &request, None, |_| {}).await.expect("export fixture");
+
+    let csv = std::fs::read(&file_path).unwrap();
+    let parsed = parse_csv_bytes(&csv, 10).expect("DBX should parse its own CSV");
+    let mappings = parsed
+        .columns
+        .iter()
+        .map(|column| TableImportColumnMapping {
+            source_column: column.clone(),
+            target_column: column.clone(),
+            target_data_type: None,
+        })
+        .collect::<Vec<_>>();
+    let batch = build_import_insert_batch_from_rows(
+        &parsed.rows,
+        &parsed.columns,
+        &mappings,
+        &[("insert_time".to_string(), "DATETIME".to_string()), ("id".to_string(), "INT".to_string())],
+        &target_table,
+        &database,
+        &DatabaseType::Mysql,
+    )
+    .expect("build import batch")
+    .expect("import batch should exist");
+
+    eprintln!("issue #8803 raw CSV: {}", String::from_utf8_lossy(&csv));
+    eprintln!("issue #8803 import SQL: {}", batch.sql);
+    assert!(String::from_utf8_lossy(&csv).contains("=\"\"2026-06-24 02:00:07\"\""));
+    assert!(batch.sql.contains("'2026-06-24 02:00:07'"));
+    execute_sql_statement(&state, &connection_id, &database, &batch.sql, None, None)
+        .await
+        .expect("import exported temporal CSV");
+    let imported = execute_sql_statement(
+        &state,
+        &connection_id,
+        &database,
+        &format!("SELECT DATE_FORMAT(insert_time, '%Y-%m-%d %H:%i:%s'), id FROM `{target_table}`"),
+        None,
+        None,
+    )
+    .await
+    .expect("query imported temporal row");
+    assert_eq!(imported.rows, vec![vec![serde_json::json!("2026-06-24 02:00:07"), serde_json::json!("695350")]]);
+
+    let cleanup_result = execute_sql_statement(&state, &connection_id, &database, &cleanup_sql, None, None).await;
+    cleanup_result.expect("cleanup fixture");
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
 #[tokio::test]
 #[ignore = "requires a writable MySQL endpoint for a 650,000-row XLSX export"]
 async fn live_mysql_xlsx_export_can_outlive_query_timeout_while_rows_keep_arriving() {