feat(transfer): support h2 table data transfer

This commit is contained in:
Hand Sonic authored and GitHub committed 2026-09-17 00:36:11 +08:00
1 parent 1c937f0cfc
commit 13a209d4c2
10 files changed
+665 -25

No files matched your search

@@ -1692,7 +1692,7 @@
"tableDataEdit": true,
"tableStructureEdit": true,
"tableImport": false,
"dataTransfer": false,
"dataTransfer": true,
"sqlFileExecution": true,
"databaseCreate": false,
"fieldLineage": false,
+331 -11
View File
@@ -1293,6 +1293,27 @@ fn rewrite_double_quoted_schema_qualifier(ddl: &str, source_schema: &str, target
map_sql_code_spans(ddl, false, |code| code.replace(&source, &target))
}
fn rewrite_h2_schema_qualifier(ddl: &str, source_schema: &str, target_schema: &str) -> String {
if source_schema == target_schema || source_schema.is_empty() {
return ddl.to_string();
}
let source = format!("\"{}\"", source_schema.replace('"', "\"\""));
let target = format!("\"{}\"", target_schema.replace('"', "\"\""));
let identifiers = Regex::new(r#""(?:""|[^"])*""#).expect("valid quoted identifier regex");
map_sql_code_spans(ddl, false, |code| {
identifiers
.replace_all(code, |captures: &regex::Captures| {
let identifier = captures.get(0).unwrap();
if identifier.as_str() == source && code[identifier.end()..].trim_start().starts_with('.') {
target.clone()
} else {
identifier.as_str().to_string()
}
})
.into_owned()
})
}
pub fn rewrite_oracle_schema_qualifier(ddl: &str, source_schema: &str, target_schema: &str) -> String {
rewrite_double_quoted_schema_qualifier(ddl, source_schema, target_schema)
}
@@ -1685,6 +1706,34 @@ fn target_column_can_be_omitted(column: &db::ColumnInfo, target_db_type: &Databa
|| (matches!(target_db_type, DatabaseType::SqlServer) && is_sqlserver_rowversion_type(&column.data_type))
}
fn h2_writable_transfer_columns(
source_columns: &[db::ColumnInfo],
target_columns: &[db::ColumnInfo],
mode: &TransferMode,
target_table: &str,
) -> Result<Vec<db::ColumnInfo>, String> {
let mut columns = Vec::new();
for source in source_columns {
let target = target_columns.iter().find(|target| target.name == source.name);
if target.is_some_and(|target| target.extra.as_deref() == Some("computed")) {
continue;
}
if *mode == TransferMode::Upsert
&& target.is_some_and(|target| is_postgres_generated_always_identity_extra(target.extra.as_deref()))
{
return Err(format!("H2 Upsert cannot assign GENERATED ALWAYS identity column '{}'. Use append/overwrite or change the target column to GENERATED BY DEFAULT.", source.name));
}
columns.push(source.clone());
}
if columns.is_empty() {
return Err("No writable H2 target columns found".to_string());
}
// Data-only overwrite must also fail before TRUNCATE, not after losing target rows.
let names = columns.iter().map(|column| column.name.clone()).collect::<Vec<_>>();
validate_preexisting_target_columns(target_columns, &names, &DatabaseType::H2, true, target_table)?;
Ok(columns)
}
fn required_unmapped_transfer_target_columns(
target_columns: &[db::ColumnInfo],
col_names: &[String],
@@ -1889,9 +1938,33 @@ fn postgres_column_type_sql(
return rewrite_postgres_schema_qualified_references(trimmed, source_schema, target_schema);
}
}
if matches!(source_db, DatabaseType::H2) {
return map_column_type(&h2_transfer_column_type(column), source_db, target_db);
}
map_column_type(&column.data_type, source_db, target_db)
}
fn h2_transfer_column_type(column: &db::ColumnInfo) -> String {
let data_type = column.data_type.trim();
if !data_type.contains('(') {
match data_type.to_ascii_lowercase().as_str() {
"varchar" | "character varying" | "char" | "character" | "binary varying" | "varbinary" | "binary" => {
if let Some(length) = column.character_maximum_length.filter(|length| *length > 0) {
return format!("{data_type}({length})");
}
}
"decimal" | "numeric" => {
if let Some(precision) = column.numeric_precision.filter(|precision| *precision > 0) {
let scale = column.numeric_scale.unwrap_or(0);
return format!("{data_type}({precision},{scale})");
}
}
_ => {}
}
}
data_type.to_string()
}
fn clickhouse_temporal_column_type(
column: &db::ColumnInfo,
source_db: &DatabaseType,
@@ -2699,6 +2772,13 @@ pub fn escape_value(val: &serde_json::Value, db_type: &DatabaseType) -> String {
}
pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, column_type: Option<&str>) -> String {
if *db_type == DatabaseType::H2
&& !val.is_null()
&& column_type.is_some_and(|kind| kind.trim().eq_ignore_ascii_case("json"))
{
let json = val.as_str().map(str::to_string).unwrap_or_else(|| val.to_string());
return format!("JSON {}", quote_string_literal(&json));
}
match val {
serde_json::Value::Null => "NULL".to_string(),
serde_json::Value::Bool(b) => match db_type {
@@ -2769,12 +2849,21 @@ pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, colum
if let Some(temporal_literal) = format_oracle_temporal_sql_literal(s, db_type, column_type) {
return temporal_literal;
}
if *db_type == DatabaseType::H2 && column_type.is_some_and(is_binary_transfer_column_type) {
if let Some(hex) = s.strip_prefix("0x").or_else(|| s.strip_prefix("0X")) {
if hex.len() % 2 == 0 && hex.bytes().all(|byte| byte.is_ascii_hexdigit()) {
return format!("X'{hex}'");
}
}
}
let literal = format_literal_string(s, db_type, column_type);
if *db_type == DatabaseType::Postgres {
return quote_postgres_string_literal(&literal);
}
let escaped = if is_postgres_family_target(db_type) || *db_type == DatabaseType::SqlServer {
let escaped = if is_postgres_family_target(db_type)
|| matches!(db_type, DatabaseType::SqlServer | DatabaseType::H2)
{
literal.replace('\'', "''")
} else {
literal.replace('\\', "\\\\").replace('\'', "''")
@@ -2800,6 +2889,9 @@ pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, colum
}
_ => {
let s = val.to_string();
if *db_type == DatabaseType::H2 {
return quote_string_literal(&s);
}
format!("'{}'", s.replace('\\', "\\\\").replace('\'', "''"))
}
}
@@ -3287,6 +3379,15 @@ pub fn map_column_type(source_type: &str, source_db: &DatabaseType, target_db: &
}
let t = source_type.to_lowercase();
let mut base = t.split('(').next().unwrap_or(&t).trim();
if matches!(source_db, DatabaseType::H2) {
base = match base {
"character varying" => "varchar",
"character large object" => "clob",
"binary varying" => "varbinary",
"binary large object" => "blob",
_ => base,
};
}
// Extract basic type, `bigint unsigned` -> `bigint`
base = base.split(' ').next().unwrap_or(base).trim();
@@ -3350,7 +3451,7 @@ pub fn map_column_type(source_type: &str, source_db: &DatabaseType, target_db: &
"decimal" | "numeric" | "number" => {
if t.contains('(') {
match target_db {
DatabaseType::Mysql | DatabaseType::SqlServer | DatabaseType::Oracle => {
DatabaseType::Mysql | DatabaseType::SqlServer | DatabaseType::Oracle | DatabaseType::H2 => {
format!("DECIMAL{}", &t[t.find('(').unwrap()..])
}
target_db if is_postgres_transfer_dialect(target_db) => {
@@ -3751,11 +3852,12 @@ impl InsertSqlTemplate {
.map(|column| transfer_column_identifier(column, db_type, quote_target_column_names))
.collect::<Vec<_>>()
.join(", ");
let overriding = if overrides_postgres_system_values && matches!(db_type, DatabaseType::Postgres) {
" OVERRIDING SYSTEM VALUE"
} else {
""
};
let overriding =
if overrides_postgres_system_values && matches!(db_type, DatabaseType::Postgres | DatabaseType::H2) {
" OVERRIDING SYSTEM VALUE"
} else {
""
};
Self {
standard_prefix: format!("INSERT INTO {full_table} ({col_list}){overriding} VALUES\n"),
oracle_into_prefix: matches!(db_type, DatabaseType::Oracle)
@@ -3967,6 +4069,14 @@ fn generate_upsert_typed_for_transfer(
}
sql
}
DatabaseType::H2 => {
let pk_list = pk_columns
.iter()
.map(|column| transfer_column_identifier(column, db_type, quote_target_column_names))
.collect::<Vec<_>>()
.join(", ");
format!("MERGE INTO {full_table} ({col_list}) KEY ({pk_list}) VALUES\n{}", value_rows.join(",\n"))
}
db_type if uses_mysql_style_upsert(db_type) => {
let mut sql = format!("INSERT INTO {full_table} ({col_list}) VALUES\n{}", value_rows.join(",\n"));
if non_pk_columns.is_empty() {
@@ -4254,6 +4364,10 @@ fn rewrite_transfer_source_table_ddl(
Some(rewrite_postgres_schema_qualified_references(sql, source_schema, target_schema))
} else if matches!((source_db_type, target_db_type), (DatabaseType::Dameng, DatabaseType::Dameng)) {
Some(strip_dameng_storage_clauses(&rewrite_double_quoted_schema_qualifier(sql, source_schema, target_schema)))
} else if matches!((source_db_type, target_db_type), (DatabaseType::H2, DatabaseType::H2)) {
let source_schema = if source_schema.trim().is_empty() { "PUBLIC" } else { source_schema };
let target_schema = if target_schema.trim().is_empty() { "PUBLIC" } else { target_schema };
Some(rewrite_h2_schema_qualifier(sql, source_schema, target_schema))
} else if is_mysql_family_target(source_db_type) && is_mysql_family_target(target_db_type) {
// The reused SHOW CREATE TABLE DDL carries the source table name; rewrite the
// CREATE TABLE header when the transfer renames the table (name case conversion).
@@ -6099,6 +6213,16 @@ fn effective_transfer_database_type(config: &ConnectionConfig) -> DatabaseType {
return DatabaseType::Oracle;
}
let is_h2 = config.connection_string.as_deref().is_some_and(|url| url.to_ascii_lowercase().starts_with("jdbc:h2:"))
|| config.jdbc_driver_class.as_deref().is_some_and(|class| class.eq_ignore_ascii_case("org.h2.Driver"))
|| config.driver_profile.as_deref().is_some_and(|profile| {
let profile = profile.to_ascii_lowercase();
profile == "h2" || profile.starts_with("h2-")
});
if is_h2 {
return DatabaseType::H2;
}
let jdbc_identity = [
config.driver_profile.as_deref().unwrap_or(""),
config.connection_string.as_deref().unwrap_or(""),
@@ -9019,8 +9143,8 @@ where
return Err(format!("No writable columns found for table {table}"));
}
let col_names: Vec<String> = writable_columns.iter().map(|c| c.name.clone()).collect();
let col_types: Vec<Option<String>> = writable_columns.iter().map(|c| Some(c.data_type.clone())).collect();
let mut col_names: Vec<String> = writable_columns.iter().map(|c| c.name.clone()).collect();
let mut col_types: Vec<Option<String>> = writable_columns.iter().map(|c| Some(c.data_type.clone())).collect();
let primary_key_columns = transfer_key_columns(&writable_columns, source_db_type);
if should_copy_data(&request.content) {
log::info!("[transfer] {} has {} columns, counting rows...", table, columns.len());
@@ -9148,7 +9272,7 @@ where
| DatabaseType::Impala
| DatabaseType::Argo
))
|| matches!(target_db_type, DatabaseType::Postgres | DatabaseType::Dameng);
|| matches!(target_db_type, DatabaseType::Postgres | DatabaseType::Dameng | DatabaseType::H2);
let target_columns = if needs_target_columns {
get_columns_for_transfer(
state,
@@ -9184,6 +9308,18 @@ where
)?;
}
if *target_db_type == DatabaseType::H2 {
let columns = h2_writable_transfer_columns(&writable_columns, &target_columns, &request.mode, &target_table)?;
col_names = columns.iter().map(|column| column.name.clone()).collect();
col_types = columns
.iter()
.map(|column| {
let target = target_columns.iter().find(|target| target.name == column.name).unwrap_or(column);
Some(target.data_type.clone())
})
.collect();
}
// Truncate target if overwrite mode (only when not rebuilding the table).
// When drop_target_before_create is true, the target table was just created
// and is already empty, so TRUNCATE is unnecessary.
@@ -9229,7 +9365,7 @@ where
let writes_identity_insert_columns = matches!(target_db_type, DatabaseType::Dameng | DatabaseType::SqlServer)
&& selected_columns_include_identity_columns(&col_names, &target_columns);
let overrides_postgres_system_values = matches!(target_db_type, DatabaseType::Postgres)
let overrides_postgres_system_values = matches!(target_db_type, DatabaseType::Postgres | DatabaseType::H2)
&& selected_columns_include_postgres_generated_always_identity_columns(&col_names, &target_columns);
// Transfer data in batches
let batch_size = if request.batch_size == 0 { 1000 } else { request.batch_size };
@@ -10469,6 +10605,190 @@ mod tests {
use serde_json::json;
#[test]
fn h2_jdbc_uses_native_transfer_dialect() {
for config in [
jdbc_transfer_config("jdbc:h2:mem:transfer", "", ""),
jdbc_transfer_config("JDBC:H2:tcp://localhost/test", "", ""),
jdbc_transfer_config("", "org.h2.Driver", ""),
jdbc_transfer_config("", "", "h2-v3"),
] {
assert_eq!(effective_transfer_database_type(&config), DatabaseType::H2);
}
assert_eq!(
effective_transfer_database_type(&jdbc_transfer_config("jdbc:other:test", "", "")),
DatabaseType::Jdbc
);
}
#[test]
fn h2_transfer_preserves_character_and_numeric_metadata() {
let columns = vec![
db::ColumnInfo { character_maximum_length: Some(1024), ..test_column("NAME", "CHARACTER VARYING") },
db::ColumnInfo { numeric_precision: Some(12), numeric_scale: Some(2), ..test_column("AMOUNT", "NUMERIC") },
];
for target in [DatabaseType::Mysql, DatabaseType::Postgres, DatabaseType::H2] {
let ddl = generate_create_table_ddl(
&columns,
"ITEMS",
"PUBLIC",
"target",
&target,
&DatabaseType::H2,
None,
None,
);
assert!(ddl.contains("(1024)"), "{ddl}");
assert!(ddl.contains("(12,2)"), "{ddl}");
assert!(!ddl.contains("CHAR(1)"), "{ddl}");
}
assert_eq!(
map_column_type("CHARACTER VARYING(1024)", &DatabaseType::H2, &DatabaseType::Mysql),
"VARCHAR(1024)"
);
assert_eq!(map_column_type("CHARACTER LARGE OBJECT", &DatabaseType::H2, &DatabaseType::Postgres), "TEXT");
assert_eq!(map_column_type("BINARY LARGE OBJECT", &DatabaseType::H2, &DatabaseType::Postgres), "BYTEA");
assert_eq!(map_column_type("BINARY VARYING(64)", &DatabaseType::H2, &DatabaseType::Mysql), "BLOB");
assert_eq!(map_column_type("DECIMAL(12,2)", &DatabaseType::Mysql, &DatabaseType::H2), "DECIMAL(12,2)");
}
#[test]
fn h2_transfer_rewrites_schema_without_changing_literals_or_identifiers() {
let sql = r#"CREATE TABLE "Source"."ITEMS" ("NAME" VARCHAR(1024) DEFAULT '"Source".ITEMS');
CREATE INDEX "IDX" ON "Source"."ITEMS" ("NAME");
-- "Source".ITEMS stays in the comment
/* "Source".ITEMS */
COMMENT ON TABLE "Source"."ITEMS" IS '"Source".ITEMS';
CREATE TABLE "Other"."prefix""Source"."NAME" ("ID" INT);"#;
let rewritten = rewrite_transfer_source_table_ddl(
sql,
"Source",
"Target",
&DatabaseType::H2,
&DatabaseType::H2,
"ITEMS",
"ITEMS",
)
.unwrap();
assert!(rewritten.contains(r#"CREATE TABLE "Target"."ITEMS""#));
assert!(rewritten.contains(r#"ON "Target"."ITEMS""#));
assert!(rewritten.contains(r#"DEFAULT '"Source".ITEMS'"#));
assert!(rewritten.contains(r#"-- "Source".ITEMS"#));
assert!(rewritten.contains(r#"/* "Source".ITEMS */"#));
assert!(rewritten.contains(r#""Other"."prefix""Source"."NAME""#));
assert_eq!(rewrite_h2_schema_qualifier(r#""A""B" . "T""#, "A\"B", "C\"D"), r#""C""D" . "T""#);
assert!(can_reuse_source_table_ddl(&DatabaseType::H2, &DatabaseType::H2, None, None, true));
assert!(!can_reuse_source_table_ddl(&DatabaseType::H2, &DatabaseType::H2, None, None, false));
}
#[test]
fn h2_upsert_merges_by_composite_primary_key() {
let columns = vec!["ID".into(), "PART".into(), "NAME".into()];
let rows = vec![vec![json!(1), json!(1), json!("o'clock")], vec![json!(1), json!(2), json!(null)]];
assert_eq!(
generate_upsert(&columns, &rows, "ITEMS", "TransferTarget", &DatabaseType::H2, &["ID".into(), "PART".into()]),
"MERGE INTO \"TransferTarget\".\"ITEMS\" (\"ID\", \"PART\", \"NAME\") KEY (\"ID\", \"PART\") VALUES\n(1, 1, 'o''clock'),\n(1, 2, NULL)"
);
assert_eq!(
generate_upsert(&["ID".into()], &[vec![json!(1)]], "ITEMS", "PUBLIC", &DatabaseType::H2, &["ID".into()]),
"MERGE INTO \"PUBLIC\".\"ITEMS\" (\"ID\") KEY (\"ID\") VALUES\n(1)"
);
}
#[test]
fn h2_transfer_skips_target_computed_columns_and_guards_always_identity_upsert() {
let source = vec![test_column("ID", "INTEGER"), test_column("QTY", "INTEGER"), test_column("TOTAL", "INTEGER")];
let mut target = source.clone();
target[0].extra = Some("generated always as identity".into());
target[2].extra = Some("computed".into());
for mode in [TransferMode::Append, TransferMode::Overwrite] {
assert_eq!(
h2_writable_transfer_columns(&source, &target, &mode, "T")
.unwrap()
.iter()
.map(|column| column.name.as_str())
.collect::<Vec<_>>(),
vec!["ID", "QTY"]
);
}
assert!(h2_writable_transfer_columns(&source, &target, &TransferMode::Upsert, "T")
.unwrap_err()
.contains("GENERATED ALWAYS"));
target[0].extra = Some("generated by default as identity".into());
assert!(h2_writable_transfer_columns(&source, &target, &TransferMode::Upsert, "T").is_ok());
// A generated source value can still be copied into an ordinary target column.
assert_eq!(h2_writable_transfer_columns(&target, &source, &TransferMode::Append, "T").unwrap().len(), 3);
assert_eq!(
InsertSqlTemplate::new(&["ID".into()], "T", "PUBLIC", &DatabaseType::H2, None, true)
.build(&["(42)".into()]),
"INSERT INTO \"PUBLIC\".\"T\" (\"ID\") OVERRIDING SYSTEM VALUE VALUES\n(42)"
);
}
#[test]
fn h2_transfer_defaults_empty_schemas_to_public() {
assert_eq!(
rewrite_transfer_source_table_ddl(
r#"CREATE TABLE "PUBLIC"."T" ("ID" INT)"#,
"",
"Target",
&DatabaseType::H2,
&DatabaseType::H2,
"T",
"T"
)
.unwrap(),
r#"CREATE TABLE "Target"."T" ("ID" INT)"#
);
assert_eq!(
rewrite_transfer_source_table_ddl(
r#"CREATE TABLE "Source"."T" ("ID" INT)"#,
"Source",
"",
&DatabaseType::H2,
&DatabaseType::H2,
"T",
"T"
)
.unwrap(),
r#"CREATE TABLE "PUBLIC"."T" ("ID" INT)"#
);
}
#[test]
fn h2_write_preflight_rejects_missing_and_required_target_columns() {
let source = vec![test_column("ID", "INTEGER"), test_column("NAME", "VARCHAR(32)")];
for mode in [TransferMode::Append, TransferMode::Overwrite, TransferMode::Upsert] {
let error = h2_writable_transfer_columns(&source, &source[..1], &mode, "ITEMS").unwrap_err();
assert!(error.contains("NAME"), "{error}");
assert!(error.contains("ITEMS"), "{error}");
let mut target = source.clone();
target.push(db::ColumnInfo { is_nullable: false, ..test_column("REQUIRED", "INTEGER") });
let error = h2_writable_transfer_columns(&source, &target, &mode, "ITEMS").unwrap_err();
assert!(error.contains("REQUIRED"), "{error}");
target[2].column_default = Some("1".into());
assert!(h2_writable_transfer_columns(&source, &target, &mode, "ITEMS").is_ok());
}
}
#[test]
fn h2_transfer_literals_preserve_json_binary_and_backslashes() {
assert_eq!(
escape_value_typed(&json!(r"C:\tmp\o'clock"), &DatabaseType::H2, Some("CHARACTER VARYING")),
r"'C:\tmp\o''clock'"
);
assert_eq!(escape_value_typed(&json!("0x00ff"), &DatabaseType::H2, Some("BINARY VARYING")), "X'00ff'");
assert_eq!(escape_value_typed(&json!("0x"), &DatabaseType::H2, Some("BLOB")), "X''");
assert_eq!(escape_value_typed(&json!("0x00ff"), &DatabaseType::H2, Some("VARCHAR")), "'0x00ff'");
assert_eq!(
escape_value_typed(&json!(r#"{"name":"o'clock"}"#), &DatabaseType::H2, Some("JSON")),
r#"JSON '{"name":"o''clock"}'"#
);
assert_eq!(escape_value_typed(&json!([1, 2]), &DatabaseType::H2, Some("JSON")), "JSON '[1,2]'");
assert_eq!(escape_value_typed(&json!(null), &DatabaseType::H2, Some("JSON")), "NULL");
assert_eq!(escape_value_typed(&json!("null"), &DatabaseType::H2, Some("JSON")), "JSON 'null'");
}
fn test_column(name: &str, data_type: &str) -> db::ColumnInfo {
db::ColumnInfo {
name: name.to_string(),
+26 -4
View File
@@ -145,10 +145,12 @@ pub(super) async fn prepare_table_ddl(
.await
{
Ok(ddl) => (ddl, true),
Err(e) if rebuild => {
// A rebuild cannot fall back to generated DDL when the source DDL
// it promised to reproduce cannot be read.
return Err(format!("Failed to read source DDL for table '{table}' before rebuilding: {e}"));
Err(e)
if rebuild || matches!((source_db_type, target_db_type), (DatabaseType::H2, DatabaseType::H2)) =>
{
// Rebuilds and H2-to-H2 transfers must not silently discard source
// constraints or generated columns when native DDL cannot be read.
return Err(format!("Failed to read source DDL for table '{table}' before creating target: {e}"));
}
Err(_) => (
generate_create_table_ddl_with_column_quoting(
@@ -375,6 +377,26 @@ mod tests {
assert!(prepared.ddl.contains("CREATE TABLE"));
}
#[tokio::test]
async fn h2_native_ddl_read_failure_does_not_fall_back_to_lossy_metadata() {
let (state, _directory) = state_fixture().await;
let error = prepare_table_ddl(
&state,
&request(false),
"child",
"child",
&DatabaseType::H2,
&DatabaseType::H2,
"missing-source-pool",
&columns(),
None,
&HashMap::new(),
)
.await
.unwrap_err();
assert!(error.contains("Failed to read source DDL"), "{error}");
}
#[tokio::test]
async fn deferred_foreign_key_uses_actual_target_and_case_converted_parent() {
let (state, _directory) = state_fixture().await;