Add Oracle UPSERT

database-commons/src/main/java/io/cdap/plugin/db/DBRecord.java

@@ -314,10 +314,7 @@ protected void upsertOperation(PreparedStatement stmt) throws SQLException {
   }
 
   private boolean fillUpdateParams(List<String> updatedKeyList, ColumnType columnType) {
-    if (operationName.equals(Operation.UPDATE) && updatedKeyList.contains(columnType.getName())) {
-      return true;
-    }
-    return false;
+    return operationName.equals(Operation.UPDATE) && updatedKeyList.contains(columnType.getName());
   }
 
   private Schema getNonNullableSchema(Schema.Field field) {

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleETLDBOutputFormat.java

@@ -0,0 +1,122 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.oracle;
+
+import io.cdap.plugin.db.sink.ETLDBOutputFormat;
+
+/**
+ * Class that extends {@link ETLDBOutputFormat} to implement the abstract methods
+ */
+public class OracleETLDBOutputFormat extends ETLDBOutputFormat {
+
+  /**
+   * This method is used to construct the upsert query for Oracle using MERGE statement.
+   * Example - MERGE INTO my_table target
+   * USING (SELECT ? AS id, ? AS name, ? AS age FROM dual) source
+   * ON (target.id = source.id)
+   * WHEN MATCHED THEN UPDATE SET target.name = source.name, target.age = source.age
+   * WHEN NOT MATCHED THEN INSERT (id, name, age) VALUES (source.id, source.name, source.age)
+   * @param table - Name of the table
+   * @param fieldNames - All the columns of the table
+   * @param listKeys - The columns used as keys for matching
+   * @return Upsert query in the form of string
+   */
+  @Override
+  public String constructUpsertQuery(String table, String[] fieldNames, String[] listKeys) {
+    if (listKeys == null) {
+      throw new IllegalArgumentException("Column names to be updated should not be null");
+    } else if (fieldNames == null) {
+      throw new IllegalArgumentException("Field names should not be null");
+    } else {
+      StringBuilder query = new StringBuilder();
+
+      // MERGE INTO target_table target
+      query.append("MERGE INTO ").append(table).append(" target ");
+
+      // USING (SELECT ? AS col1, ? AS col2, ... FROM dual) source
+      query.append("USING (SELECT ");
+      for (int i = 0; i < fieldNames.length; ++i) {
+        query.append("? AS ").append(fieldNames[i]);
+        if (i != fieldNames.length - 1) {
+          query.append(", ");
+        }
+      }
+      query.append(" FROM dual) source ");
+
+      // ON (target.key1 = source.key1 AND target.key2 = source.key2 ...)
+      query.append("ON (");
+      for (int i = 0; i < listKeys.length; ++i) {
+        query.append("target.").append(listKeys[i]).append(" = source.").append(listKeys[i]);
+        if (i != listKeys.length - 1) {
+          query.append(" AND ");
+        }
+      }
+      query.append(") ");
+
+      // WHEN MATCHED THEN UPDATE SET target.col1 = source.col1, target.col2 = source.col2 ...
+      // Only update non-key columns
+      query.append("WHEN MATCHED THEN UPDATE SET ");
+      boolean firstUpdateColumn = true;
+      for (String fieldName : fieldNames) {
+        boolean isKeyColumn = false;
+        for (String listKey : listKeys) {
+          String listKeyNoQuote = listKey.replace("\"", "");
+          if (listKeyNoQuote.equals(fieldName)) {
+            isKeyColumn = true;
+            break;
+          }
+        }
+        if (!isKeyColumn) {
+          if (!firstUpdateColumn) {
+            query.append(", ");
+          }
+          query.append("target.").append(fieldName).append(" = source.").append(fieldName);
+          firstUpdateColumn = false;
+        }
+      }
+
+      // WHEN NOT MATCHED THEN INSERT (col1, col2, ...) VALUES (source.col1, source.col2, ...)
+      query.append(" WHEN NOT MATCHED THEN INSERT (");
+      for (int i = 0; i < fieldNames.length; ++i) {
+        query.append(fieldNames[i]);
+        if (i != fieldNames.length - 1) {
+          query.append(", ");
+        }
+      }
+      query.append(") VALUES (");
+      for (int i = 0; i < fieldNames.length; ++i) {
+        query.append("source.").append(fieldNames[i]);
+        if (i != fieldNames.length - 1) {
+          query.append(", ");
+        }
+      }
+      query.append(")");
+
+      return query.toString();
+    }
+  }
+
+  @Override
+  public String constructUpdateQuery(String table, String[] fieldNames, String[] listKeys) {
+    // Oracle JDBC does not accept a trailing semicolon in prepared statements.
+    String query = super.constructUpdateQuery(table, fieldNames, listKeys);
+    if (query.endsWith(";")) {
+      return query.substring(0, query.length() - 1);
+    }
+    return query;
+  }
+}

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSink.java

@@ -23,6 +23,7 @@
 import io.cdap.cdap.api.annotation.MetadataProperty;
 import io.cdap.cdap.api.annotation.Name;
 import io.cdap.cdap.api.annotation.Plugin;
+import io.cdap.cdap.api.data.batch.Output;
 import io.cdap.cdap.api.data.format.StructuredRecord;
 import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.batch.BatchSink;
@@ -31,6 +32,7 @@
 import io.cdap.plugin.common.Asset;
 import io.cdap.plugin.common.ConfigUtil;
 import io.cdap.plugin.common.LineageRecorder;
+import io.cdap.plugin.common.batch.sink.SinkOutputFormatProvider;
 import io.cdap.plugin.common.db.DBErrorDetailsProvider;
 import io.cdap.plugin.db.DBRecord;
 import io.cdap.plugin.db.SchemaReader;
@@ -60,7 +62,8 @@ public OracleSink(OracleSinkConfig oracleSinkConfig) {
 
   @Override
   protected DBRecord getDBRecord(StructuredRecord output) {
-    return new OracleSinkDBRecord(output, columnTypes);
+    return new OracleSinkDBRecord(output, columnTypes, oracleSinkConfig.getOperationName(),
+        oracleSinkConfig.getRelationTableKey());
   }
 
   @Override
@@ -72,6 +75,13 @@ protected FieldsValidator getFieldsValidator() {
   protected SchemaReader getSchemaReader() {
     return new OracleSinkSchemaReader();
   }
+
+  @Override
+  protected void addOutputContext(BatchSinkContext context) {
+    context.addOutput(Output.of(oracleSinkConfig.getReferenceName(),
+        new SinkOutputFormatProvider(OracleETLDBOutputFormat.class, getConfiguration())));
+  }
+
   @Override
   protected LineageRecorder getLineageRecorder(BatchSinkContext context) {
     String fqn = DBUtils.constructFQN("oracle",

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSinkDBRecord.java

@@ -19,6 +19,7 @@
 import io.cdap.cdap.api.data.format.StructuredRecord;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.plugin.db.ColumnType;
+import io.cdap.plugin.db.Operation;
 import io.cdap.plugin.db.SchemaReader;
 
 import java.sql.PreparedStatement;
@@ -31,9 +32,12 @@
  */
 public class OracleSinkDBRecord extends OracleSourceDBRecord {
 
-  public OracleSinkDBRecord(StructuredRecord record, List<ColumnType> columnTypes) {
+  public OracleSinkDBRecord(StructuredRecord record, List<ColumnType> columnTypes, Operation operationName,
+      String relationTableKey) {
     this.record = record;
     this.columnTypes = columnTypes;
+    this.operationName = operationName;
+    this.relationTableKey = relationTableKey;
   }
 
   @Override

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceDBRecord.java

@@ -116,8 +116,15 @@ protected void handleField(ResultSet resultSet, StructuredRecord.Builder recordB
   @Override
   protected void writeNonNullToDB(PreparedStatement stmt, Schema fieldSchema,
                                   String fieldName, int fieldIndex) throws SQLException {
-    int sqlType = columnTypes.get(fieldIndex).getType();
     int sqlIndex = fieldIndex + 1;
+    int sqlType = Types.OTHER;
+    // avoid OOB exception in case of mismatch between columnTypes and record schema fields
+    for (ColumnType columnType : columnTypes) {
+      if (columnType.getName().equals(fieldName)) {
+        sqlType = columnType.getType();
+        break;
+      }
+    }
 
     // TIMESTAMP and TIMESTAMPTZ types needs to be handled using the specific oracle types to ensure that the data
     // inserted matches with the provided value. As Oracle driver internally alters the values provided

oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleETLDBOutputFormatTest.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.oracle;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class OracleETLDBOutputFormatTest {
+
+  private final OracleETLDBOutputFormat outputFormat = new OracleETLDBOutputFormat();
+
+  @Test
+  public void testConstructUpsertQueryBasic() {
+    String[] fieldNames = {"id", "name", "age"};
+    String[] listKeys = {"id"};
+    String table = "my_table";
+
+    String result = outputFormat.constructUpsertQuery(table, fieldNames, listKeys);
+
+    String expected = "MERGE INTO my_table target " +
+        "USING (SELECT ? AS id, ? AS name, ? AS age FROM dual) source " +
+        "ON (target.id = source.id) " +
+        "WHEN MATCHED THEN UPDATE SET target.name = source.name, target.age = source.age " +
+        "WHEN NOT MATCHED THEN INSERT (id, name, age) VALUES (source.id, source.name, source.age)";
+
+    Assert.assertEquals(expected, result);
+  }
+
+  @Test
+  public void testConstructUpsertQueryMultipleKeys() {
+    String[] fieldNames = {"id", "code", "name", "value"};
+    String[] listKeys = {"id", "code"};
+    String table = "composite_key_table";
+
+    String result = outputFormat.constructUpsertQuery(table, fieldNames, listKeys);
+
+    String expected = "MERGE INTO composite_key_table target "
+        + "USING (SELECT ? AS id, ? AS code, ? AS name, ? AS value FROM dual) source "
+        + "ON (target.id = source.id AND target.code = source.code) "
+        + "WHEN MATCHED THEN UPDATE SET target.name = source.name, target.value = source.value "
+        + "WHEN NOT MATCHED THEN INSERT (id, code, name, value) VALUES (source.id, source.code, source.name, source"
+        + ".value)";
+
+    Assert.assertEquals(expected, result);
+  }
+
+  @Test
+  public void testConstructUpsertQuerySingleField() {
+    String[] fieldNames = {"id", "name"};
+    String[] listKeys = {"id"};
+    String table = "single_field_update_table";
+
+    String result = outputFormat.constructUpsertQuery(table, fieldNames, listKeys);
+
+    String expected = "MERGE INTO single_field_update_table target " +
+        "USING (SELECT ? AS id, ? AS name FROM dual) source " +
+        "ON (target.id = source.id) " +
+        "WHEN MATCHED THEN UPDATE SET target.name = source.name " +
+        "WHEN NOT MATCHED THEN INSERT (id, name) VALUES (source.id, source.name)";
+
+    Assert.assertEquals(expected, result);
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testConstructUpsertQueryNullListKeys() {
+    String[] fieldNames = {"id", "name", "age"};
+    String table = "my_table";
+
+    outputFormat.constructUpsertQuery(table, fieldNames, null);
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testConstructUpsertQueryNullFieldNames() {
+    String[] listKeys = {"id"};
+    String table = "my_table";
+
+    outputFormat.constructUpsertQuery(table, null, listKeys);
+  }
+
+  @Test
+  public void testConstructUpsertQueryAllFieldsAreKeys() {
+    String[] fieldNames = {"id", "code"};
+    String[] listKeys = {"id", "code"};
+    String table = "all_keys_table";
+
+    String result = outputFormat.constructUpsertQuery(table, fieldNames, listKeys);
+
+    // When all fields are keys, the UPDATE SET clause will be empty after "SET "
+    // Note: There's an extra space before "WHEN NOT MATCHED" due to implementation
+    String expected = "MERGE INTO all_keys_table target " +
+        "USING (SELECT ? AS id, ? AS code FROM dual) source " +
+        "ON (target.id = source.id AND target.code = source.code) " +
+        "WHEN MATCHED THEN UPDATE SET  " +
+        "WHEN NOT MATCHED THEN INSERT (id, code) VALUES (source.id, source.code)";
+
+    Assert.assertEquals(expected, result);
+  }
+
+  @Test
+  public void testConstructUpsertQueryWithSpecialTableName() {
+    String[] fieldNames = {"id", "name"};
+    String[] listKeys = {"id"};
+    String table = "SCHEMA.MY_TABLE";
+
+    String result = outputFormat.constructUpsertQuery(table, fieldNames, listKeys);
+
+    String expected = "MERGE INTO SCHEMA.MY_TABLE target " +
+        "USING (SELECT ? AS id, ? AS name FROM dual) source " +
+        "ON (target.id = source.id) " +
+        "WHEN MATCHED THEN UPDATE SET target.name = source.name " +
+        "WHEN NOT MATCHED THEN INSERT (id, name) VALUES (source.id, source.name)";
+
+    Assert.assertEquals(expected, result);
+  }
+}

oracle-plugin/widgets/Oracle-batchsink.json

@@ -192,6 +192,29 @@
           "label": "Schema Name",
           "name": "dbSchemaName"
         },
+        {
+          "widget-type": "radio-group",
+          "label": "Operation Name",
+          "name": "operationName",
+          "widget-attributes": {
+            "default": "insert",
+            "layout": "inline",
+            "options": [
+              {
+                "id": "insert",
+                "label": "INSERT"
+              },
+              {
+                "id": "update",
+                "label": "UPDATE"
+              },
+              {
+                "id": "upsert",
+                "label": "UPSERT"
+              }
+            ]
+          }
+        },
         {
           "widget-type": "hidden",
           "label": "Operation Name",
@@ -201,9 +224,10 @@
           }
         },
         {
-          "widget-type": "hidden",
+          "name": "relationTableKey",
+          "widget-type": "csv",
           "label": "Table Key",
-          "name": "relationTableKey"
+          "widget-attributes": {}
         }
       ]
     },