Added oracle database support

README.md

@@ -8,10 +8,10 @@
 [![maven](https://img.shields.io/maven-central/v/io.seata/seata-parent.svg)](https://search.maven.org/search?q=io.seata)
 [![](https://img.shields.io/twitter/follow/seataio.svg?label=Follow&style=social&logoWidth=0)](https://twitter.com/intent/follow?screen_name=seataio)
 
+
 ## What is Seata?
 
 A **distributed transaction solution** with high performance and ease of use for **microservices** architecture.
-
 ### Distributed Transaction Problem in Microservices
 
 Let's imagine a traditional monolithic application. Its business is built up with 3 modules. They use a single local data source.

rm-datasource/src/main/java/io/seata/rm/datasource/AsyncWorker.java

@@ -15,20 +15,7 @@
  */
 package io.seata.rm.datasource;
 
-import java.sql.Connection;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
+import com.alibaba.druid.util.JdbcConstants;
 import io.seata.common.exception.NotSupportYetException;
 import io.seata.common.exception.ShouldNeverHappenException;
 import io.seata.common.thread.NamedThreadFactory;
@@ -40,9 +27,24 @@
 import io.seata.core.model.ResourceManagerInbound;
 import io.seata.rm.DefaultResourceManager;
 import io.seata.rm.datasource.undo.UndoLogManager;
+import io.seata.rm.datasource.undo.UndoLogManagerOracle;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
 import static io.seata.core.constants.ConfigurationKeys.CLIENT_ASYNC_COMMIT_BUFFER_LIMIT;
 
 /**
@@ -69,8 +71,7 @@ private static class Phase2Context {
          * @param resourceId      the resource id
          * @param applicationData the application data
          */
-        public Phase2Context(BranchType branchType, String xid, long branchId, String resourceId,
-                             String applicationData) {
+        public Phase2Context(BranchType branchType, String xid, long branchId, String resourceId, String applicationData) {
             this.xid = xid;
             this.branchId = branchId;
             this.resourceId = resourceId;
@@ -102,19 +103,17 @@ public Phase2Context(BranchType branchType, String xid, long branchId, String re
     }
 
     private static int ASYNC_COMMIT_BUFFER_LIMIT = ConfigurationFactory.getInstance().getInt(
-        CLIENT_ASYNC_COMMIT_BUFFER_LIMIT, 10000);
+            CLIENT_ASYNC_COMMIT_BUFFER_LIMIT, 10000);
+
+    private static final BlockingQueue<Phase2Context> ASYNC_COMMIT_BUFFER = new LinkedBlockingQueue<>(ASYNC_COMMIT_BUFFER_LIMIT);
 
-    private static final BlockingQueue<Phase2Context> ASYNC_COMMIT_BUFFER = new LinkedBlockingQueue<>(
-        ASYNC_COMMIT_BUFFER_LIMIT);
 
     private static ScheduledExecutorService timerExecutor;
 
     @Override
-    public BranchStatus branchCommit(BranchType branchType, String xid, long branchId, String resourceId,
-                                     String applicationData) throws TransactionException {
+    public BranchStatus branchCommit(BranchType branchType, String xid, long branchId, String resourceId, String applicationData) throws TransactionException {
         if (!ASYNC_COMMIT_BUFFER.offer(new Phase2Context(branchType, xid, branchId, resourceId, applicationData))) {
-            LOGGER.warn("Async commit buffer is FULL. Rejected branch [" + branchId + "/" + xid
-                + "] will be handled by housekeeping later.");
+            LOGGER.warn("Async commit buffer is FULL. Rejected branch [" + branchId + "/" + xid + "] will be handled by housekeeping later.");
         }
         return BranchStatus.PhaseTwo_Committed;
     }
@@ -125,7 +124,7 @@ public BranchStatus branchCommit(BranchType branchType, String xid, long branchI
     public synchronized void init() {
         LOGGER.info("Async Commit Buffer Limit: " + ASYNC_COMMIT_BUFFER_LIMIT);
         timerExecutor = new ScheduledThreadPoolExecutor(1,
-            new NamedThreadFactory("AsyncWorker", 1, true));
+                new NamedThreadFactory("AsyncWorker", 1, true));
         timerExecutor.scheduleAtFixedRate(new Runnable() {
             @Override
             public void run() {
@@ -160,11 +159,11 @@ private void doBranchCommits() {
 
         for (Map.Entry<String, List<Phase2Context>> entry : mappedContexts.entrySet()) {
             Connection conn = null;
+            DataSourceProxy dataSourceProxy;
             try {
                 try {
-                    DataSourceManager resourceManager = (DataSourceManager)DefaultResourceManager.get()
-                        .getResourceManager(BranchType.AT);
-                    DataSourceProxy dataSourceProxy = resourceManager.get(entry.getKey());
+                    DataSourceManager resourceManager = (DataSourceManager) DefaultResourceManager.get().getResourceManager(BranchType.AT);
+                    dataSourceProxy = resourceManager.get(entry.getKey());
                     if (dataSourceProxy == null) {
                         throw new ShouldNeverHappenException("Failed to find resource on " + entry.getKey());
                     }
@@ -180,9 +179,13 @@ private void doBranchCommits() {
                     xids.add(commitContext.xid);
                     branchIds.add(commitContext.branchId);
                     int maxSize = xids.size() > branchIds.size() ? xids.size() : branchIds.size();
-                    if (maxSize == UNDOLOG_DELETE_LIMIT_SIZE) {
+                    if(maxSize == UNDOLOG_DELETE_LIMIT_SIZE){
                         try {
-                            UndoLogManager.batchDeleteUndoLog(xids, branchIds, conn);
+                            if(JdbcConstants.ORACLE.equalsIgnoreCase(dataSourceProxy.getDbType())) {
+                                UndoLogManagerOracle.batchDeleteUndoLog(xids, branchIds, conn);
+                            } else {
+                                UndoLogManager.batchDeleteUndoLog(xids, branchIds, conn);
+                            }
                         } catch (Exception ex) {
                             LOGGER.warn("Failed to batch delete undo log [" + branchIds + "/" + xids + "]", ex);
                         }
@@ -196,8 +199,12 @@ private void doBranchCommits() {
                 }
 
                 try {
-                    UndoLogManager.batchDeleteUndoLog(xids, branchIds, conn);
-                } catch (Exception ex) {
+                    if(JdbcConstants.ORACLE.equalsIgnoreCase(dataSourceProxy.getDbType())) {
+                        UndoLogManagerOracle.batchDeleteUndoLog(xids, branchIds, conn);
+                    } else {
+                        UndoLogManager.batchDeleteUndoLog(xids, branchIds, conn);
+                    }
+                }catch (Exception ex) {
                     LOGGER.warn("Failed to batch delete undo log [" + branchIds + "/" + xids + "]", ex);
                 }
 
@@ -214,8 +221,7 @@ private void doBranchCommits() {
     }
 
     @Override
-    public BranchStatus branchRollback(BranchType branchType, String xid, long branchId, String resourceId,
-                                       String applicationData) throws TransactionException {
+    public BranchStatus branchRollback(BranchType branchType, String xid, long branchId, String resourceId, String applicationData) throws TransactionException {
         throw new NotSupportYetException();
 
     }

rm-datasource/src/main/java/io/seata/rm/datasource/ConnectionProxy.java

@@ -15,6 +15,7 @@
  */
 package io.seata.rm.datasource;
 
+import com.alibaba.druid.util.JdbcConstants;
 import java.sql.Connection;
 import java.sql.SQLException;
 
@@ -28,6 +29,7 @@
 import io.seata.rm.datasource.exec.LockConflictException;
 import io.seata.rm.datasource.undo.SQLUndoLog;
 import io.seata.rm.datasource.undo.UndoLogManager;
+import io.seata.rm.datasource.undo.UndoLogManagerOracle;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -184,7 +186,11 @@ private void processGlobalTransactionCommit() throws SQLException {
 
         try {
             if (context.hasUndoLog()) {
-                UndoLogManager.flushUndoLogs(this);
+               if(JdbcConstants.ORACLE.equalsIgnoreCase(this.getDbType())) {
+                   UndoLogManagerOracle.flushUndoLogs(this);
+               } else {
+                   UndoLogManager.flushUndoLogs(this);
+               }
             }
             targetConnection.commit();
         } catch (Throwable ex) {

rm-datasource/src/main/java/io/seata/rm/datasource/DataSourceManager.java

@@ -15,6 +15,7 @@
  */
 package io.seata.rm.datasource;
 
+import com.alibaba.druid.util.JdbcConstants;
 import java.net.InetSocketAddress;
 import java.util.List;
 import java.util.Map;
@@ -43,8 +44,14 @@
 import io.seata.discovery.registry.RegistryFactory;
 import io.seata.rm.AbstractResourceManager;
 import io.seata.rm.datasource.undo.UndoLogManager;
+import io.seata.rm.datasource.undo.UndoLogManagerOracle;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeoutException;
 
 import static io.seata.common.exception.FrameworkErrorCode.NoAvailableService;
 
@@ -176,7 +183,14 @@ public BranchStatus branchRollback(BranchType branchType, String xid, long branc
             throw new ShouldNeverHappenException();
         }
         try {
-            UndoLogManager.undo(dataSourceProxy, xid, branchId);
+            if(JdbcConstants.ORACLE.equalsIgnoreCase(dataSourceProxy.getDbType())) {
+                UndoLogManagerOracle.undo(dataSourceProxy, xid, branchId);
+            }
+            else if(JdbcConstants.MYSQL.equalsIgnoreCase(dataSourceProxy.getDbType())){
+                UndoLogManager.undo(dataSourceProxy, xid, branchId);
+            } else {
+                throw new NotSupportYetException("DbType[" + dataSourceProxy.getDbType() + "] is not support yet!");
+            }
         } catch (TransactionException te) {
             if (LOGGER.isInfoEnabled()) {
                 LOGGER.info("branchRollback failed reason [{}]", te.getMessage());

rm-datasource/src/main/java/io/seata/rm/datasource/exec/BaseTransactionalExecutor.java

@@ -31,6 +31,9 @@
 import java.util.List;
 import java.util.StringJoiner;
 
+
+import com.alibaba.druid.util.JdbcConstants;
+import io.seata.rm.datasource.sql.struct.TableMetaCacheOracle;
 /**
  * The type Base transactional executor.
  *
@@ -150,8 +153,13 @@ protected TableMeta getTableMeta() {
      * @return the table meta
      */
     protected TableMeta getTableMeta(String tableName) {
-        if (tableMeta == null) {
-             tableMeta = TableMetaCache.getTableMeta(statementProxy.getConnectionProxy().getDataSourceProxy(), tableName);
+        if (tableMeta != null) {
+            return tableMeta;
+        }
+        if(JdbcConstants.ORACLE.equalsIgnoreCase(statementProxy.getConnectionProxy().getDbType())) {
+            tableMeta = TableMetaCacheOracle.getTableMeta(statementProxy.getConnectionProxy().getDataSourceProxy(), tableName);
+        } else {
+            tableMeta = TableMetaCache.getTableMeta(statementProxy.getConnectionProxy().getDataSourceProxy(), tableName);
         }
         return tableMeta;
     }

rm-datasource/src/main/java/io/seata/rm/datasource/exec/ExecuteTemplate.java

@@ -15,14 +15,14 @@
  */
 package io.seata.rm.datasource.exec;
 
-import java.sql.SQLException;
-import java.sql.Statement;
-
 import io.seata.core.context.RootContext;
 import io.seata.rm.datasource.StatementProxy;
 import io.seata.rm.datasource.sql.SQLRecognizer;
 import io.seata.rm.datasource.sql.SQLVisitorFactory;
 
+import java.sql.SQLException;
+import java.sql.Statement;
+
 /**
  * The type Execute template.
  *

rm-datasource/src/main/java/io/seata/rm/datasource/sql/SQLVisitorFactory.java

@@ -26,7 +26,10 @@
 import io.seata.rm.datasource.sql.druid.MySQLInsertRecognizer;
 import io.seata.rm.datasource.sql.druid.MySQLSelectForUpdateRecognizer;
 import io.seata.rm.datasource.sql.druid.MySQLUpdateRecognizer;
-
+import io.seata.rm.datasource.sql.druid.oracle.OracleDeleteRecognizer;
+import io.seata.rm.datasource.sql.druid.oracle.OracleInsertRecognizer;
+import io.seata.rm.datasource.sql.druid.oracle.OracleSelectForUpdateRecognizer;
+import io.seata.rm.datasource.sql.druid.oracle.OracleUpdateRecognizer;
 import java.util.List;
 
 /**
@@ -62,7 +65,19 @@ public static SQLRecognizer get(String sql, String dbType) {
                     recognizer = new MySQLSelectForUpdateRecognizer(sql, ast);
                 }
             }
-        } else {
+        }  else if (JdbcConstants.ORACLE.equalsIgnoreCase(dbType)) {
+            if (ast instanceof SQLInsertStatement) {
+                recognizer = new OracleInsertRecognizer(sql, ast);
+            } else if (ast instanceof SQLUpdateStatement) {
+                recognizer = new OracleUpdateRecognizer(sql, ast);
+            } else if (ast instanceof SQLDeleteStatement) {
+                recognizer = new OracleDeleteRecognizer(sql, ast);
+            } else if (ast instanceof SQLSelectStatement) {
+                if (((SQLSelectStatement) ast).getSelect().getQueryBlock().isForUpdate()) {
+                    recognizer = new OracleSelectForUpdateRecognizer(sql, ast);
+                }
+            }
+        }else {
             throw new UnsupportedOperationException("Just support MySQL by now!");
         }
         return recognizer;