feat(settings): add global date and time formats
This commit is contained in:
31 files changed
+1109
-88
No files matched your search
@@ -256,7 +256,7 @@ pub struct SqlServerStreamExportSummary {
|
||||
}
|
||||
|
||||
pub enum SqlServerStreamItem<'a> {
|
||||
Columns(&'a [String]),
|
||||
Columns { columns: &'a [String], column_types: &'a [String] },
|
||||
Row(&'a [serde_json::Value]),
|
||||
}
|
||||
|
||||
@@ -638,6 +638,7 @@ pub async fn stream_first_result_set(
|
||||
let mut stream = sqlserver_driver_result(client.query(query_sql.as_str(), &[])).await?;
|
||||
let mut active_result_index: Option<usize> = None;
|
||||
let mut columns: Vec<String> = Vec::new();
|
||||
let mut column_types: Vec<String> = Vec::new();
|
||||
let mut columns_emitted = false;
|
||||
let mut rows_exported = 0_u64;
|
||||
|
||||
@@ -663,7 +664,8 @@ pub async fn stream_first_result_set(
|
||||
if active_result_index.is_none() {
|
||||
active_result_index = Some(metadata.result_index());
|
||||
columns = columns_from_metadata(&metadata);
|
||||
on_item(SqlServerStreamItem::Columns(&columns))?;
|
||||
column_types = column_types_from_metadata(&metadata);
|
||||
on_item(SqlServerStreamItem::Columns { columns: &columns, column_types: &column_types })?;
|
||||
columns_emitted = true;
|
||||
}
|
||||
}
|
||||
@@ -671,7 +673,8 @@ pub async fn stream_first_result_set(
|
||||
if active_result_index.is_none() {
|
||||
active_result_index = Some(row.result_index());
|
||||
columns = row.columns().iter().map(|c| c.name().to_string()).collect();
|
||||
on_item(SqlServerStreamItem::Columns(&columns))?;
|
||||
column_types = row.columns().iter().map(sqlserver_column_type_name).collect();
|
||||
on_item(SqlServerStreamItem::Columns { columns: &columns, column_types: &column_types })?;
|
||||
columns_emitted = true;
|
||||
}
|
||||
if Some(row.result_index()) != active_result_index {
|
||||
@@ -688,7 +691,7 @@ pub async fn stream_first_result_set(
|
||||
}
|
||||
|
||||
if !columns_emitted {
|
||||
on_item(SqlServerStreamItem::Columns(&columns))?;
|
||||
on_item(SqlServerStreamItem::Columns { columns: &columns, column_types: &column_types })?;
|
||||
}
|
||||
Ok(SqlServerStreamExportSummary { columns, rows_exported })
|
||||
}
|
||||
|
||||
@@ -61,6 +61,7 @@ pub mod table_export;
|
||||
pub mod table_import;
|
||||
pub mod table_structure_sql;
|
||||
pub mod task_supervisor;
|
||||
pub mod temporal_format;
|
||||
pub mod text_export;
|
||||
pub mod token_usage;
|
||||
pub mod transfer;
|
||||
|
||||
@@ -81,6 +81,8 @@ pub struct QueryResultExportRequest {
|
||||
pub client_session_id: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub execution_id: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub date_time_format: Option<String>,
|
||||
}
|
||||
|
||||
fn split_excel_cell_text(value: &str) -> Vec<String> {
|
||||
@@ -563,6 +565,11 @@ async fn export_query_result_core_inner(
|
||||
result.rows.truncate(this_page);
|
||||
}
|
||||
let row_count = result.rows.len();
|
||||
let formatted_rows = crate::temporal_format::format_temporal_export_rows_with_string_types(
|
||||
&result.rows,
|
||||
&column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
|
||||
if format == "csv" || format == "txt" {
|
||||
if let Some(file) = text_file.as_mut() {
|
||||
@@ -570,12 +577,12 @@ async fn export_query_result_core_inner(
|
||||
let header = format_text_export_header(&format, &columns);
|
||||
file.write_all(header.as_bytes()).map_err(|e| format!("Failed to write export header: {e}"))?;
|
||||
if row_count > 0 {
|
||||
let rows = format_text_export_rows(&format, &result.rows);
|
||||
let rows = format_text_export_rows(&format, &formatted_rows);
|
||||
write!(file, "\n{rows}").map_err(|e| format!("Failed to write export rows: {e}"))?;
|
||||
}
|
||||
wrote_text_header = true;
|
||||
} else if row_count > 0 {
|
||||
let rows = format_text_export_rows(&format, &result.rows);
|
||||
let rows = format_text_export_rows(&format, &formatted_rows);
|
||||
write!(file, "\n{rows}").map_err(|e| format!("Failed to write export rows: {e}"))?;
|
||||
}
|
||||
}
|
||||
@@ -591,7 +598,7 @@ async fn export_query_result_core_inner(
|
||||
)?);
|
||||
}
|
||||
if let Some(writer) = xlsx.as_mut() {
|
||||
for row in &result.rows {
|
||||
for row in &formatted_rows {
|
||||
writer.write_row(row).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
}
|
||||
}
|
||||
@@ -708,6 +715,7 @@ async fn try_export_postgres_query_result_stream(
|
||||
if xlsx_hard_limit_active { row_limit.map(|limit| limit.saturating_add(1)) } else { row_limit };
|
||||
let progress_row_interval = request.page_size.max(1) as u64;
|
||||
let mut columns: Vec<String> = Vec::new();
|
||||
let mut temporal_column_types: Vec<String> = Vec::new();
|
||||
let mut rows_exported = 0_u64;
|
||||
let mut last_progress_rows = 0_u64;
|
||||
let mut last_progress_at = Instant::now();
|
||||
@@ -735,6 +743,7 @@ async fn try_export_postgres_query_result_stream(
|
||||
match item {
|
||||
crate::db::postgres::PostgresQueryStreamItem::Columns { columns: stream_columns, column_types } => {
|
||||
columns = stream_columns;
|
||||
temporal_column_types = column_types.clone();
|
||||
if let Some(file) = text_file.as_mut() {
|
||||
let header = format_text_export_header(format, &columns);
|
||||
file.write_all(header.as_bytes()).map_err(|e| format!("Failed to write export header: {e}"))?;
|
||||
@@ -753,18 +762,23 @@ async fn try_export_postgres_query_result_stream(
|
||||
if xlsx_hard_limit_active && rows_exported as usize >= XLSX_MAX_DATA_ROWS {
|
||||
return Err(XLSX_ROW_LIMIT_ERROR.to_string());
|
||||
}
|
||||
let formatted = crate::temporal_format::format_temporal_export_row_with_string_types(
|
||||
&row,
|
||||
&temporal_column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
if let Some(file) = text_file.as_mut() {
|
||||
let rows = format_text_export_rows(format, std::slice::from_ref(&row));
|
||||
let rows = format_text_export_rows(format, std::slice::from_ref(&formatted));
|
||||
write!(file, "\n{rows}").map_err(|e| format!("Failed to write export rows: {e}"))?;
|
||||
} else if let Some(writer) = xlsx.as_mut() {
|
||||
writer.write_row(&row).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
writer.write_row(&formatted).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
} else {
|
||||
let xlsx_file =
|
||||
File::create(&request.file_path).map_err(|e| format!("Failed to create XLSX file: {e}"))?;
|
||||
xlsx =
|
||||
Some(start_query_result_xlsx_workbook(BufWriter::new(xlsx_file), request, &columns, &[])?);
|
||||
if let Some(writer) = xlsx.as_mut() {
|
||||
writer.write_row(&row).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
writer.write_row(&formatted).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
}
|
||||
}
|
||||
rows_exported += 1;
|
||||
@@ -870,6 +884,7 @@ async fn try_export_mysql_query_result_stream(
|
||||
if xlsx_hard_limit_active { row_limit.map(|limit| limit.saturating_add(1)) } else { row_limit };
|
||||
let progress_row_interval = request.page_size.max(1) as u64;
|
||||
let mut columns: Vec<String> = Vec::new();
|
||||
let mut temporal_column_types: Vec<String> = Vec::new();
|
||||
let mut rows_exported = 0_u64;
|
||||
let mut last_progress_rows = 0_u64;
|
||||
let mut last_progress_at = Instant::now();
|
||||
@@ -946,6 +961,7 @@ async fn try_export_mysql_query_result_stream(
|
||||
match item {
|
||||
crate::db::mysql::MySqlQueryStreamItem::Columns { columns: stream_columns, column_types } => {
|
||||
columns = stream_columns;
|
||||
temporal_column_types = column_types.clone();
|
||||
if let Some(file) = text_file.as_mut() {
|
||||
let header = format_text_export_header(format, &columns);
|
||||
file.write_all(header.as_bytes()).map_err(|e| format!("Failed to write export header: {e}"))?;
|
||||
@@ -964,18 +980,23 @@ async fn try_export_mysql_query_result_stream(
|
||||
if xlsx_hard_limit_active && rows_exported as usize >= XLSX_MAX_DATA_ROWS {
|
||||
return Err(XLSX_ROW_LIMIT_ERROR.to_string());
|
||||
}
|
||||
let formatted = crate::temporal_format::format_temporal_export_row_with_string_types(
|
||||
&row,
|
||||
&temporal_column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
if let Some(file) = text_file.as_mut() {
|
||||
let rows = format_text_export_rows(format, std::slice::from_ref(&row));
|
||||
let rows = format_text_export_rows(format, std::slice::from_ref(&formatted));
|
||||
write!(file, "\n{rows}").map_err(|e| format!("Failed to write export rows: {e}"))?;
|
||||
} else if let Some(writer) = xlsx.as_mut() {
|
||||
writer.write_row(&row).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
writer.write_row(&formatted).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
} else {
|
||||
let xlsx_file =
|
||||
File::create(&request.file_path).map_err(|e| format!("Failed to create XLSX file: {e}"))?;
|
||||
xlsx =
|
||||
Some(start_query_result_xlsx_workbook(BufWriter::new(xlsx_file), request, &columns, &[])?);
|
||||
if let Some(writer) = xlsx.as_mut() {
|
||||
writer.write_row(&row).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
writer.write_row(&formatted).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
}
|
||||
}
|
||||
rows_exported += 1;
|
||||
@@ -1115,6 +1136,7 @@ async fn try_export_clickhouse_query_result_stream(
|
||||
if xlsx_hard_limit_active { row_limit.map(|limit| limit.saturating_add(1)) } else { row_limit };
|
||||
let progress_row_interval = request.page_size.max(1) as u64;
|
||||
let mut columns: Vec<String> = Vec::new();
|
||||
let mut temporal_column_types: Vec<String> = Vec::new();
|
||||
let mut rows_exported = 0_u64;
|
||||
let mut last_progress_rows = 0_u64;
|
||||
let mut last_progress_at = Instant::now();
|
||||
@@ -1142,6 +1164,7 @@ async fn try_export_clickhouse_query_result_stream(
|
||||
column_types,
|
||||
} => {
|
||||
columns = stream_columns;
|
||||
temporal_column_types = column_types.clone();
|
||||
if let Some(file) = text_file.as_mut() {
|
||||
let header = format_text_export_header(format, &columns);
|
||||
file.write_all(header.as_bytes()).map_err(|e| format!("Failed to write export header: {e}"))?;
|
||||
@@ -1160,18 +1183,23 @@ async fn try_export_clickhouse_query_result_stream(
|
||||
if xlsx_hard_limit_active && rows_exported as usize >= XLSX_MAX_DATA_ROWS {
|
||||
return Err(XLSX_ROW_LIMIT_ERROR.to_string());
|
||||
}
|
||||
let formatted = crate::temporal_format::format_temporal_export_row_with_string_types(
|
||||
&row,
|
||||
&temporal_column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
if let Some(file) = text_file.as_mut() {
|
||||
let rows = format_text_export_rows(format, std::slice::from_ref(&row));
|
||||
let rows = format_text_export_rows(format, std::slice::from_ref(&formatted));
|
||||
write!(file, "\n{rows}").map_err(|e| format!("Failed to write export rows: {e}"))?;
|
||||
} else if let Some(writer) = xlsx.as_mut() {
|
||||
writer.write_row(&row).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
writer.write_row(&formatted).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
} else {
|
||||
let xlsx_file =
|
||||
File::create(&request.file_path).map_err(|e| format!("Failed to create XLSX file: {e}"))?;
|
||||
xlsx =
|
||||
Some(start_query_result_xlsx_workbook(BufWriter::new(xlsx_file), request, &columns, &[])?);
|
||||
if let Some(writer) = xlsx.as_mut() {
|
||||
writer.write_row(&row).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
writer.write_row(&formatted).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
}
|
||||
}
|
||||
rows_exported += 1;
|
||||
@@ -1267,6 +1295,7 @@ async fn try_export_sqlserver_query_result_stream(
|
||||
let stream_row_limit =
|
||||
if xlsx_hard_limit_active { row_limit.map(|limit| limit.saturating_add(1)) } else { row_limit };
|
||||
let mut columns: Vec<String> = Vec::new();
|
||||
let mut temporal_column_types: Vec<String> = Vec::new();
|
||||
let mut rows_exported = 0_u64;
|
||||
let mut last_progress_rows = 0_u64;
|
||||
let mut last_progress_at = Instant::now();
|
||||
@@ -1299,8 +1328,9 @@ async fn try_export_sqlserver_query_result_stream(
|
||||
cancel_token.clone(),
|
||||
|item| {
|
||||
match item {
|
||||
crate::db::sqlserver::SqlServerStreamItem::Columns(stream_columns) => {
|
||||
crate::db::sqlserver::SqlServerStreamItem::Columns { columns: stream_columns, column_types } => {
|
||||
columns = stream_columns.to_vec();
|
||||
temporal_column_types = column_types.to_vec();
|
||||
if let Some(file) = text_file.as_mut() {
|
||||
let header = format_text_export_header(format, &columns);
|
||||
file.write_all(header.as_bytes()).map_err(|e| format!("Failed to write export header: {e}"))?;
|
||||
@@ -1315,18 +1345,23 @@ async fn try_export_sqlserver_query_result_stream(
|
||||
if xlsx_hard_limit_active && rows_exported as usize >= XLSX_MAX_DATA_ROWS {
|
||||
return Err(XLSX_ROW_LIMIT_ERROR.to_string());
|
||||
}
|
||||
let formatted = crate::temporal_format::format_temporal_export_row_with_string_types(
|
||||
row,
|
||||
&temporal_column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
if let Some(file) = text_file.as_mut() {
|
||||
let rows = format_text_export_rows(format, &[row.to_vec()]);
|
||||
let rows = format_text_export_rows(format, std::slice::from_ref(&formatted));
|
||||
write!(file, "\n{rows}").map_err(|e| format!("Failed to write export rows: {e}"))?;
|
||||
} else if let Some(writer) = xlsx.as_mut() {
|
||||
writer.write_row(row).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
writer.write_row(&formatted).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
} else {
|
||||
let xlsx_file =
|
||||
File::create(&request.file_path).map_err(|e| format!("Failed to create XLSX file: {e}"))?;
|
||||
xlsx =
|
||||
Some(start_query_result_xlsx_workbook(BufWriter::new(xlsx_file), request, &columns, &[])?);
|
||||
if let Some(writer) = xlsx.as_mut() {
|
||||
writer.write_row(row).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
writer.write_row(&formatted).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
}
|
||||
}
|
||||
rows_exported += 1;
|
||||
@@ -1394,6 +1429,7 @@ mod tests {
|
||||
keyset_optimization_enabled: true,
|
||||
client_session_id: None,
|
||||
execution_id: None,
|
||||
date_time_format: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -56,6 +56,8 @@ pub struct TableExportRequest {
|
||||
pub batch_size: Option<usize>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub row_limit: Option<usize>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub date_time_format: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
@@ -516,7 +518,12 @@ async fn try_export_native_table_stream(
|
||||
&cancelled,
|
||||
cancel_token.clone(),
|
||||
|row| {
|
||||
let row_csv = format_csv_rows(&[row.to_vec()]);
|
||||
let formatted = crate::temporal_format::format_temporal_export_row(
|
||||
row,
|
||||
column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
let row_csv = format_csv_rows(&[formatted]);
|
||||
write!(file, "\n{row_csv}").map_err(|e| format!("Failed to write CSV rows: {e}"))?;
|
||||
rows_exported += 1;
|
||||
if rows_exported % progress_interval == 0 {
|
||||
@@ -555,7 +562,12 @@ async fn try_export_native_table_stream(
|
||||
&cancelled,
|
||||
cancel_token.clone(),
|
||||
|row| {
|
||||
let row_tsv = format_tsv_rows(&[row.to_vec()]);
|
||||
let formatted = crate::temporal_format::format_temporal_export_row(
|
||||
row,
|
||||
column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
let row_tsv = format_tsv_rows(&[formatted]);
|
||||
write!(file, "\n{row_tsv}").map_err(|e| format!("Failed to write TXT rows: {e}"))?;
|
||||
rows_exported += 1;
|
||||
if rows_exported % progress_interval == 0 {
|
||||
@@ -578,14 +590,14 @@ async fn try_export_native_table_stream(
|
||||
result
|
||||
}
|
||||
"xlsx" => {
|
||||
let column_types = export_column_types(request);
|
||||
let xlsx_column_types = export_column_types(request);
|
||||
let xlsx_file =
|
||||
std::fs::File::create(&request.file_path).map_err(|e| format!("Failed to create XLSX file: {e}"))?;
|
||||
let mut writer = start_streaming_xlsx_workbook(
|
||||
BufWriter::new(xlsx_file),
|
||||
Some(&request.table_name),
|
||||
col_names,
|
||||
&column_types,
|
||||
&xlsx_column_types,
|
||||
)?;
|
||||
let result = stream_native_table_rows(
|
||||
state,
|
||||
@@ -596,7 +608,12 @@ async fn try_export_native_table_stream(
|
||||
&cancelled,
|
||||
cancel_token.clone(),
|
||||
|row| {
|
||||
writer.write_row(row).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
let formatted = crate::temporal_format::format_temporal_export_row(
|
||||
row,
|
||||
column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
writer.write_row(&formatted).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
rows_exported += 1;
|
||||
if rows_exported % progress_interval == 0 {
|
||||
on_progress(TableExportProgress {
|
||||
@@ -645,7 +662,12 @@ async fn try_export_native_table_stream(
|
||||
if !is_first_row {
|
||||
file.write_all(b",\n").map_err(|e| format!("Failed to write JSON: {e}"))?;
|
||||
}
|
||||
write_json_row_object(&mut file, col_names, row)?;
|
||||
let formatted = crate::temporal_format::format_temporal_export_row(
|
||||
row,
|
||||
column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
write_json_row_object(&mut file, col_names, &formatted)?;
|
||||
is_first_row = false;
|
||||
rows_exported += 1;
|
||||
if rows_exported % progress_interval == 0 {
|
||||
@@ -684,7 +706,12 @@ async fn try_export_native_table_stream(
|
||||
&cancelled,
|
||||
cancel_token.clone(),
|
||||
|row| {
|
||||
let rows_markdown = format_markdown_rows(&[row.to_vec()]);
|
||||
let formatted = crate::temporal_format::format_temporal_export_row(
|
||||
row,
|
||||
column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
let rows_markdown = format_markdown_rows(&[formatted]);
|
||||
if !rows_markdown.is_empty() {
|
||||
if wrote_rows {
|
||||
file.write_all(b"\n").map_err(|e| format!("Failed to write Markdown: {e}"))?;
|
||||
@@ -1016,15 +1043,20 @@ pub async fn export_table_data_core(
|
||||
if row_count == 0 {
|
||||
break;
|
||||
}
|
||||
let formatted_rows = crate::temporal_format::format_temporal_export_rows(
|
||||
&result.rows,
|
||||
&column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
|
||||
if is_first_batch {
|
||||
// First batch: write header + rows via format_csv
|
||||
let csv_content = format_csv(&col_names, &result.rows);
|
||||
let csv_content = format_csv(&col_names, &formatted_rows);
|
||||
file.write_all(csv_content.as_bytes()).map_err(|e| format!("Failed to write CSV: {e}"))?;
|
||||
is_first_batch = false;
|
||||
} else {
|
||||
// Subsequent batches: write rows only (prepend newline for separation)
|
||||
let rows_csv = format_csv_rows(&result.rows);
|
||||
let rows_csv = format_csv_rows(&formatted_rows);
|
||||
if !rows_csv.is_empty() {
|
||||
write!(file, "\n{rows_csv}").map_err(|e| format!("Failed to write CSV rows: {e}"))?;
|
||||
}
|
||||
@@ -1098,13 +1130,18 @@ pub async fn export_table_data_core(
|
||||
if row_count == 0 {
|
||||
break;
|
||||
}
|
||||
let formatted_rows = crate::temporal_format::format_temporal_export_rows(
|
||||
&result.rows,
|
||||
&column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
|
||||
if is_first_batch {
|
||||
let rows_tsv = format_tsv_rows(&result.rows);
|
||||
let rows_tsv = format_tsv_rows(&formatted_rows);
|
||||
write!(file, "\n{rows_tsv}").map_err(|e| format!("Failed to write TXT rows: {e}"))?;
|
||||
is_first_batch = false;
|
||||
} else {
|
||||
let rows_tsv = format_tsv_rows(&result.rows);
|
||||
let rows_tsv = format_tsv_rows(&formatted_rows);
|
||||
if !rows_tsv.is_empty() {
|
||||
write!(file, "\n{rows_tsv}").map_err(|e| format!("Failed to write TXT rows: {e}"))?;
|
||||
}
|
||||
@@ -1135,7 +1172,7 @@ pub async fn export_table_data_core(
|
||||
}
|
||||
}
|
||||
"xlsx" => {
|
||||
let column_types = export_column_types(request);
|
||||
let xlsx_column_types = export_column_types(request);
|
||||
// Create a dedicated file handle for the streaming XLSX writer
|
||||
// instead of cloning the outer BufWriter's handle. This avoids
|
||||
// sharing a file descriptor between two independent buffers.
|
||||
@@ -1145,7 +1182,7 @@ pub async fn export_table_data_core(
|
||||
BufWriter::new(xlsx_file),
|
||||
Some(&request.table_name),
|
||||
&col_names,
|
||||
&column_types,
|
||||
&xlsx_column_types,
|
||||
)?;
|
||||
|
||||
loop {
|
||||
@@ -1188,7 +1225,12 @@ pub async fn export_table_data_core(
|
||||
}
|
||||
|
||||
for row in &result.rows {
|
||||
writer.write_row(row).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
let formatted = crate::temporal_format::format_temporal_export_row(
|
||||
row,
|
||||
&column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
writer.write_row(&formatted).map_err(|e| format!("Failed to write XLSX row: {e}"))?;
|
||||
}
|
||||
rows_exported += row_count as u64;
|
||||
|
||||
@@ -1278,7 +1320,12 @@ pub async fn export_table_data_core(
|
||||
if !is_first_row {
|
||||
file.write_all(b",\n").map_err(|e| format!("Failed to write JSON: {e}"))?;
|
||||
}
|
||||
write_json_row_object(&mut file, &col_names, row)?;
|
||||
let formatted = crate::temporal_format::format_temporal_export_row(
|
||||
row,
|
||||
&column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
write_json_row_object(&mut file, &col_names, &formatted)?;
|
||||
is_first_row = false;
|
||||
}
|
||||
|
||||
@@ -1350,7 +1397,12 @@ pub async fn export_table_data_core(
|
||||
break;
|
||||
}
|
||||
|
||||
let rows_markdown = format_markdown_rows(&result.rows);
|
||||
let formatted_rows = crate::temporal_format::format_temporal_export_rows(
|
||||
&result.rows,
|
||||
&column_types,
|
||||
request.date_time_format.as_deref(),
|
||||
);
|
||||
let rows_markdown = format_markdown_rows(&formatted_rows);
|
||||
if !rows_markdown.is_empty() {
|
||||
if wrote_rows {
|
||||
file.write_all(b"\n").map_err(|e| format!("Failed to write Markdown: {e}"))?;
|
||||
@@ -1622,6 +1674,7 @@ mod tests {
|
||||
skip_count: false,
|
||||
batch_size: Some(500),
|
||||
row_limit: Some(1000),
|
||||
date_time_format: None,
|
||||
};
|
||||
|
||||
let sql = table_cursor_sql(
|
||||
@@ -1669,6 +1722,7 @@ mod tests {
|
||||
skip_count: false,
|
||||
batch_size: Some(100),
|
||||
row_limit: None,
|
||||
date_time_format: None,
|
||||
};
|
||||
let sql = table_cursor_sql(&request, &DatabaseType::Oracle, &columns, &primary_keys);
|
||||
assert_eq!(sql, "SELECT \"ID\", \"NAME\" FROM \"APP\".\"USERS\"");
|
||||
|
||||
@@ -200,6 +200,8 @@ pub struct TableImportRequest {
|
||||
#[serde(default)]
|
||||
pub create_table: bool,
|
||||
pub batch_size: usize,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub date_time_format: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
@@ -1334,6 +1336,29 @@ pub fn build_import_insert_batch_from_rows(
|
||||
table: &str,
|
||||
schema: &str,
|
||||
db_type: &DatabaseType,
|
||||
) -> Result<Option<ImportSqlBatch>, String> {
|
||||
build_import_insert_batch_from_rows_with_format(
|
||||
rows,
|
||||
columns,
|
||||
mappings,
|
||||
target_column_types,
|
||||
table,
|
||||
schema,
|
||||
db_type,
|
||||
None,
|
||||
)
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn build_import_insert_batch_from_rows_with_format(
|
||||
rows: &[Vec<serde_json::Value>],
|
||||
columns: &[String],
|
||||
mappings: &[TableImportColumnMapping],
|
||||
target_column_types: &[(String, String)],
|
||||
table: &str,
|
||||
schema: &str,
|
||||
db_type: &DatabaseType,
|
||||
date_time_format: Option<&str>,
|
||||
) -> Result<Option<ImportSqlBatch>, String> {
|
||||
if rows.is_empty() {
|
||||
return Ok(None);
|
||||
@@ -1365,7 +1390,16 @@ pub fn build_import_insert_batch_from_rows(
|
||||
.map(|row| {
|
||||
mapped
|
||||
.iter()
|
||||
.map(|(source_index, _)| row.get(*source_index).cloned().unwrap_or(serde_json::Value::Null))
|
||||
.enumerate()
|
||||
.map(|(target_index, (source_index, _))| {
|
||||
let value = row.get(*source_index).cloned().unwrap_or(serde_json::Value::Null);
|
||||
normalize_import_temporal_value(
|
||||
&value,
|
||||
column_types.get(target_index).and_then(|data_type| data_type.as_deref()),
|
||||
db_type,
|
||||
date_time_format,
|
||||
)
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
@@ -1377,6 +1411,21 @@ fn supports_multi_row_insert_values(db_type: &DatabaseType) -> bool {
|
||||
!matches!(db_type, DatabaseType::Oracle | DatabaseType::OceanbaseOracle | DatabaseType::Iris)
|
||||
}
|
||||
|
||||
fn normalize_import_temporal_value(
|
||||
value: &serde_json::Value,
|
||||
data_type: Option<&str>,
|
||||
db_type: &DatabaseType,
|
||||
date_time_format: Option<&str>,
|
||||
) -> serde_json::Value {
|
||||
let oracle_date_time = matches!(db_type, DatabaseType::Oracle | DatabaseType::OceanbaseOracle)
|
||||
&& data_type.is_some_and(|data_type| data_type.trim().eq_ignore_ascii_case("date"));
|
||||
crate::temporal_format::normalize_temporal_import_value(
|
||||
value,
|
||||
if oracle_date_time { Some("datetime") } else { data_type },
|
||||
date_time_format,
|
||||
)
|
||||
}
|
||||
|
||||
pub fn build_import_insert_batches(
|
||||
data: &ParsedImportFile,
|
||||
mappings: &[TableImportColumnMapping],
|
||||
@@ -1385,6 +1434,29 @@ pub fn build_import_insert_batches(
|
||||
schema: &str,
|
||||
db_type: &DatabaseType,
|
||||
batch_size: usize,
|
||||
) -> Result<Vec<ImportSqlBatch>, String> {
|
||||
build_import_insert_batches_with_format(
|
||||
data,
|
||||
mappings,
|
||||
target_column_types,
|
||||
table,
|
||||
schema,
|
||||
db_type,
|
||||
batch_size,
|
||||
None,
|
||||
)
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn build_import_insert_batches_with_format(
|
||||
data: &ParsedImportFile,
|
||||
mappings: &[TableImportColumnMapping],
|
||||
target_column_types: &[(String, String)],
|
||||
table: &str,
|
||||
schema: &str,
|
||||
db_type: &DatabaseType,
|
||||
batch_size: usize,
|
||||
date_time_format: Option<&str>,
|
||||
) -> Result<Vec<ImportSqlBatch>, String> {
|
||||
if *db_type == DatabaseType::CloudflareD1 {
|
||||
return crate::db::cloudflare_d1::build_import_insert_batches(
|
||||
@@ -1419,7 +1491,16 @@ pub fn build_import_insert_batches(
|
||||
.map(|row| {
|
||||
mapped
|
||||
.iter()
|
||||
.map(|(source_index, _)| row.get(*source_index).cloned().unwrap_or(serde_json::Value::Null))
|
||||
.enumerate()
|
||||
.map(|(target_index, (source_index, _))| {
|
||||
let value = row.get(*source_index).cloned().unwrap_or(serde_json::Value::Null);
|
||||
normalize_import_temporal_value(
|
||||
&value,
|
||||
column_types.get(target_index).and_then(|data_type| data_type.as_deref()),
|
||||
db_type,
|
||||
date_time_format,
|
||||
)
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
@@ -2036,7 +2117,7 @@ where
|
||||
pending_rows.push(delimited_record_to_row(&record, columns.len(), config));
|
||||
|
||||
if pending_rows.len() >= effective_batch_size {
|
||||
let batch = match build_import_insert_batch_from_rows(
|
||||
let batch = match build_import_insert_batch_from_rows_with_format(
|
||||
&pending_rows,
|
||||
&columns,
|
||||
&request.mappings,
|
||||
@@ -2044,6 +2125,7 @@ where
|
||||
&request.table,
|
||||
&request.schema,
|
||||
db_type,
|
||||
request.date_time_format.as_deref(),
|
||||
) {
|
||||
Ok(Some(batch)) => batch,
|
||||
Ok(None) => {
|
||||
@@ -2086,7 +2168,7 @@ where
|
||||
});
|
||||
return Err("Import cancelled".to_string());
|
||||
}
|
||||
let batch = match build_import_insert_batch_from_rows(
|
||||
let batch = match build_import_insert_batch_from_rows_with_format(
|
||||
&pending_rows,
|
||||
&columns,
|
||||
&request.mappings,
|
||||
@@ -2094,6 +2176,7 @@ where
|
||||
&request.table,
|
||||
&request.schema,
|
||||
db_type,
|
||||
request.date_time_format.as_deref(),
|
||||
) {
|
||||
Ok(Some(batch)) => batch,
|
||||
Ok(None) => ImportSqlBatch { sql: String::new(), row_count: 0 },
|
||||
@@ -2163,7 +2246,7 @@ where
|
||||
target_column_types = created_column_types.clone().unwrap_or_default();
|
||||
}
|
||||
|
||||
let batches = match build_import_insert_batches(
|
||||
let batches = match build_import_insert_batches_with_format(
|
||||
&parsed,
|
||||
&request.mappings,
|
||||
&target_column_types,
|
||||
@@ -2171,6 +2254,7 @@ where
|
||||
&request.schema,
|
||||
db_type,
|
||||
batch_size,
|
||||
request.date_time_format.as_deref(),
|
||||
) {
|
||||
Ok(batches) => batches,
|
||||
Err(error) => {
|
||||
@@ -3115,6 +3199,85 @@ mod tests {
|
||||
}]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_insert_batches_normalize_oracle_unpadded_slash_dates() {
|
||||
let mappings = vec![TableImportColumnMapping {
|
||||
source_column: "created_at".to_string(),
|
||||
target_column: "created_at".to_string(),
|
||||
target_data_type: None,
|
||||
}];
|
||||
let data = ParsedImportFile {
|
||||
columns: vec!["created_at".to_string()],
|
||||
rows: vec![vec![serde_json::json!("2024/2/25 13:02:15")]],
|
||||
total_rows: 1,
|
||||
effective_encoding: None,
|
||||
};
|
||||
|
||||
let batches = build_import_insert_batches(
|
||||
&data,
|
||||
&mappings,
|
||||
&[("created_at".to_string(), "DATE".to_string())],
|
||||
"events",
|
||||
"APP",
|
||||
&DatabaseType::Oracle,
|
||||
500,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
batches[0].sql,
|
||||
"INSERT INTO \"APP\".\"events\" (\"created_at\") VALUES\n(TO_DATE('2024-02-25 13:02:15', 'YYYY-MM-DD HH24:MI:SS'))"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_insert_batch_normalizes_oracle_date_and_timestamp_columns() {
|
||||
let mappings = vec![
|
||||
TableImportColumnMapping {
|
||||
source_column: "event_id".to_string(),
|
||||
target_column: "EVENT_ID".to_string(),
|
||||
target_data_type: None,
|
||||
},
|
||||
TableImportColumnMapping {
|
||||
source_column: "created_at".to_string(),
|
||||
target_column: "CREATED_AT".to_string(),
|
||||
target_data_type: None,
|
||||
},
|
||||
TableImportColumnMapping {
|
||||
source_column: "updated_at".to_string(),
|
||||
target_column: "UPDATED_AT".to_string(),
|
||||
target_data_type: None,
|
||||
},
|
||||
];
|
||||
let rows = vec![vec![
|
||||
serde_json::json!(1),
|
||||
serde_json::json!("2024/2/25 13:02:15"),
|
||||
serde_json::json!("2024/2/25 14:03:16"),
|
||||
]];
|
||||
|
||||
let batch = build_import_insert_batch_from_rows_with_format(
|
||||
&rows,
|
||||
&["event_id".to_string(), "created_at".to_string(), "updated_at".to_string()],
|
||||
&mappings,
|
||||
&[
|
||||
("EVENT_ID".to_string(), "NUMBER".to_string()),
|
||||
("CREATED_AT".to_string(), "DATE".to_string()),
|
||||
("UPDATED_AT".to_string(), "TIMESTAMP(6)".to_string()),
|
||||
],
|
||||
"EVENTS",
|
||||
"APP",
|
||||
&DatabaseType::Oracle,
|
||||
Some("YYYY/M/D HH:mm:ss"),
|
||||
)
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
batch.sql,
|
||||
"INSERT INTO \"APP\".\"EVENTS\" (\"EVENT_ID\", \"CREATED_AT\", \"UPDATED_AT\") VALUES\n(1, TO_DATE('2024-02-25 13:02:15', 'YYYY-MM-DD HH24:MI:SS'), TO_TIMESTAMP('2024-02-25 14:03:16', 'YYYY-MM-DD HH24:MI:SS'))"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_insert_batches_preserve_sqlserver_unicode_text() {
|
||||
let mappings = vec![TableImportColumnMapping {
|
||||
|
||||
@@ -0,0 +1,321 @@
|
||||
use chrono::{DateTime, FixedOffset, NaiveDate, NaiveDateTime, NaiveTime};
|
||||
use serde_json::Value;
|
||||
use std::fmt::Write as _;
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum TemporalKind {
|
||||
Date,
|
||||
Time,
|
||||
DateTime,
|
||||
DateTimeWithTimeZone,
|
||||
}
|
||||
|
||||
enum ParsedTemporal {
|
||||
Zoned(DateTime<FixedOffset>),
|
||||
DateTime(NaiveDateTime),
|
||||
Date(NaiveDate),
|
||||
Time(NaiveTime),
|
||||
}
|
||||
|
||||
fn temporal_kind(data_type: Option<&str>) -> Option<TemporalKind> {
|
||||
let normalized = data_type?.trim().to_ascii_lowercase().replace(char::is_whitespace, " ");
|
||||
let base = normalized.split(['(', ':', ' ']).next().unwrap_or("");
|
||||
if matches!(base, "datetimeoffset" | "datetimeoffsetn" | "timestamptz")
|
||||
|| (base == "timestamp"
|
||||
&& (normalized.contains("with time zone") || normalized.contains("with local time zone")))
|
||||
{
|
||||
return Some(TemporalKind::DateTimeWithTimeZone);
|
||||
}
|
||||
match base {
|
||||
"date" | "date32" | "daten" => Some(TemporalKind::Date),
|
||||
"time" | "time64" | "timen" | "timetz" => Some(TemporalKind::Time),
|
||||
"datetime" | "datetime2" | "datetime4" | "datetime64" | "datetimen" | "smalldatetime" | "timestamp"
|
||||
| "timestampdty" => Some(TemporalKind::DateTime),
|
||||
_ if base.starts_with("timestamp_") => Some(TemporalKind::DateTime),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn dayjs_to_chrono_pattern(pattern: &str) -> Option<String> {
|
||||
let pattern = pattern.trim();
|
||||
if pattern.is_empty() || pattern.len() > 100 || pattern.contains('%') {
|
||||
return None;
|
||||
}
|
||||
let tokens = [
|
||||
("YYYY", "%Y"),
|
||||
("SSS", "%3f"),
|
||||
("ZZ", "%z"),
|
||||
("MM", "%m"),
|
||||
("DD", "%d"),
|
||||
("HH", "%H"),
|
||||
("mm", "%M"),
|
||||
("ss", "%S"),
|
||||
("M", "%-m"),
|
||||
("D", "%-d"),
|
||||
("H", "%-H"),
|
||||
("m", "%-M"),
|
||||
("s", "%-S"),
|
||||
("Z", "%:z"),
|
||||
];
|
||||
let mut output = String::with_capacity(pattern.len() * 2);
|
||||
let mut index = 0;
|
||||
while index < pattern.len() {
|
||||
let remaining = &pattern[index..];
|
||||
if remaining.starts_with('[') {
|
||||
let close = remaining.find(']')?;
|
||||
output.push_str(&remaining[1..close]);
|
||||
index += close + 1;
|
||||
continue;
|
||||
}
|
||||
if let Some((token, replacement)) = tokens.iter().find(|(token, _)| remaining.starts_with(token)) {
|
||||
output.push_str(replacement);
|
||||
index += token.len();
|
||||
continue;
|
||||
}
|
||||
let ch = remaining.chars().next()?;
|
||||
// Reject unknown Day.js tokens instead of silently exporting different text than the frontend displays.
|
||||
if ch.is_ascii_alphabetic() {
|
||||
return None;
|
||||
}
|
||||
output.push(ch);
|
||||
index += ch.len_utf8();
|
||||
}
|
||||
Some(output)
|
||||
}
|
||||
|
||||
fn parse_with_pattern(value: &str, pattern: &str) -> Option<ParsedTemporal> {
|
||||
let pattern = dayjs_to_chrono_pattern(pattern)?;
|
||||
DateTime::parse_from_str(value, &pattern)
|
||||
.map(ParsedTemporal::Zoned)
|
||||
.ok()
|
||||
.or_else(|| NaiveDateTime::parse_from_str(value, &pattern).map(ParsedTemporal::DateTime).ok())
|
||||
.or_else(|| NaiveDate::parse_from_str(value, &pattern).map(ParsedTemporal::Date).ok())
|
||||
.or_else(|| NaiveTime::parse_from_str(value, &pattern).map(ParsedTemporal::Time).ok())
|
||||
}
|
||||
|
||||
fn parse_known_temporal(value: &str) -> Option<ParsedTemporal> {
|
||||
if let Ok(parsed) = DateTime::parse_from_rfc3339(value) {
|
||||
return Some(ParsedTemporal::Zoned(parsed));
|
||||
}
|
||||
for pattern in ["%Y-%m-%d %H:%M:%S%.f", "%Y-%m-%dT%H:%M:%S%.f", "%Y/%m/%d %H:%M:%S%.f", "%Y/%m/%dT%H:%M:%S%.f"] {
|
||||
if let Ok(parsed) = NaiveDateTime::parse_from_str(value, pattern) {
|
||||
return Some(ParsedTemporal::DateTime(parsed));
|
||||
}
|
||||
}
|
||||
for pattern in ["%Y-%m-%d", "%Y/%m/%d"] {
|
||||
if let Ok(parsed) = NaiveDate::parse_from_str(value, pattern) {
|
||||
return Some(ParsedTemporal::Date(parsed));
|
||||
}
|
||||
}
|
||||
for pattern in ["%H:%M:%S%.f", "%H:%M:%S"] {
|
||||
if let Ok(parsed) = NaiveTime::parse_from_str(value, pattern) {
|
||||
return Some(ParsedTemporal::Time(parsed));
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
fn parse_temporal(value: &str, preferred_pattern: Option<&str>) -> Option<ParsedTemporal> {
|
||||
preferred_pattern
|
||||
.filter(|pattern| !pattern.trim().is_empty())
|
||||
.and_then(|pattern| parse_with_pattern(value, pattern))
|
||||
.or_else(|| parse_known_temporal(value))
|
||||
}
|
||||
|
||||
fn format_parsed(parsed: ParsedTemporal, pattern: &str) -> Option<String> {
|
||||
let pattern = dayjs_to_chrono_pattern(pattern)?;
|
||||
let mut output = String::new();
|
||||
// Chrono reports missing date/time fields through fmt::Error; propagate it so exports preserve the raw value.
|
||||
match parsed {
|
||||
ParsedTemporal::Zoned(value) => write!(&mut output, "{}", value.format(&pattern)),
|
||||
ParsedTemporal::DateTime(value) => write!(&mut output, "{}", value.format(&pattern)),
|
||||
ParsedTemporal::Date(value) => write!(&mut output, "{}", value.format(&pattern)),
|
||||
ParsedTemporal::Time(value) => write!(&mut output, "{}", value.format(&pattern)),
|
||||
}
|
||||
.ok()?;
|
||||
Some(output)
|
||||
}
|
||||
|
||||
pub fn format_temporal_export_value(value: &Value, data_type: Option<&str>, pattern: Option<&str>) -> Value {
|
||||
let Some(pattern) = pattern.filter(|pattern| !pattern.trim().is_empty()) else {
|
||||
return value.clone();
|
||||
};
|
||||
if temporal_kind(data_type).is_none() {
|
||||
return value.clone();
|
||||
}
|
||||
let Some(raw) = value.as_str() else {
|
||||
return value.clone();
|
||||
};
|
||||
parse_known_temporal(raw)
|
||||
.and_then(|parsed| format_parsed(parsed, pattern))
|
||||
.map(Value::String)
|
||||
.unwrap_or_else(|| value.clone())
|
||||
}
|
||||
|
||||
pub fn format_temporal_export_row(row: &[Value], column_types: &[Option<String>], pattern: Option<&str>) -> Vec<Value> {
|
||||
row.iter()
|
||||
.enumerate()
|
||||
.map(|(index, value)| {
|
||||
format_temporal_export_value(
|
||||
value,
|
||||
column_types.get(index).and_then(|data_type| data_type.as_deref()),
|
||||
pattern,
|
||||
)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub fn format_temporal_export_rows(
|
||||
rows: &[Vec<Value>],
|
||||
column_types: &[Option<String>],
|
||||
pattern: Option<&str>,
|
||||
) -> Vec<Vec<Value>> {
|
||||
rows.iter().map(|row| format_temporal_export_row(row, column_types, pattern)).collect()
|
||||
}
|
||||
|
||||
pub fn format_temporal_export_row_with_string_types(
|
||||
row: &[Value],
|
||||
column_types: &[String],
|
||||
pattern: Option<&str>,
|
||||
) -> Vec<Value> {
|
||||
row.iter()
|
||||
.enumerate()
|
||||
.map(|(index, value)| format_temporal_export_value(value, column_types.get(index).map(String::as_str), pattern))
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub fn format_temporal_export_rows_with_string_types(
|
||||
rows: &[Vec<Value>],
|
||||
column_types: &[String],
|
||||
pattern: Option<&str>,
|
||||
) -> Vec<Vec<Value>> {
|
||||
rows.iter().map(|row| format_temporal_export_row_with_string_types(row, column_types, pattern)).collect()
|
||||
}
|
||||
|
||||
pub fn normalize_temporal_import_value(value: &Value, data_type: Option<&str>, pattern: Option<&str>) -> Value {
|
||||
let Some(kind) = temporal_kind(data_type) else {
|
||||
return value.clone();
|
||||
};
|
||||
let Some(raw) = value.as_str() else {
|
||||
return value.clone();
|
||||
};
|
||||
let Some(parsed) = parse_temporal(raw.trim(), pattern) else {
|
||||
return value.clone();
|
||||
};
|
||||
|
||||
let normalized = match (kind, parsed) {
|
||||
(TemporalKind::Date, ParsedTemporal::Zoned(value)) => value.date_naive().format("%Y-%m-%d").to_string(),
|
||||
(TemporalKind::Date, ParsedTemporal::DateTime(value)) => value.date().format("%Y-%m-%d").to_string(),
|
||||
(TemporalKind::Date, ParsedTemporal::Date(value)) => value.format("%Y-%m-%d").to_string(),
|
||||
(TemporalKind::Date, ParsedTemporal::Time(_)) => return value.clone(),
|
||||
(TemporalKind::Time, ParsedTemporal::Zoned(value)) => value.time().format("%H:%M:%S%.f").to_string(),
|
||||
(TemporalKind::Time, ParsedTemporal::DateTime(value)) => value.time().format("%H:%M:%S%.f").to_string(),
|
||||
(TemporalKind::Time, ParsedTemporal::Time(value)) => value.format("%H:%M:%S%.f").to_string(),
|
||||
(TemporalKind::Time, ParsedTemporal::Date(_)) => return value.clone(),
|
||||
(TemporalKind::DateTime, ParsedTemporal::Zoned(value)) => {
|
||||
value.naive_local().format("%Y-%m-%d %H:%M:%S%.f").to_string()
|
||||
}
|
||||
(TemporalKind::DateTime, ParsedTemporal::DateTime(value)) => value.format("%Y-%m-%d %H:%M:%S%.f").to_string(),
|
||||
(TemporalKind::DateTime, ParsedTemporal::Date(date)) => {
|
||||
let Some(value) = date.and_hms_opt(0, 0, 0) else {
|
||||
return value.clone();
|
||||
};
|
||||
value.format("%Y-%m-%d %H:%M:%S").to_string()
|
||||
}
|
||||
(TemporalKind::DateTime, ParsedTemporal::Time(_)) => return value.clone(),
|
||||
(TemporalKind::DateTimeWithTimeZone, ParsedTemporal::Zoned(value)) => {
|
||||
value.format("%Y-%m-%dT%H:%M:%S%.f%:z").to_string()
|
||||
}
|
||||
(TemporalKind::DateTimeWithTimeZone, ParsedTemporal::DateTime(value)) => {
|
||||
value.format("%Y-%m-%d %H:%M:%S%.f").to_string()
|
||||
}
|
||||
(TemporalKind::DateTimeWithTimeZone, ParsedTemporal::Date(date)) => {
|
||||
let Some(value) = date.and_hms_opt(0, 0, 0) else {
|
||||
return value.clone();
|
||||
};
|
||||
value.format("%Y-%m-%d %H:%M:%S").to_string()
|
||||
}
|
||||
(TemporalKind::DateTimeWithTimeZone, ParsedTemporal::Time(_)) => return value.clone(),
|
||||
};
|
||||
Value::String(normalized)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use serde_json::json;
|
||||
|
||||
#[test]
|
||||
fn normalizes_unpadded_slash_dates_for_import() {
|
||||
assert_eq!(
|
||||
normalize_temporal_import_value(&json!("2024/2/25 13:02:15"), Some("DATE"), None),
|
||||
json!("2024-02-25")
|
||||
);
|
||||
assert_eq!(
|
||||
normalize_temporal_import_value(
|
||||
&json!("25.02.2024 13:02:15"),
|
||||
Some("TIMESTAMP(6)"),
|
||||
Some("DD.MM.YYYY HH:mm:ss")
|
||||
),
|
||||
json!("2024-02-25 13:02:15")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn formats_only_typed_temporal_export_values() {
|
||||
let row = vec![json!(1), json!("2024-02-25 13:02:15"), json!("2024-02-25 13:02:15")];
|
||||
assert_eq!(
|
||||
format_temporal_export_row(
|
||||
&row,
|
||||
&[Some("NUMBER".into()), Some("TIMESTAMP".into()), Some("VARCHAR2".into())],
|
||||
Some("YYYY/M/D HH:mm:ss")
|
||||
),
|
||||
vec![json!(1), json!("2024/2/25 13:02:15"), json!("2024-02-25 13:02:15")]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn preserves_raw_export_values_when_pattern_requires_missing_fields() {
|
||||
assert_eq!(
|
||||
format_temporal_export_value(&json!("2024-02-25"), Some("DATE"), Some("YYYY-MM-DD HH:mm:ss")),
|
||||
json!("2024-02-25")
|
||||
);
|
||||
assert_eq!(
|
||||
format_temporal_export_value(&json!("13:02:15"), Some("TIME"), Some("YYYY-MM-DD HH:mm:ss")),
|
||||
json!("13:02:15")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_unsupported_dayjs_tokens_but_allows_literal_text() {
|
||||
assert_eq!(dayjs_to_chrono_pattern("MM/DD/YYYY hh:mm A"), None);
|
||||
assert_eq!(dayjs_to_chrono_pattern("YYYY-MM-DD [at] HH:mm:ss"), Some("%Y-%m-%d at %H:%M:%S".into()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recognizes_common_driver_temporal_type_aliases() {
|
||||
for data_type in ["DateTime64(3)", "date32", "timestamp_ns", "TimeStampDTY", "datetimeoffsetn", "timen"] {
|
||||
assert!(temporal_kind(Some(data_type)).is_some(), "{data_type}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn export_formatting_preserves_offset_datetime_fields() {
|
||||
assert_eq!(
|
||||
format_temporal_export_value(&json!("2024-02-25T13:02:15Z"), Some("DATE"), Some("YYYY/M/D HH:mm:ss")),
|
||||
json!("2024/2/25 13:02:15")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn import_normalization_preserves_timezone_offsets() {
|
||||
assert_eq!(
|
||||
normalize_temporal_import_value(
|
||||
&json!("2024-02-25T13:02:15+08:00"),
|
||||
Some("timestamp with time zone"),
|
||||
None
|
||||
),
|
||||
json!("2024-02-25T13:02:15+08:00")
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -1073,8 +1073,8 @@ pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, colum
|
||||
if let Some(numeric_literal) = format_mysql_numeric_string_literal(s, db_type, column_type) {
|
||||
return numeric_literal;
|
||||
}
|
||||
if let Some(date_literal) = format_oracle_date_sql_literal(s, db_type, column_type) {
|
||||
return date_literal;
|
||||
if let Some(temporal_literal) = format_oracle_temporal_sql_literal(s, db_type, column_type) {
|
||||
return temporal_literal;
|
||||
}
|
||||
|
||||
let literal = format_literal_string(s, db_type, column_type);
|
||||
@@ -1178,21 +1178,46 @@ fn format_mysql_binary_sql_literal(value: &str, db_type: &DatabaseType, column_t
|
||||
}
|
||||
}
|
||||
|
||||
fn format_oracle_date_sql_literal(value: &str, db_type: &DatabaseType, column_type: Option<&str>) -> Option<String> {
|
||||
fn format_oracle_temporal_sql_literal(
|
||||
value: &str,
|
||||
db_type: &DatabaseType,
|
||||
column_type: Option<&str>,
|
||||
) -> Option<String> {
|
||||
if !matches!(db_type, DatabaseType::Oracle | DatabaseType::OceanbaseOracle) {
|
||||
return None;
|
||||
}
|
||||
if temporal_column_kind(column_type) != Some("date") {
|
||||
return None;
|
||||
}
|
||||
let kind = temporal_column_kind(column_type)?;
|
||||
let normalized_column_type = column_type?.trim().to_ascii_lowercase();
|
||||
let parts = oracle_export_date_parts(value)?;
|
||||
Some(format_oracle_date_sql_literal_parts(&parts))
|
||||
match kind {
|
||||
"date" => Some(format_oracle_date_sql_literal_parts(&parts)),
|
||||
"datetime"
|
||||
if (normalized_column_type.contains("with time zone")
|
||||
|| normalized_column_type.contains("with local time zone"))
|
||||
&& parts.zone.is_some() =>
|
||||
{
|
||||
let fraction = parts.fraction.unwrap_or_default();
|
||||
let mask = if fraction.is_empty() { "YYYY-MM-DD HH24:MI:SS" } else { "YYYY-MM-DD HH24:MI:SS.FF" };
|
||||
let zone = match parts.zone.unwrap_or_default() {
|
||||
"Z" | "z" => "+00:00",
|
||||
zone => zone,
|
||||
};
|
||||
Some(format!("TO_TIMESTAMP_TZ('{} {}{fraction} {zone}', '{mask} TZH:TZM')", parts.date, parts.time))
|
||||
}
|
||||
"datetime" => {
|
||||
let fraction = parts.fraction.unwrap_or_default();
|
||||
let mask = if fraction.is_empty() { "YYYY-MM-DD HH24:MI:SS" } else { "YYYY-MM-DD HH24:MI:SS.FF" };
|
||||
Some(format!("TO_TIMESTAMP('{} {}{fraction}', '{mask}')", parts.date, parts.time))
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
struct OracleExportDateParts<'a> {
|
||||
date: &'a str,
|
||||
time: &'a str,
|
||||
fraction: Option<&'a str>,
|
||||
zone: Option<&'a str>,
|
||||
}
|
||||
|
||||
fn format_oracle_date_sql_literal_parts(parts: &OracleExportDateParts<'_>) -> String {
|
||||
@@ -1218,7 +1243,7 @@ fn oracle_export_date_parts(value: &str) -> Option<OracleExportDateParts<'_>> {
|
||||
return None;
|
||||
}
|
||||
if bytes.len() == 10 {
|
||||
return Some(OracleExportDateParts { date, time: "00:00:00", fraction: None });
|
||||
return Some(OracleExportDateParts { date, time: "00:00:00", fraction: None, zone: None });
|
||||
}
|
||||
let separator = *bytes.get(10)?;
|
||||
if separator != b'T' && separator != b' ' {
|
||||
@@ -1233,7 +1258,7 @@ fn oracle_export_date_parts(value: &str) -> Option<OracleExportDateParts<'_>> {
|
||||
}
|
||||
let rest = &value[19..];
|
||||
if rest.is_empty() || is_timezone_suffix(rest) {
|
||||
return Some(OracleExportDateParts { date, time, fraction: None });
|
||||
return Some(OracleExportDateParts { date, time, fraction: None, zone: (!rest.is_empty()).then_some(rest) });
|
||||
}
|
||||
if let Some(after_dot) = rest.strip_prefix('.') {
|
||||
let digit_count = after_dot.bytes().take_while(|byte| byte.is_ascii_digit()).count();
|
||||
@@ -1242,7 +1267,12 @@ fn oracle_export_date_parts(value: &str) -> Option<OracleExportDateParts<'_>> {
|
||||
}
|
||||
let zone = &after_dot[digit_count..];
|
||||
if zone.is_empty() || is_timezone_suffix(zone) {
|
||||
return Some(OracleExportDateParts { date, time, fraction: Some(&value[19..19 + 1 + digit_count]) });
|
||||
return Some(OracleExportDateParts {
|
||||
date,
|
||||
time,
|
||||
fraction: Some(&value[19..19 + 1 + digit_count]),
|
||||
zone: (!zone.is_empty()).then_some(zone),
|
||||
});
|
||||
}
|
||||
}
|
||||
None
|
||||
@@ -5987,12 +6017,20 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
sql,
|
||||
"INSERT INTO \"APP\".\"events\" (\"id\", \"created_on\", \"created_at\", \"raw_text\") VALUES\n(1, TO_DATE('2022-08-25 09:58:43', 'YYYY-MM-DD HH24:MI:SS'), '2022-08-25T09:58:43Z', '2022-08-25T09:58:43Z')"
|
||||
"INSERT INTO \"APP\".\"events\" (\"id\", \"created_on\", \"created_at\", \"raw_text\") VALUES\n(1, TO_DATE('2022-08-25 09:58:43', 'YYYY-MM-DD HH24:MI:SS'), TO_TIMESTAMP('2022-08-25 09:58:43', 'YYYY-MM-DD HH24:MI:SS'), '2022-08-25T09:58:43Z')"
|
||||
);
|
||||
assert_eq!(
|
||||
escape_value_typed(&json!("2022-08-25T00:00:00Z"), &DatabaseType::Oracle, Some("DATE")),
|
||||
"DATE '2022-08-25'"
|
||||
);
|
||||
assert_eq!(
|
||||
escape_value_typed(
|
||||
&json!("2022-08-25T09:58:43.123456+08:00"),
|
||||
&DatabaseType::Oracle,
|
||||
Some("TIMESTAMP(6) WITH TIME ZONE")
|
||||
),
|
||||
"TO_TIMESTAMP_TZ('2022-08-25 09:58:43.123456 +08:00', 'YYYY-MM-DD HH24:MI:SS.FF TZH:TZM')"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -98,6 +98,7 @@ async fn live_clickhouse_query_result_export_xlsx_streams_random_order_query_onc
|
||||
keyset_optimization_enabled: false,
|
||||
client_session_id: None,
|
||||
execution_id: Some(format!("live-clickhouse-query-export-{suffix}")),
|
||||
date_time_format: None,
|
||||
};
|
||||
let done_seen = AtomicBool::new(false);
|
||||
let result = export_query_result_core(&state, &request, None, |progress| {
|
||||
|
||||
@@ -165,6 +165,7 @@ async fn live_mysql_query_result_export_xlsx_streams_single_query_without_duplic
|
||||
keyset_optimization_enabled: false,
|
||||
client_session_id: None,
|
||||
execution_id: Some(format!("live-mysql-query-export-{suffix}")),
|
||||
date_time_format: None,
|
||||
};
|
||||
let done_seen = AtomicBool::new(false);
|
||||
let result = export_query_result_core(&state, &request, None, |progress| {
|
||||
|
||||
@@ -130,6 +130,7 @@ async fn live_postgres_query_result_export_uses_single_streamed_query() {
|
||||
keyset_optimization_enabled: true,
|
||||
client_session_id: None,
|
||||
execution_id: Some(format!("live-postgres-query-export-{suffix}")),
|
||||
date_time_format: None,
|
||||
};
|
||||
let done_seen = AtomicBool::new(false);
|
||||
let result = export_query_result_core(&state, &request, None, |progress| {
|
||||
|
||||
@@ -279,7 +279,7 @@ async fn live_sqlserver_stream_first_result_set_exports_cte_query_rows() {
|
||||
let mut rows = Vec::new();
|
||||
let result = dbx_core::db::sqlserver::stream_first_result_set(&mut client, &sql, None, None, |item| {
|
||||
match item {
|
||||
dbx_core::db::sqlserver::SqlServerStreamItem::Columns(stream_columns) => {
|
||||
dbx_core::db::sqlserver::SqlServerStreamItem::Columns { columns: stream_columns, .. } => {
|
||||
columns = stream_columns.to_vec();
|
||||
}
|
||||
dbx_core::db::sqlserver::SqlServerStreamItem::Row(row) => {
|
||||
@@ -369,6 +369,7 @@ async fn live_sqlserver_query_result_export_streams_cte_query_to_csv() {
|
||||
keyset_optimization_enabled: true,
|
||||
client_session_id: None,
|
||||
execution_id: Some(format!("live-sqlserver-export-{suffix}")),
|
||||
date_time_format: None,
|
||||
};
|
||||
let done_seen = AtomicBool::new(false);
|
||||
let result = export_query_result_core(&state, &request, None, |progress| {
|
||||
|
||||
Reference in new issue
Block a user