github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-19 16:16:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(csv-import): batch OceanBase Oracle imports through INSERT ALL

Browse Source
This commit is contained in:
Dagest
2026-09-28 21:32:13 +08:00
committed by GitHub
parent 516c04b8ee
commit b3bffe8c71
2 changed files with 135 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/dbx-core/src/data/table_import.rs
+55 -3
View File
@@ -4579,8 +4579,8 @@ fn effective_import_batch_size(db_type: &DatabaseType, requested: usize) -> usiz
// Some backends impose stricter limits than the UI batch setting; clamp here so every
// import path, including streaming producers, uses the same safe value.
let max_rows = match db_type {
DatabaseType::Oracle => MAX_ORACLE_IMPORT_BATCH_ROWS,
DatabaseType::OceanbaseOracle | DatabaseType::Iris => 1,
DatabaseType::Oracle | DatabaseType::OceanbaseOracle => MAX_ORACLE_IMPORT_BATCH_ROWS,
DatabaseType::Iris => 1,
DatabaseType::CloudflareD1 => 100,
DatabaseType::SqlServer => 1000,
DatabaseType::Sqlite => SQLITE_APPEND_COMMIT_ROWS,
@@ -11748,10 +11748,62 @@ mod tests {
assert_eq!(batches[1].row_count, 1);
}
#[test]
fn oceanbase_oracle_import_batches_rows_through_insert_all() {
let mappings = vec![TableImportColumnMapping {
source_column: "id".to_string(),
target_column: "id".to_string(),
target_data_type: None,
}];
let data = ParsedImportFile {
columns: vec!["id".to_string()],
rows: vec![vec![serde_json::json!(1)], vec![serde_json::json!(2)], vec![serde_json::json!(3)]],
total_rows: 3,
effective_encoding: None,
};
let batches =
build_import_insert_batches(&data, &mappings, &[], "items", "SQLUSER", &DatabaseType::OceanbaseOracle, 100)
.unwrap();
assert_eq!(batches.len(), 1);
assert_eq!(
batches[0].sql,
"INSERT ALL\nINTO \"SQLUSER\".\"items\" (\"id\") VALUES (1)\nINTO \"SQLUSER\".\"items\" (\"id\") VALUES (2)\nINTO \"SQLUSER\".\"items\" (\"id\") VALUES (3)\nSELECT 1 FROM dual"
);
assert_eq!(batches[0].row_count, 3);
}
#[test]
fn oceanbase_oracle_import_keeps_single_row_values_statement() {
let mappings = vec![TableImportColumnMapping {
source_column: "id".to_string(),
target_column: "id".to_string(),
target_data_type: None,
}];
let data = ParsedImportFile {
columns: vec!["id".to_string()],
rows: vec![vec![serde_json::json!(1)]],
total_rows: 1,
effective_encoding: None,
};
let batches =
build_import_insert_batches(&data, &mappings, &[], "items", "SQLUSER", &DatabaseType::OceanbaseOracle, 100)
.unwrap();
assert_eq!(batches.len(), 1);
assert_eq!(batches[0].sql, "INSERT INTO \"SQLUSER\".\"items\" (\"id\") VALUES\n(1)");
assert_eq!(batches[0].row_count, 1);
}
#[test]
fn import_batch_row_limits_match_database_dialects() {
assert_eq!(effective_import_batch_size(&DatabaseType::Oracle, 1000), 500);
assert_eq!(effective_import_batch_size(&DatabaseType::OceanbaseOracle, 1000), 1);
// OceanBase's Oracle mode shares the INSERT ALL template with Oracle, so the
// importer may batch rows instead of issuing one INSERT per row.
assert_eq!(effective_import_batch_size(&DatabaseType::OceanbaseOracle, 1000), 500);
assert_eq!(effective_import_batch_size(&DatabaseType::OceanbaseOracle, 1), 1);
assert_eq!(effective_import_batch_size(&DatabaseType::Iris, 1000), 1);
assert_eq!(effective_import_batch_size(&DatabaseType::CloudflareD1, 1000), 100);
assert_eq!(effective_import_batch_size(&DatabaseType::SqlServer, 1001), 1000);
crates/dbx-core/src/data/transfer.rs
+80 -4
View File
@@ -120,7 +120,7 @@ impl SqlBatchLimits {
pub(crate) fn for_database(db_type: &DatabaseType, requested_max_rows: usize) -> Self {
let max_rows = requested_max_rows.max(1).min(match db_type {
DatabaseType::SqlServer => MAX_SQLSERVER_INSERT_ROWS,
DatabaseType::Oracle => MAX_ORACLE_INSERT_ALL_ROWS,
DatabaseType::Oracle | DatabaseType::OceanbaseOracle => MAX_ORACLE_INSERT_ALL_ROWS,
_ => usize::MAX,
});
let target_sql_bytes = match db_type {
@@ -3920,7 +3920,12 @@ impl InsertSqlTemplate {
};
Self {
standard_prefix: format!("INSERT INTO {full_table} ({col_list}){overriding} VALUES\n"),
oracle_into_prefix: matches!(db_type, DatabaseType::Oracle)
// OceanBase's Oracle mode rejects the comma-separated multi-row VALUES form
// just like Oracle itself, but it does implement Oracle's multi-table
// `INSERT ALL ... SELECT 1 FROM dual` syntax, so it shares this template.
// Without it the importer falls back to single-row INSERTs (one network
// round trip per row), which makes CSV imports orders of magnitude slower.
oracle_into_prefix: matches!(db_type, DatabaseType::Oracle | DatabaseType::OceanbaseOracle)
.then(|| format!("INTO {full_table} ({col_list}) VALUES ")),
// Xugu accepts consecutive row constructors (`VALUES (...) (...)`) but rejects
// the comma-separated multi-row form emitted by the generic template.
@@ -4396,6 +4401,9 @@ fn max_transfer_write_rows(db_type: &DatabaseType, mode: &TransferMode) -> usize
(DatabaseType::SqlServer, TransferMode::Append | TransferMode::Overwrite) => MAX_SQLSERVER_INSERT_ROWS,
(DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo, _) => 500,
(DatabaseType::Oracle, TransferMode::Append | TransferMode::Overwrite) => MAX_ORACLE_INSERT_ALL_ROWS,
// The INSERT ALL template is shared with OceanBase's Oracle mode, so its
// append/overwrite batches must clamp to the same per-statement row count.
(DatabaseType::OceanbaseOracle, TransferMode::Append | TransferMode::Overwrite) => MAX_ORACLE_INSERT_ALL_ROWS,
(DatabaseType::Oracle, TransferMode::Upsert) => MAX_ORACLE_MERGE_ROWS,
_ => usize::MAX,
}
@@ -4814,7 +4822,7 @@ pub(crate) fn generate_insert_typed_sql_batches_from_value_rows_with_options(
let max_rows = limits.max_rows.max(1).min(match db_type {
DatabaseType::SqlServer => MAX_SQLSERVER_INSERT_ROWS,
DatabaseType::Oracle => MAX_ORACLE_INSERT_ALL_ROWS,
DatabaseType::Oracle | DatabaseType::OceanbaseOracle => MAX_ORACLE_INSERT_ALL_ROWS,
_ => usize::MAX,
});
let target_sql_bytes = limits.target_sql_bytes.max(1);
@@ -4995,7 +5003,7 @@ fn generate_insert_sql_batches_from_value_rows(
let max_rows = limits.max_rows.max(1).min(match db_type {
DatabaseType::SqlServer => MAX_SQLSERVER_INSERT_ROWS,
DatabaseType::Oracle => MAX_ORACLE_INSERT_ALL_ROWS,
DatabaseType::Oracle | DatabaseType::OceanbaseOracle => MAX_ORACLE_INSERT_ALL_ROWS,
_ => usize::MAX,
});
let target_sql_bytes = limits.target_sql_bytes.max(1);
@@ -16747,6 +16755,74 @@ SELECT 1 FROM dual"#
);
}
#[test]
fn oceanbase_oracle_multi_row_insert_uses_insert_all() {
// OceanBase's Oracle mode rejects multi-row VALUES but implements Oracle's
// INSERT ALL, so the import/transfer generators must batch through it instead
// of degrading to one INSERT per row.
let sql = generate_insert_typed(
&[String::from("id"), String::from("name")],
&[Some(String::from("number")), Some(String::from("varchar2(64)"))],
&[vec![json!(1), json!("Ada")], vec![json!(2), json!("O'Brien")]],
"INSTR_CATEGORY",
"APP",
&DatabaseType::OceanbaseOracle,
None,
);
assert_eq!(
sql,
r#"INSERT ALL
INTO "APP"."INSTR_CATEGORY" ("id", "name") VALUES (1, 'Ada')
INTO "APP"."INSTR_CATEGORY" ("id", "name") VALUES (2, 'O''Brien')
SELECT 1 FROM dual"#
);
assert!(!sql.contains("),\n("));
}
#[test]
fn oceanbase_oracle_single_row_insert_keeps_values_shape() {
let sql = generate_insert_typed(
&[String::from("id"), String::from("name")],
&[Some(String::from("number")), Some(String::from("varchar2(64)"))],
&[vec![json!(1), json!("Ada")]],
"INSTR_CATEGORY",
"APP",
&DatabaseType::OceanbaseOracle,
None,
);
assert_eq!(
sql,
r#"INSERT INTO "APP"."INSTR_CATEGORY" ("id", "name") VALUES
(1, 'Ada')"#
);
}
#[test]
fn oceanbase_oracle_transfer_write_batches_limit_insert_all_rows() {
let rows = (0..(MAX_ORACLE_INSERT_ALL_ROWS + 1)).map(|index| vec![json!(index)]).collect::<Vec<_>>();
let statements = generate_transfer_write_sql_batches(
&TransferMode::Append,
&[String::from("id")],
&[Some(String::from("number"))],
&rows,
"INSTR_CATEGORY",
"APP",
&DatabaseType::OceanbaseOracle,
&[],
None,
false,
false,
)
.unwrap();
assert_eq!(statements.len(), 2);
assert_eq!(statements[0].matches("\nINTO ").count(), MAX_ORACLE_INSERT_ALL_ROWS);
assert!(statements[0].starts_with("INSERT ALL\nINTO "));
assert!(statements[0].ends_with("SELECT 1 FROM dual"));
}
#[test]
fn oracle_transfer_write_batches_limit_insert_all_rows() {
let rows = (0..(MAX_ORACLE_INSERT_ALL_ROWS + 1)).map(|index| vec![json!(index)]).collect::<Vec<_>>();