diff --git a/changes/en-us/develop.md b/changes/en-us/develop.md
index da1c53be175..6780d3be941 100644
--- a/changes/en-us/develop.md
+++ b/changes/en-us/develop.md
@@ -15,8 +15,10 @@ Add changes here for all PR submitted to the develop branch.
 - [[#5245](https://github.com/seata/seata/pull/5245)] fix the incomplete dependency of distribution module
 - [[#5239](https://github.com/seata/seata/pull/5239)] fix `getConfig` throw `ClassCastException` when use JDK proxy
 - [[#5281](https://github.com/seata/seata/pull/5281)] parallel request handle throw IndexOutOfBoundsException
+- [[#5288](https://github.com/seata/seata/pull/5288)] fix auto-increment of pk columns in Oracle in AT mode
 - [[#5287](https://github.com/seata/seata/pull/5287)] fix auto-increment of pk columns in PostgreSQL in AT mode
 
+
 ### optimize:
 - [[#5208](https://github.com/seata/seata/pull/5208)] optimize throwable getCause once more
 - [[#5212](https://github.com/seata/seata/pull/5212)] optimize log message level
diff --git a/changes/zh-cn/develop.md b/changes/zh-cn/develop.md
index 238f6492182..4091ea8e7ab 100644
--- a/changes/zh-cn/develop.md
+++ b/changes/zh-cn/develop.md
@@ -15,8 +15,10 @@
 - [[#5245](https://github.com/seata/seata/pull/5245)] 修复不完整的distribution模块依赖
 - [[#5239](https://github.com/seata/seata/pull/5239)] 修复当使用JDK代理时，`getConfig` 方法获取部分配置时抛出 `ClassCastException` 异常的问题
 - [[#5281](https://github.com/seata/seata/pull/5281)] 修复并行rm请求处理时数组索引越界问题
+- [[#5288](https://github.com/seata/seata/pull/5288)] 修复AT模式下oracle的主键列自增的问题
 - [[#5287](https://github.com/seata/seata/pull/5287)] 修复AT模式下pgsql的主键列自增的问题
 
+
 ### optimize:
 - [[#5208](https://github.com/seata/seata/pull/5208)] 优化多次重复获取Throwable#getCause问题
 - [[#5212](https://github.com/seata/seata/pull/5212)] 优化不合理的日志信息级别
diff --git a/rm-datasource/src/main/java/io/seata/rm/datasource/exec/oracle/OracleInsertExecutor.java b/rm-datasource/src/main/java/io/seata/rm/datasource/exec/oracle/OracleInsertExecutor.java
index e4220026c2e..deec875345e 100644
--- a/rm-datasource/src/main/java/io/seata/rm/datasource/exec/oracle/OracleInsertExecutor.java
+++ b/rm-datasource/src/main/java/io/seata/rm/datasource/exec/oracle/OracleInsertExecutor.java
@@ -17,19 +17,23 @@
 
 import io.seata.common.loader.LoadLevel;
 import io.seata.common.loader.Scope;
+import io.seata.common.util.CollectionUtils;
 import io.seata.rm.datasource.StatementProxy;
 import io.seata.rm.datasource.exec.BaseInsertExecutor;
 import io.seata.rm.datasource.exec.StatementCallback;
+import io.seata.sqlparser.SQLInsertRecognizer;
 import io.seata.sqlparser.SQLRecognizer;
 import io.seata.sqlparser.struct.Null;
 import io.seata.sqlparser.struct.Sequenceable;
 import io.seata.sqlparser.struct.SqlMethodExpr;
 import io.seata.sqlparser.struct.SqlSequenceExpr;
+import io.seata.sqlparser.util.ColumnUtils;
 import io.seata.sqlparser.util.JdbcConstants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.sql.SQLException;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -56,9 +60,55 @@ public OracleInsertExecutor(StatementProxy statementProxy, StatementCallback sta
         super(statementProxy, statementCallback, sqlRecognizer);
     }
 
+    /**
+     * 1. If the insert columns are not empty and do not contain any pk columns,
+     * it means that there is no pk value in the insert rows, then all the pk values should come from auto-increment.
+     * <p>
+     * 2. The pk value exists in insert rows. The possible situations are:
+     * <ul>
+     *     <li>The insert columns are empty: all pk values can be obtained from insert rows</li>
+     *     <li>The insert columns contain at least one pk column: first obtain the existing pk value from the insert rows, and other from auto-increment</li>
+     * </ul>
+     *
+     * @return {@link Map}<{@link String}, {@link List}<{@link Object}>>
+     * @throws SQLException the sql exception
+     */
     @Override
     public Map<String, List<Object>> getPkValues() throws SQLException {
-        return getPkValuesByColumn();
+        List<String> pkColumnNameList = getTableMeta().getPrimaryKeyOnlyName();
+        Map<String, List<Object>> pkValuesMap = new HashMap<>(pkColumnNameList.size());
+
+        // first obtain the existing pk value from the insert rows (if exists)
+        if (!containsColumns() || containsAnyPk()) {
+            pkValuesMap.putAll(getPkValuesByColumn());
+        }
+        // other from auto-increment
+        for (String columnName : pkColumnNameList) {
+            if (!pkValuesMap.containsKey(columnName)) {
+                pkValuesMap.put(columnName, getGeneratedKeys(columnName));
+            }
+        }
+        return pkValuesMap;
+    }
+
+    /**
+     * Whether the insert columns contain any pk columns
+     *
+     * @return true: contain at least one pk column. false: do not contain any pk columns
+     */
+    public boolean containsAnyPk() {
+        SQLInsertRecognizer recognizer = (SQLInsertRecognizer)sqlRecognizer;
+        List<String> insertColumns = recognizer.getInsertColumns();
+        if (CollectionUtils.isEmpty(insertColumns)) {
+            return false;
+        }
+        List<String> pkColumnNameList = getTableMeta().getPrimaryKeyOnlyName();
+        if (CollectionUtils.isEmpty(pkColumnNameList)) {
+            return false;
+        }
+        List<String> newColumns = ColumnUtils.delEscape(insertColumns, getDbType());
+        return pkColumnNameList.stream().anyMatch(pkColumn -> newColumns.contains(pkColumn)
+            || CollectionUtils.toUpperList(newColumns).contains(pkColumn.toUpperCase()));
     }
 
     @Override
diff --git a/rm-datasource/src/test/java/io/seata/rm/datasource/exec/OracleInsertExecutorTest.java b/rm-datasource/src/test/java/io/seata/rm/datasource/exec/OracleInsertExecutorTest.java
index 3397f9ab7bd..b6dcdde79bf 100644
--- a/rm-datasource/src/test/java/io/seata/rm/datasource/exec/OracleInsertExecutorTest.java
+++ b/rm-datasource/src/test/java/io/seata/rm/datasource/exec/OracleInsertExecutorTest.java
@@ -32,8 +32,10 @@
 import org.mockito.Mockito;
 
 import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -198,6 +200,144 @@ public void testStatement_pkValueByAuto_NotSupportYetException() throws Exceptio
 
     }
 
+    @Test
+    public void testGetPkValues_SinglePk() throws SQLException {
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+
+        List<String> pkColumns = new ArrayList<>();
+        pkColumns.add(ID_COLUMN);
+        doReturn(pkColumns).when(tableMeta).getPrimaryKeyOnlyName();
+
+        // mock pk values from insert rows
+        Map<String, List<Object>> mockPkValuesFromColumn = new HashMap<>();
+        mockPkValuesFromColumn.put(ID_COLUMN, Collections.singletonList(PK_VALUE_ID + 1));
+        doReturn(mockPkValuesFromColumn).when(insertExecutor).getPkValuesByColumn();
+
+        // mock pk values from auto increment
+        List<Object> mockPkValuesAutoGenerated = Collections.singletonList(PK_VALUE_ID);
+        doReturn(mockPkValuesAutoGenerated).when(insertExecutor).getGeneratedKeys(ID_COLUMN);
+
+        // situation1: insert columns are empty
+        List<String> columns = new ArrayList<>();
+        when(sqlInsertRecognizer.getInsertColumns()).thenReturn(columns);
+        when(sqlInsertRecognizer.insertColumnsIsEmpty()).thenReturn(true);
+        Assertions.assertIterableEquals(mockPkValuesFromColumn.entrySet(), insertExecutor.getPkValues().entrySet());
+
+        // situation2: insert columns contain the pk column
+        columns = new ArrayList<>();
+        columns.add(ID_COLUMN);
+        columns.add(USER_NAME_COLUMN);
+        when(sqlInsertRecognizer.getInsertColumns()).thenReturn(columns);
+        when(sqlInsertRecognizer.insertColumnsIsEmpty()).thenReturn(false);
+        Assertions.assertIterableEquals(mockPkValuesFromColumn.entrySet(), insertExecutor.getPkValues().entrySet());
+
+        // situation3: insert columns are not empty and do not contain the pk column
+        columns = new ArrayList<>();
+        columns.add(USER_NAME_COLUMN);
+        when(sqlInsertRecognizer.getInsertColumns()).thenReturn(columns);
+        when(sqlInsertRecognizer.insertColumnsIsEmpty()).thenReturn(false);
+        Assertions.assertIterableEquals(
+            Collections.singletonMap(ID_COLUMN, mockPkValuesAutoGenerated).entrySet(),
+            insertExecutor.getPkValues().entrySet());
+    }
+
+    @Test
+    public void testGetPkValues_MultiPk() throws SQLException {
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+
+        List<String> pkColumns = new ArrayList<>();
+        pkColumns.add(ID_COLUMN);
+        pkColumns.add(USER_ID_COLUMN);
+        doReturn(pkColumns).when(tableMeta).getPrimaryKeyOnlyName();
+
+        // mock all pk values from insert rows
+        Map<String, List<Object>> mockAllPkValuesFromColumn = new HashMap<>();
+        mockAllPkValuesFromColumn.put(ID_COLUMN, Collections.singletonList(PK_VALUE_ID + 1));
+        mockAllPkValuesFromColumn.put(USER_ID_COLUMN, Collections.singletonList(PK_VALUE_USER_ID + 1));
+        doReturn(mockAllPkValuesFromColumn).when(insertExecutor).getPkValuesByColumn();
+
+        // mock pk values from auto increment
+        List<Object> mockPkValuesAutoGenerated_ID = Collections.singletonList(PK_VALUE_ID);
+        doReturn(mockPkValuesAutoGenerated_ID).when(insertExecutor).getGeneratedKeys(ID_COLUMN);
+        List<Object> mockPkValuesAutoGenerated_USER_ID = Collections.singletonList(PK_VALUE_USER_ID);
+        doReturn(mockPkValuesAutoGenerated_USER_ID).when(insertExecutor).getGeneratedKeys(USER_ID_COLUMN);
+
+        // situation1: insert columns are empty
+        List<String> insertColumns = new ArrayList<>();
+        when(sqlInsertRecognizer.getInsertColumns()).thenReturn(insertColumns);
+        when(sqlInsertRecognizer.insertColumnsIsEmpty()).thenReturn(true);
+        Assertions.assertIterableEquals(mockAllPkValuesFromColumn.entrySet(), insertExecutor.getPkValues().entrySet());
+
+        // situation2: insert columns contain all pk columns
+        insertColumns = new ArrayList<>();
+        insertColumns.add(ID_COLUMN);
+        insertColumns.add(USER_ID_COLUMN);
+        insertColumns.add(USER_NAME_COLUMN);
+        when(sqlInsertRecognizer.getInsertColumns()).thenReturn(insertColumns);
+        when(sqlInsertRecognizer.insertColumnsIsEmpty()).thenReturn(false);
+        Assertions.assertIterableEquals(mockAllPkValuesFromColumn.entrySet(), insertExecutor.getPkValues().entrySet());
+
+        // situation3: insert columns contain partial pk columns
+        insertColumns = new ArrayList<>();
+        insertColumns.add(ID_COLUMN);
+        insertColumns.add(USER_NAME_COLUMN);
+        when(sqlInsertRecognizer.getInsertColumns()).thenReturn(insertColumns);
+        when(sqlInsertRecognizer.insertColumnsIsEmpty()).thenReturn(false);
+
+        Map<String, List<Object>> mockPkValuesFromColumn_ID = new HashMap<>();
+        mockPkValuesFromColumn_ID.put(ID_COLUMN, Collections.singletonList(PK_VALUE_ID + 1));
+        doReturn(mockPkValuesFromColumn_ID).when(insertExecutor).getPkValuesByColumn();
+
+        Map<String, List<Object>> expectPkValues = new HashMap<>(mockPkValuesFromColumn_ID);
+        expectPkValues.put(USER_ID_COLUMN, mockPkValuesAutoGenerated_USER_ID);
+        Assertions.assertIterableEquals(expectPkValues.entrySet(), insertExecutor.getPkValues().entrySet());
+
+        // situation4: insert columns are not empty and do not contain the pk column
+        insertColumns = new ArrayList<>();
+        insertColumns.add(USER_NAME_COLUMN);
+        when(sqlInsertRecognizer.getInsertColumns()).thenReturn(insertColumns);
+        when(sqlInsertRecognizer.insertColumnsIsEmpty()).thenReturn(false);
+
+        doReturn(new HashMap<>()).when(insertExecutor).getPkValuesByColumn();
+
+        expectPkValues = new HashMap<>();
+        expectPkValues.put(ID_COLUMN, mockPkValuesAutoGenerated_ID);
+        expectPkValues.put(USER_ID_COLUMN, mockPkValuesAutoGenerated_USER_ID);
+        Assertions.assertIterableEquals(expectPkValues.entrySet(), insertExecutor.getPkValues().entrySet());
+    }
+
+    @Test
+    public void testContainsAnyPK() {
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+
+        Assertions.assertFalse(insertExecutor.containsAnyPk());
+
+        mockInsertColumns();
+        doReturn(null).when(tableMeta).getPrimaryKeyOnlyName();
+        Assertions.assertFalse(insertExecutor.containsAnyPk());
+
+        List<String> pkColumns = new ArrayList<>();
+        pkColumns.add(System.currentTimeMillis() + "");
+        doReturn(pkColumns).when(tableMeta).getPrimaryKeyOnlyName();
+        Assertions.assertFalse(insertExecutor.containsAnyPk());
+
+        pkColumns = new ArrayList<>();
+        pkColumns.add(ID_COLUMN);
+        doReturn(pkColumns).when(tableMeta).getPrimaryKeyOnlyName();
+        Assertions.assertTrue(insertExecutor.containsAnyPk());
+
+        pkColumns = new ArrayList<>();
+        pkColumns.add(ID_COLUMN);
+        pkColumns.add(USER_ID_COLUMN);
+        doReturn(pkColumns).when(tableMeta).getPrimaryKeyOnlyName();
+        Assertions.assertTrue(insertExecutor.containsAnyPk());
+
+        pkColumns = new ArrayList<>();
+        pkColumns.add(ID_COLUMN);
+        pkColumns.add(System.currentTimeMillis() + "");
+        doReturn(pkColumns).when(tableMeta).getPrimaryKeyOnlyName();
+        Assertions.assertTrue(insertExecutor.containsAnyPk());
+    }
 
     private List<String> mockInsertColumns() {
         List<String> columns = new ArrayList<>();