fix(oceanbase): use rowid for keyless Oracle rows

crates/dbx-core/src/data_grid_sql.rs

@@ -16,8 +16,8 @@ use data_grid_tdengine_sql::{
 
 use crate::models::connection::DatabaseType;
 use crate::sql_dialect::{
-    firebird_rows_clause, quote_table_identifier, table_pagination_strategy, uses_single_row_insert_statements,
-    TablePaginationStrategy,
+    firebird_rows_clause, quote_table_identifier, table_pagination_strategy, uses_oracle_row_id,
+    uses_single_row_insert_statements, TablePaginationStrategy,
 };
 use crate::transfer::{format_ch_array_sql_literal, format_pg_array_sql_literal};
 
@@ -828,6 +828,9 @@ fn validate_data_grid_save(options: &DataGridSaveStatementOptions) -> Option<Str
     if let Some(error) = validate_tdengine_existing_rows(options) {
         return Some(error);
     }
+    if let Some(error) = validate_oracle_keyless_lob_predicate(options) {
+        return Some(error);
+    }
 
     let not_null_columns: Vec<String> = options
         .table_meta
@@ -878,6 +881,24 @@ fn validate_data_grid_save(options: &DataGridSaveStatementOptions) -> Option<Str
     None
 }
 
+fn validate_oracle_keyless_lob_predicate(options: &DataGridSaveStatementOptions) -> Option<String> {
+    if !uses_oracle_row_id(options.database_type)
+        || !options.table_meta.primary_keys.is_empty()
+        || (options.dirty_rows.is_empty() && options.deleted_rows.is_empty())
+    {
+        return None;
+    }
+    let has_lob_column =
+        options.table_meta.columns.as_deref().unwrap_or(&[]).iter().any(|column| is_oracle_lob_type(&column.data_type));
+    if !has_lob_column {
+        return None;
+    }
+
+    // LOB equality is unsupported in Oracle-compatible SQL. Refuse unsafe
+    // keyless writes instead of dropping LOB predicates and risking extra rows.
+    Some("Cannot safely update or delete this Oracle-compatible row because the table has LOB columns but no primary key or ROWID identifier.".to_string())
+}
+
 fn validate_clickhouse_mutable_updates(options: &DataGridSaveStatementOptions) -> Option<String> {
     if options.database_type != Some(DatabaseType::ClickHouse) || options.dirty_rows.is_empty() {
         return None;
@@ -1961,8 +1982,16 @@ fn is_textual_column_type(data_type: &str) -> bool {
         || lower.starts_with("national character varying")
 }
 
+fn is_oracle_lob_type(data_type: &str) -> bool {
+    let lower = data_type.trim().trim_matches('"').to_ascii_lowercase();
+    let base = lower.split(['(', ':', ' ']).next().unwrap_or("");
+    matches!(base, "blob" | "clob" | "nclob" | "bfile" | "lob")
+        || lower.starts_with("binary large object")
+        || lower.starts_with("character large object")
+}
+
 fn is_oracle_row_id(database_type: Option<DatabaseType>, name: Option<&str>) -> bool {
-    database_type == Some(DatabaseType::Oracle) && name.is_some_and(|name| name.eq_ignore_ascii_case(DBX_ROWID_COLUMN))
+    uses_oracle_row_id(database_type) && name.is_some_and(|name| name.eq_ignore_ascii_case(DBX_ROWID_COLUMN))
 }
 
 pub(crate) fn is_neo4j_element_id(database_type: Option<DatabaseType>, name: Option<&str>) -> bool {
@@ -3370,6 +3399,141 @@ mod tests {
         );
     }
 
+    #[test]
+    fn prepares_oceanbase_oracle_lob_deletes_with_synthetic_rowid() {
+        let result = prepare_data_grid_save(DataGridSaveStatementOptions {
+            database_type: Some(DatabaseType::OceanbaseOracle),
+            table_meta: DataGridTableMeta {
+                catalog: None,
+                database: None,
+                schema: Some("APP".to_string()),
+                table_name: "DATA_REPORT_SUB_TASK".to_string(),
+                primary_keys: vec![DBX_ROWID_COLUMN.to_string()],
+                columns: Some(vec![
+                    column(DBX_ROWID_COLUMN, "VARCHAR2", false, None),
+                    column("ID", "VARCHAR2(100)", false, None),
+                    column("SMC_RESPONSE", "CLOB", true, None),
+                    column("RAW_PAYLOAD", "BLOB", true, None),
+                    column("ARCHIVE_VALUE", "LOB", true, None),
+                ]),
+            },
+            columns: vec![
+                DBX_ROWID_COLUMN.to_string(),
+                "ID".to_string(),
+                "SMC_RESPONSE".to_string(),
+                "RAW_PAYLOAD".to_string(),
+                "ARCHIVE_VALUE".to_string(),
+            ],
+            source_columns: None,
+            rows: vec![
+                vec![json!("*AAABk1AAEAAAAAgAAA"), json!("task-1"), json!("response"), json!("0011"), json!("archive")],
+                vec![json!("*AAABk1AAEAAAAAgAAB"), json!("task-2"), Value::Null, Value::Null, Value::Null],
+            ],
+            dirty_rows: vec![],
+            deleted_rows: vec![0, 1],
+            new_rows: vec![],
+        });
+
+        assert_eq!(result.validation_error, None);
+        assert_eq!(
+            result.statements,
+            vec![
+                "DELETE FROM \"APP\".\"DATA_REPORT_SUB_TASK\" WHERE ROWIDTOCHAR(ROWID) = '*AAABk1AAEAAAAAgAAA';",
+                "DELETE FROM \"APP\".\"DATA_REPORT_SUB_TASK\" WHERE ROWIDTOCHAR(ROWID) = '*AAABk1AAEAAAAAgAAB';",
+            ]
+        );
+    }
+
+    #[test]
+    fn prepares_oceanbase_oracle_lob_delete_with_declared_primary_key() {
+        let result = prepare_data_grid_save(DataGridSaveStatementOptions {
+            database_type: Some(DatabaseType::OceanbaseOracle),
+            table_meta: DataGridTableMeta {
+                catalog: None,
+                database: None,
+                schema: Some("APP".to_string()),
+                table_name: "DOCUMENTS".to_string(),
+                primary_keys: vec!["ID".to_string()],
+                columns: Some(vec![
+                    column("ID", "NUMBER", false, None),
+                    column("TITLE", "VARCHAR2(100)", false, None),
+                    column("BODY", "CLOB", true, None),
+                    column("CONTENT", "BLOB", true, None),
+                ]),
+            },
+            columns: vec!["ID".to_string(), "TITLE".to_string(), "BODY".to_string(), "CONTENT".to_string()],
+            source_columns: None,
+            rows: vec![vec![json!(42), json!("report"), json!("body"), Value::Null]],
+            dirty_rows: vec![],
+            deleted_rows: vec![0],
+            new_rows: vec![],
+        });
+
+        assert_eq!(result.validation_error, None);
+        assert_eq!(result.statements, vec!["DELETE FROM \"APP\".\"DOCUMENTS\" WHERE \"ID\" = 42;"]);
+    }
+
+    #[test]
+    fn rejects_oceanbase_oracle_keyless_lob_writes_without_rowid() {
+        let result = prepare_data_grid_save(DataGridSaveStatementOptions {
+            database_type: Some(DatabaseType::OceanbaseOracle),
+            table_meta: DataGridTableMeta {
+                catalog: None,
+                database: None,
+                schema: Some("APP".to_string()),
+                table_name: "DOCUMENTS".to_string(),
+                primary_keys: vec![],
+                columns: Some(vec![column("TITLE", "VARCHAR2(100)", false, None), column("BODY", "CLOB", true, None)]),
+            },
+            columns: vec!["TITLE".to_string(), "BODY".to_string()],
+            source_columns: None,
+            rows: vec![vec![json!("duplicate title"), json!("unique body")]],
+            dirty_rows: vec![],
+            deleted_rows: vec![0],
+            new_rows: vec![],
+        });
+
+        assert_eq!(
+            result.validation_error.as_deref(),
+            Some("Cannot safely update or delete this Oracle-compatible row because the table has LOB columns but no primary key or ROWID identifier.")
+        );
+        assert!(result.statements.is_empty());
+        assert!(result.rollback_statements.is_empty());
+    }
+
+    #[test]
+    fn preserves_oceanbase_oracle_keyless_predicates_for_comparable_columns() {
+        let result = prepare_data_grid_save(DataGridSaveStatementOptions {
+            database_type: Some(DatabaseType::OceanbaseOracle),
+            table_meta: DataGridTableMeta {
+                catalog: None,
+                database: None,
+                schema: Some("APP".to_string()),
+                table_name: "TASK_STATUS".to_string(),
+                primary_keys: vec![],
+                columns: Some(vec![
+                    column("TASK_NAME", "VARCHAR2(100)", false, None),
+                    column("STATUS", "VARCHAR2(16)", true, None),
+                ]),
+            },
+            columns: vec!["TASK_NAME".to_string(), "STATUS".to_string()],
+            source_columns: None,
+            rows: vec![vec![json!("task-1"), json!("RUNNING")], vec![json!("task-2"), Value::Null]],
+            dirty_rows: vec![],
+            deleted_rows: vec![0, 1],
+            new_rows: vec![],
+        });
+
+        assert_eq!(result.validation_error, None);
+        assert_eq!(
+            result.statements,
+            vec![
+                "DELETE FROM \"APP\".\"TASK_STATUS\" WHERE \"TASK_NAME\" = 'task-1' AND \"STATUS\" = 'RUNNING';",
+                "DELETE FROM \"APP\".\"TASK_STATUS\" WHERE \"TASK_NAME\" = 'task-2' AND \"STATUS\" IS NULL;",
+            ]
+        );
+    }
+
     #[test]
     fn formats_mysql_bit_literals_without_string_quotes() {
         let bit = column("flag", "bit(1)", true, None);

crates/dbx-core/src/database_export.rs

@@ -697,9 +697,10 @@ pub fn build_export_insert_statements(options: BuildExportInsertStatementsOption
 }
 
 pub(crate) fn is_internal_export_column(database_type: Option<DatabaseType>, column: &str) -> bool {
-    // Oracle ROWID is injected only to identify editable rows. It is not a
-    // physical table column and must never propagate into exported SQL.
-    database_type == Some(DatabaseType::Oracle) && column.eq_ignore_ascii_case(crate::sql_dialect::DBX_ROWID_COLUMN)
+    // Oracle-compatible ROWID is injected only to identify editable rows. It
+    // is not a physical table column and must never propagate into exports.
+    crate::sql_dialect::uses_oracle_row_id(database_type)
+        && column.eq_ignore_ascii_case(crate::sql_dialect::DBX_ROWID_COLUMN)
 }
 
 fn is_postgres_tsvector_export_column(database_type: Option<DatabaseType>, column_type: Option<&str>) -> bool {
@@ -1758,6 +1759,24 @@ mod tests {
         assert_eq!(statements, vec!["INSERT INTO \"APP\".\"USERS\" (\"ID\", \"NAME\") VALUES (1, 'Ada');"]);
     }
 
+    #[test]
+    fn oceanbase_oracle_export_omits_synthetic_rowid_from_insert_columns() {
+        let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
+            database_type: Some(DatabaseType::OceanbaseOracle),
+            schema: Some("APP".to_string()),
+            table_name: Some("USERS".to_string()),
+            qualified_table_name: None,
+            columns: vec!["__DBX_ROWID".to_string(), "ID".to_string(), "NAME".to_string()],
+            column_types: vec![Some("VARCHAR2".to_string()), Some("NUMBER".to_string()), Some("VARCHAR2".to_string())],
+            column_extras: Vec::new(),
+            rows: vec![vec![json!("*AAABk1AAEAAAAAgAAA"), json!(1), json!("Ada")]],
+            batch_size: Some(100),
+        })
+        .unwrap();
+
+        assert_eq!(statements, vec!["INSERT INTO \"APP\".\"USERS\" (\"ID\", \"NAME\") VALUES (1, 'Ada');"]);
+    }
+
     #[test]
     fn non_oracle_export_preserves_dbx_rowid_named_column() {
         let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {

crates/dbx-core/src/sql_dialect.rs

@@ -8,7 +8,7 @@ mod tests;
 
 pub use capabilities::{
     firebird_rows_clause, is_schema_aware, pagination_strategy, table_pagination_strategy, uses_fetch_first,
-    uses_single_row_insert_statements, PaginationContext, TablePaginationStrategy,
+    uses_oracle_row_id, uses_single_row_insert_statements, PaginationContext, TablePaginationStrategy,
 };
 pub use identifiers::{
     normalize_where_input, qualified_table_name, qualified_table_name_with_catalog, quote_table_identifier,

crates/dbx-core/src/sql_dialect/capabilities.rs

@@ -65,6 +65,10 @@ pub fn uses_fetch_first(database_type: DatabaseType) -> bool {
     matches!(database_type, DatabaseType::Oracle | DatabaseType::Dameng | DatabaseType::Db2)
 }
 
+pub fn uses_oracle_row_id(database_type: Option<DatabaseType>) -> bool {
+    matches!(database_type, Some(DatabaseType::Oracle | DatabaseType::OceanbaseOracle))
+}
+
 /// Oracle 系方言不支持 `INSERT ... VALUES (...), (...)` 多行语法，
 /// 复制为 INSERT 与导出 INSERT 都需按行生成单条语句。
 pub fn uses_single_row_insert_statements(database_type: DatabaseType) -> bool {

crates/dbx-core/src/sql_dialect/table_select.rs

@@ -1,6 +1,8 @@
 use crate::models::connection::DatabaseType;
 
-use super::capabilities::{firebird_rows_clause, table_pagination_strategy, uses_fetch_first, TablePaginationStrategy};
+use super::capabilities::{
+    firebird_rows_clause, table_pagination_strategy, uses_oracle_row_id, TablePaginationStrategy,
+};
 use super::identifiers::{
     normalize_where_input, qualified_table_name, qualified_table_name_with_catalog, quote_table_identifier,
 };
@@ -50,7 +52,7 @@ pub fn build_table_data_select_sql(options: TableDataSelectSqlOptions) -> String
     // Oracle join views can raise ORA-01445 when ROWID is selected; keep the
     // synthetic ROWID fallback scoped to base-table reads.
     let include_oracle_row_id = options.include_row_id
-        && database_type == Some(DatabaseType::Oracle)
+        && uses_oracle_row_id(database_type)
         && !is_view_table_type(options.table_type.as_deref());
 
     let select_columns = if include_oracle_row_id {
@@ -72,8 +74,7 @@ pub fn build_table_data_select_sql(options: TableDataSelectSqlOptions) -> String
     } else {
         rownum_select_columns.clone()
     };
-    let table_alias =
-        if include_oracle_row_id && database_type.is_some_and(uses_fetch_first) { format!("{table} t") } else { table };
+    let table_alias = if include_oracle_row_id { format!("{table} t") } else { table };
 
     match table_pagination_strategy(database_type) {
         TablePaginationStrategy::IrisTop => {

crates/dbx-core/src/sql_dialect/tests.rs

@@ -784,6 +784,24 @@ fn builds_oracle_and_neo4j_table_data_queries() {
         }),
         "SELECT \"__DBX_ROWID\", \"ID\", \"NAME\" FROM (SELECT ROWIDTOCHAR(t.ROWID) AS \"__DBX_ROWID\", t.* FROM \"DBXTEST\".\"DBX_LOAD_TABLE_006\" t) WHERE ROWNUM <= 100"
     );
+    assert_eq!(
+        build_table_data_select_sql(TableDataSelectSqlOptions {
+            database_type: Some(DatabaseType::OceanbaseOracle),
+            schema: Some("APP".to_string()),
+            table_name: "DATA_REPORT_SUB_TASK".to_string(),
+            table_type: Some("TABLE".to_string()),
+            primary_keys: vec![DBX_ROWID_COLUMN.to_string()],
+            columns: vec!["ID".to_string(), "SMC_RESPONSE".to_string()],
+            fallback_order_columns: Vec::new(),
+            order_by: None,
+            limit: Some(100),
+            offset: None,
+            where_input: None,
+            include_row_id: true,
+            ..Default::default()
+        }),
+        "SELECT \"__DBX_ROWID\", \"ID\", \"SMC_RESPONSE\" FROM (SELECT ROWIDTOCHAR(t.ROWID) AS \"__DBX_ROWID\", t.* FROM \"APP\".\"DATA_REPORT_SUB_TASK\" t) WHERE ROWNUM <= 100"
+    );
     assert_eq!(
         build_table_data_select_sql(TableDataSelectSqlOptions {
             database_type: Some(DatabaseType::Oracle),