!2612 openGauss端实现全局逻辑时钟功能

Merge pull request !2612 from 周聪/glt_20221213_cherrypick

src/bin/initdb/initdb.cpp

@@ -207,6 +207,7 @@ static char* snapshot_names[] = { "schema", "create", "prepare", "sample", "publ
 static char* snapshot_files[SNAPSHOT_LEN];
 static bool enableDCF = false;
 #endif
+static bool enableGLT = false;
 static bool made_new_pgdata = false;
 static bool found_existing_pgdata = false;
 static bool made_new_xlogdir = false;
@@ -1594,6 +1595,12 @@ static void setup_config(void)
 
         conflines = ss_addnodeparmater(conflines);
     }
+    
+    if (enableGLT) {
+        nRet = strcpy_s(repltok, sizeof(repltok), "enable_glt = on");
+        securec_check_ss_c(nRet, "\0", "\0");
+        conflines = replace_token(conflines, "#enable_glt = off", repltok);
+    }
 
     nRet = sprintf_s(path, sizeof(path), "%s/postgresql.conf", pg_data);
     securec_check_ss_c(nRet, "\0", "\0");
@@ -3969,6 +3976,7 @@ int main(int argc, char* argv[])
         {"socketpath", required_argument, NULL, 16},
         {"enable-dss", no_argument, NULL, 17},
         {"dms_url", required_argument, NULL, 18},
+        {"enable-glt", no_argument, NULL, 19},
         {NULL, 0, NULL, 0}};
 
     int c, i, ret;
@@ -4274,6 +4282,10 @@ int main(int argc, char* argv[])
                 check_input_spec_char(optarg);
                 ss_nodedatainfo = xstrdup(optarg);
                 break;
+            case 19:
+                enableGLT = true;
+                break;
+
             default:
                 /* getopt_long already emitted a complaint */
                 write_stderr(_("Try \"%s --help\" for more information.\n"), progname);

src/common/backend/utils/misc/guc/guc_storage.cpp

@@ -647,7 +647,17 @@ static void InitStorageConfigureNamesBool()
             check_mix_replication_param,
             NULL,
             NULL},
-
+        {{"enable_glt",
+            PGC_POSTMASTER,
+            NODE_SINGLENODE,
+            UNGROUPED,
+            gettext_noop("Enable GLT in openGauss-shardingsphere distributed transaction."),
+            NULL},
+            &g_instance.attr.attr_storage.glt_attr.enable_glt,
+            false,
+            NULL,
+            NULL,
+            NULL},
 #ifndef ENABLE_MULTIPLE_NODES
         {{"enable_dcf",
             PGC_POSTMASTER,

src/common/backend/utils/misc/postgresql_single.conf.sample

@@ -793,6 +793,11 @@ audit_enabled = on
 #enable_codegen_print = off		# dump the IR function
 #codegen_cost_threshold = 10000		# the threshold to allow use LLVM Optimization
 
+#------------------------------------------------------------------------------
+# GLOBAL LOGIC TIME
+#------------------------------------------------------------------------------
+#enable_glt = off
+
 #------------------------------------------------------------------------------
 # JOB SCHEDULER OPTIONS
 #------------------------------------------------------------------------------
@@ -835,3 +840,4 @@ job_queue_processes = 10        # Number of concurrent jobs, optional: [0..1000]
 #ss_scrlock_worker_count = 2
 #ss_scrlock_worker_bind_core = ''
 #ss_scrlock_server_bind_core = ''
+

src/gausskernel/CMakeLists.txt

@@ -163,6 +163,7 @@ endif()
 list(APPEND gaussdb_objects
     $<TARGET_OBJECTS:common_backend_pgxc_single_barrier>
     $<TARGET_OBJECTS:common_backend_pgxc_single_copy>
+    $<TARGET_OBJECTS:gausskernel_storage_access_glt>
     $<TARGET_OBJECTS:gausskernel_storage_access_heap>
     $<TARGET_OBJECTS:gausskernel_storage_access_index>
     $<TARGET_OBJECTS:gausskernel_storage_access_nbtree>

src/gausskernel/process/postmaster/postmaster.cpp

@@ -80,6 +80,7 @@
 #endif
 
 #include "access/cbmparsexlog.h"
+#include "access/glt.h"
 #include "access/obs/obs_am.h"
 #include "access/transam.h"
 #include "access/ustore/undo/knl_uundoapi.h"
@@ -1563,6 +1564,8 @@ int PostmasterMain(int argc, char* argv[])
      */
     InitializePostmasterGUC();
 
+    gltInitMethod(g_instance.attr.attr_storage.glt_attr.enable_glt);
+
     t_thrd.myLogicTid = noProcLogicTid + POSTMASTER_LID;
     if (output_config_variable != NULL) {
         /*

src/gausskernel/process/tcop/postgres.cpp

@@ -42,6 +42,7 @@
 #include "access/xact.h"
 #include "access/ustore/undo/knl_uundoapi.h"
 #include "access/double_write.h"
+#include "access/glt.h"
 #include "catalog/namespace.h"
 #include "catalog/pg_authid.h"
 #include "catalog/pg_database.h"
@@ -9171,6 +9172,10 @@ int PostgresMain(int argc, char* argv[], const char* dbname, const char* usernam
 
                 stmt_name = pq_getmsgstring(&input_message);
                 query_string = pq_getmsgstring(&input_message);
+
+                /* check sql if is a glt command */
+                gltMethods->checkIsGLTCommand(query_string);
+
                 if (strlen(query_string) > SECUREC_MEM_MAX_LEN) {
                     ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                                     errmsg("Too long query_string.")));
@@ -11909,4 +11914,4 @@ void ResetInterruptCxt()
 
     t_thrd.int_cxt.CritSectionCount = 0;
 
-}
+}

src/gausskernel/storage/access/CMakeLists.txt

@@ -9,6 +9,7 @@ set(CMAKE_MODULE_PATH
     ${CMAKE_CURRENT_SOURCE_DIR}/common
     ${CMAKE_CURRENT_SOURCE_DIR}/gin
     ${CMAKE_CURRENT_SOURCE_DIR}/gist
+    ${CMAKE_CURRENT_SOURCE_DIR}/glt
     ${CMAKE_CURRENT_SOURCE_DIR}/hash
     ${CMAKE_CURRENT_SOURCE_DIR}/hbstore
     ${CMAKE_CURRENT_SOURCE_DIR}/heap
@@ -36,6 +37,7 @@ add_subdirectory(cbtree)
 add_subdirectory(common)
 add_subdirectory(gin)
 add_subdirectory(gist)
+add_subdirectory(glt)
 add_subdirectory(hash)
 add_subdirectory(hbstore)
 add_subdirectory(heap)

src/gausskernel/storage/access/Makefile

@@ -2,7 +2,7 @@ subdir = src/gausskernel/storage/access
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-SUBDIRS	    = cbtree common heap index nbtree ubtree psort rmgrdesc transam obs hash spgist gist gin hbstore redo table ustore 
+SUBDIRS	    = cbtree common heap index nbtree ubtree psort rmgrdesc transam obs hash spgist gist gin hbstore redo table ustore glt
 ifeq ($(enable_lite_mode), no)
 SUBDIRS += archive
 endif

src/gausskernel/storage/access/glt/CMakeLists.txt

@@ -0,0 +1,21 @@
+#This is the main CMAKE for build all components.
+AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} TGT_glt_SRC)
+
+set(TGT_glt_INC 
+    ${PROJECT_OPENGS_DIR}/contrib/log_fdw
+    ${PROJECT_TRUNK_DIR}/distribute/bin/gds
+    ${PROJECT_SRC_DIR}/include/libcomm
+    ${PROJECT_SRC_DIR}/include
+    ${PROJECT_SRC_DIR}/lib/gstrace
+    ${LZ4_INCLUDE_PATH}
+    ${LIBCGROUP_INCLUDE_PATH}
+    ${LIBORC_INCLUDE_PATH}
+    ${EVENT_INCLUDE_PATH}
+    ${PROTOBUF_INCLUDE_PATH}
+    ${ZLIB_INCLUDE_PATH}
+)
+
+set(table_DEF_OPTIONS ${MACRO_OPTIONS})
+set(table_COMPILE_OPTIONS ${OPTIMIZE_OPTIONS} ${OS_OPTIONS} ${PROTECT_OPTIONS} ${WARNING_OPTIONS} ${BIN_SECURE_OPTIONS} ${CHECK_OPTIONS})
+set(table_LINK_OPTIONS ${BIN_LINK_OPTIONS})
+add_static_objtarget(gausskernel_storage_access_glt TGT_glt_SRC TGT_glt_INC "${table_DEF_OPTIONS}" "${table_COMPILE_OPTIONS}" "${table_LINK_OPTIONS}")

src/gausskernel/storage/access/glt/Makefile

@@ -0,0 +1,14 @@
+subdir = src/gausskernel/storage/access/glt
+top_builddir = ../../../../..
+include $(top_builddir)/src/Makefile.global
+
+ifneq "$(MAKECMDGOALS)" "clean"
+  ifneq "$(MAKECMDGOALS)" "distclean"
+     ifneq "$(shell which g++ |grep hutaf_llt |wc -l)" "1"
+        -include $(DEPEND)
+     endif
+  endif
+endif
+OBJS = glt.o
+
+include $(top_srcdir)/src/gausskernel/common.mk

src/gausskernel/storage/access/glt/glt.cpp

@@ -0,0 +1,256 @@
+#include "access/glt.h"
+#include "access/slru.h"
+#include "storage/procarray.h"
+
+/* the max lenght of sql for glt*/
+const int glt_max_token_num = 5;
+const int glt_command_length = 40;
+const TransactionId glt_default_xmin = 2;
+const CommitSeqNo invalid_csn = -1;
+const TransactionId GLT_ERROR_XID = -1;
+bool enable_glt = false;
+GLTMethods* gltMethods = NULL;
+
+void gltInitMethod(bool flag);
+
+CommitSeqNo gltGetCommitCSNBeforeCommitInProgress(CommitSeqNo csn, TransactionId xid);
+void gltSetLRUCacheXid(CommitSeqNo csn, TransactionId xid);
+void gltGetCommitCSNInXact(TransactionId xid);
+void gltGetCommitCSNInTwophase(TransactionId xid);
+CommitSeqNo separatingGLTCommand(const char* query_string, char* (&token)[10000]);
+CommitSeqNo string2UnsignLong(const char* str);
+bool gltCheckCommand(const char* query_string);
+CommitSeqNo gltGetSnapshotCSN();
+TransactionId gltGetSnapshotXmin(CommitSeqNo csn, TransactionId xmin);
+
+CommitSeqNo gltOffGetCommitCSNBeforeCommitInProgress(CommitSeqNo csn, TransactionId xid);
+void gltOffGetCommitCSNInXact(TransactionId xid);
+void gltOffGetCommitCSNInTwophase(TransactionId xid);
+bool gltOffCheckCommand(const char* query_string);
+CommitSeqNo gltOffGetSnapshotCSN();
+TransactionId gltOffGetSnapshotXmin(CommitSeqNo csn, TransactionId xmin);
+
+void gltInitMethod(bool flag)
+{
+    if (flag) {
+        enable_glt = true;
+        gltMethods = (GLTMethods*) malloc(sizeof(GLTMethods));
+        gltMethods->getCommitCSNBeforeCommitInProgress = gltGetCommitCSNBeforeCommitInProgress;
+        gltMethods->getCommitCSNInTwophase = gltGetCommitCSNInTwophase;
+        gltMethods->getCommitCSNInXact = gltGetCommitCSNInXact;
+        gltMethods->checkIsGLTCommand = gltCheckCommand;
+        gltMethods->getSnapshotCSN = gltGetSnapshotCSN;
+        gltMethods->getSnapshotXmin = gltGetSnapshotXmin;
+    } else {
+        enable_glt = false;
+        gltMethods = (GLTMethods*) malloc(sizeof(GLTMethods));
+        gltMethods->getCommitCSNBeforeCommitInProgress = gltOffGetCommitCSNBeforeCommitInProgress;
+        gltMethods->getCommitCSNInTwophase = gltOffGetCommitCSNInTwophase;
+        gltMethods->getCommitCSNInXact = gltOffGetCommitCSNInXact;
+        gltMethods->checkIsGLTCommand = gltOffCheckCommand;
+        gltMethods->getSnapshotCSN = gltOffGetSnapshotCSN;
+        gltMethods->getSnapshotXmin = gltOffGetSnapshotXmin;
+    }
+}
+
+CommitSeqNo gltGetCommitCSNBeforeCommitInProgress(CommitSeqNo csn, TransactionId xid) 
+{
+    LWLockAcquire(GLTLRULock, LW_SHARED);
+    ereport(DEBUG1,
+        (errmsg("LWLockAcquire(GLTLRULock, LW_EXCLUSIVE)")));
+    CommitSeqNo latestCSN;
+
+    if (t_thrd.proc->gltCommitCSN != 0) {
+        latestCSN = t_thrd.proc->gltCommitCSN;
+    } else {
+        latestCSN = t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo - 1;
+    }
+    latestCSN = latestCSN > csn ? latestCSN : csn;
+    LWLockRelease(GLTLRULock);
+    ereport(DEBUG1,
+         (errmsg("LWLockRelease(GLTLRULock)")));
+    return latestCSN;
+}
+
+void getCommitCSN(TransactionId xid) {
+    LWLockAcquire(GLTLRULock, LW_EXCLUSIVE);
+    ereport(DEBUG1,
+        (errmsg("LWLockAcquire(GLTLRULock, LW_EXCLUSIVE)")));
+    if (t_thrd.proc->gltCommitCSN != 0) {
+        t_thrd.proc->commitCSN = t_thrd.proc->gltCommitCSN;
+    } else {
+        t_thrd.proc->commitCSN = t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo - 1;
+    }
+    if (t_thrd.proc->gltCommitCSN >= t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo) {
+        t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo = t_thrd.proc->gltCommitCSN + 1;
+    }
+    t_thrd.proc->gltCommitCSN = 0;
+    t_thrd.proc->gltSnapshotCSN = 0;
+
+    LWLockRelease(GLTLRULock);
+    ereport(DEBUG1,
+         (errmsg("LWLockRelease(GLTLRULock)")));
+}
+
+void gltGetCommitCSNInXact(TransactionId xid)
+{
+    getCommitCSN(xid);
+}
+
+void gltGetCommitCSNInTwophase(TransactionId xid)
+{
+    getCommitCSN(xid);
+}
+
+CommitSeqNo separatingGLTCommand(const char* query_string, char token[][glt_command_length])
+{
+    char chBuffer[glt_command_length];
+    char *pchDilem = " ";
+    char *pchStrTmpIn = NULL;
+    char *pchTmp = NULL;
+    int num = 0;
+    strncpy_s(chBuffer, sizeof(chBuffer), query_string, sizeof(chBuffer) - 1);
+    pchTmp = chBuffer;
+
+    while(NULL != ( pchTmp = strtok_r( pchTmp, pchDilem, &pchStrTmpIn) ) && num <= glt_command_length)
+	{
+		strncpy_s(token[num], sizeof(token[num]), pchTmp, sizeof(token[num]) - 1);
+		pchTmp = NULL;
+        num++;
+	}
+    return num;
+}
+
+/*
+ * check a request is the glt command from shardingsphere
+ * there are two kind of template "SELECT XXXX AS SETSNAPSHOTCSN" and "SELECT XXXXX AS SETSNAPSHOTCSN"
+ * if true, store the csn in a private variable of thread
+ * if false, do nothing
+ */
+bool gltCheckCommand(const char* query_string)
+{
+    const int glt_request_token_num = 4;
+    if (strlen(query_string) == 0 || strlen(query_string) >= glt_command_length) {
+        return false;
+    }
+    const char* set_snapshotcsn_request = "SETSNAPSHOTCSN";
+    const char* set_commitcsn_request = "SETCOMMITCSN";
+    const char* select_substring = "SELECT";
+    const char* as_substring = "AS";
+    CommitSeqNo num = 0;
+    char token[glt_max_token_num][glt_command_length];
+    memset(token, 0, sizeof(token));
+
+    num = separatingGLTCommand(query_string, token);
+
+
+    if (num != glt_request_token_num) {
+        return false;
+    }
+    if (strncmp(token[0], select_substring, strlen(select_substring)) != 0 || strncmp(token[2], as_substring, strlen(as_substring)) != 0) {
+        return false;
+    }
+    if (strncmp(token[3], set_snapshotcsn_request, strlen(set_snapshotcsn_request)) == 0) {
+        CommitSeqNo csn = string2UnsignLong(token[1]);
+        if (csn == invalid_csn) {
+            return false;
+        }
+        t_thrd.proc->gltSnapshotCSN = csn;
+    } else if (strncmp(token[3], set_commitcsn_request, strlen(set_commitcsn_request)) == 0) {
+        CommitSeqNo csn = string2UnsignLong(token[1]);
+        if (csn == invalid_csn) {
+            return false;
+        }
+        t_thrd.proc->gltCommitCSN = csn;
+    } else {
+        return false;
+    }
+    return true;
+}
+
+/*
+ * parse a string into uint64
+ * if the string can't be parsed, return -1
+ */
+CommitSeqNo string2UnsignLong(const char* str) 
+{
+    CommitSeqNo csn = 0;
+
+    if (str == NULL || strlen(str) == 0) {
+        return invalid_csn;
+    }
+
+    for (uint i = 0; i < strlen(str); i++) {
+        if (str[i] >= '0' && str[i] <= '9') {
+            csn = csn * 10 + (str[i] - '0');
+        } else {
+            return invalid_csn;
+        }
+    }
+    return csn;
+}
+
+CommitSeqNo gltGetSnapshotCSN()
+{
+    ereport(DEBUG1,
+        (errmsg("gltGetSnapshotCSN: gltSnapshotCSN  %ld nextCommitSeqNo %ld",
+            t_thrd.proc->gltSnapshotCSN,
+            pg_atomic_read_u64(&t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo) )));
+    if (t_thrd.proc->gltSnapshotCSN != 0) {
+        return t_thrd.proc->gltSnapshotCSN;
+    } else {
+        return pg_atomic_read_u64(&t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo);
+    }
+}
+
+TransactionId gltGetSnapshotXmin(CommitSeqNo csn, TransactionId xmin) 
+{
+    TransactionId result;
+    if (t_thrd.proc->gltSnapshotCSN != 0 && enable_glt) {
+        result = (TransactionId)glt_default_xmin;
+    } else {
+        result =  xmin;
+    }
+    ereport(DEBUG1,
+        (errmsg("gltGetSnapshotXmin: csn %ld xmin %ld result %ld",
+                csn, xmin, result)));
+    return result;
+}
+
+CommitSeqNo gltOffGetCommitCSNBeforeCommitInProgress(CommitSeqNo csn, TransactionId xid)
+{
+    CommitSeqNo latestCSN = t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo;
+    latestCSN = latestCSN > csn ? latestCSN : csn;
+    return latestCSN;
+}
+
+void gltOffGetCommitCSNInXact(TransactionId xid)
+{
+    setCommitCsn(getLocalNextCSN());
+}
+
+void gltOffGetCommitCSNInTwophase(TransactionId xid)
+{
+    setCommitCsn(getLocalNextCSN());
+}
+
+/*
+ * check a request is the glt command from shardingsphere
+ * there are two kind of template "SELECT XXXX AS SETSNAPSHOTCSN" and "SELECT XXXXX AS SETSNAPSHOTCSN"
+ * if true, store the csn in a private variable of thread
+ * if false, do nothing
+ */
+bool gltOffCheckCommand(const char* query_string)
+{
+    return false;
+}
+
+CommitSeqNo gltOffGetSnapshotCSN()
+{
+    return pg_atomic_read_u64(&t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo);
+}
+
+TransactionId gltOffGetSnapshotXmin(CommitSeqNo csn, TransactionId xmin) 
+{
+    return xmin;
+}

src/gausskernel/storage/access/transam/csnlog.cpp

@@ -116,6 +116,10 @@ void CSNLogSetCommitSeqNo(TransactionId xid, int nsubxids, TransactionId *subxid
     int offset = 0;
     TransactionId topxid = xid;
 
+    ereport(DEBUG1,
+        (errmsg("CSNLogSetCommitSeqNo xid %ld csn %lu", 
+        xid, csn)));
+
     /* for standby node, don't set invalid or abort csn mark. */
     if ((t_thrd.xact_cxt.useLocalSnapshot ||
          g_instance.attr.attr_storage.IsRoachStandbyCluster) &&

src/gausskernel/storage/access/transam/twophase.cpp

@@ -93,6 +93,7 @@
 #include "access/xloginsert.h"
 #include "access/xlogutils.h"
 #include "access/cstore_am.h"
+#include "access/glt.h"
 #include "catalog/pg_type.h"
 #include "catalog/storage.h"
 #include "commands/tablespace.h"
@@ -2497,7 +2498,11 @@ void FinishPreparedTransaction(const char *gid, bool isCommit)
          */
         if (useLocalXid || !IsPostmasterEnvironment || GTM_FREE_MODE) {
             SetXact2CommitInProgress(xid, 0);
-            setCommitCsn(getLocalNextCSN());
+            if (enable_glt) {
+                gltMethods->getCommitCSNInTwophase(xid);
+            } else {
+                setCommitCsn(getLocalNextCSN());
+            }
         }
 
 #ifdef ENABLE_MOT

src/gausskernel/storage/access/transam/xact.cpp

@@ -37,6 +37,7 @@
 #include "access/csnlog.h"
 #include "access/cstore_am.h"
 #include "access/cstore_rewrite.h"
+#include "access/glt.h"
 #include "access/multixact.h"
 #include "access/subtrans.h"
 #include "access/transam.h"
@@ -1621,7 +1622,11 @@ static TransactionId RecordTransactionCommit(void)
             /* set commit CSN and update global CSN in gtm free mode. */
             SetXact2CommitInProgress(xid, 0);
 #endif
-            setCommitCsn(getLocalNextCSN());
+            if (enable_glt) {
+                gltMethods->getCommitCSNInXact(xid);
+            } else {
+                setCommitCsn(getLocalNextCSN());
+            }
         } else {
             /* for dn auto commit condition, get a new next csn from gtm. */
             if (TransactionIdIsNormal(xid) &&
@@ -8090,8 +8095,13 @@ CommitSeqNo SetXact2CommitInProgress(TransactionId xid, CommitSeqNo csn)
 {
     int nchildren;
     TransactionId *children = NULL;
-    CommitSeqNo latestCSN = t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo;
-    latestCSN = latestCSN > csn ? latestCSN : csn;
+    CommitSeqNo latestCSN;
+    if (enable_glt) {
+        latestCSN = gltMethods->getCommitCSNBeforeCommitInProgress(csn, xid);
+    } else {
+        latestCSN = t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo;
+        latestCSN = latestCSN > csn ? latestCSN : csn;
+    }
     if (!TransactionIdIsValid(xid))
         xid = GetTopTransactionIdIfAny();
     /*

src/gausskernel/storage/ipc/procarray.cpp

@@ -76,6 +76,7 @@
 
 #include "access/clog.h"
 #include "access/csnlog.h"
+#include "access/glt.h"
 #include "access/subtrans.h"
 #include "access/transam.h"
 #include "access/twophase.h"
@@ -1920,6 +1921,10 @@ RETRY:
             }
         }
 
+        if (enable_glt) {
+            snapshot->snapshotcsn = gltMethods->getSnapshotCSN();
+        }
+
         if (result) {
             if (GTM_LITE_MODE) {
                 /* gtm lite check csn, if not pass, try to get local snapshot form multiversion again */
@@ -2108,7 +2113,11 @@ RETRY_GET:
     }
 #endif
 
-    snapshot->snapshotcsn = pg_atomic_read_u64(&t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo);
+    if (enable_glt) {
+        snapshot->snapshotcsn = gltMethods->getSnapshotCSN();
+    } else {
+        snapshot->snapshotcsn = pg_atomic_read_u64(&t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo);
+    }
 
     if (GTM_LITE_MODE) {  /* gtm lite check csn, should always pass the check */
         (void)set_proc_csn_and_check("GetLocalSnapshotDataFromProc", snapshot->snapshotcsn,
@@ -2171,6 +2180,9 @@ GROUP_GET_SNAPSHOT:
 #endif
 
     snapshot->xmin = xmin;
+    if (enable_glt) {
+        snapshot->xmin = gltMethods->getSnapshotXmin(snapshot->snapshotcsn, snapshot->xmin);
+    }
     snapshot->xmax = xmax;
     snapshot->curcid = GetCurrentCommandId(false);
 
@@ -4598,7 +4610,12 @@ Snapshot GetLocalSnapshotData(Snapshot snapshot)
     u_sess->utils_cxt.RecentXmin = snapxid->xmin;
     snapshot->xmin = snapxid->xmin;
     snapshot->xmax = snapxid->xmax;
-    snapshot->snapshotcsn = snapxid->snapshotcsn;
+    if (enable_glt) {
+        snapshot->snapshotcsn = gltMethods->getSnapshotCSN();
+        snapshot->xmin = gltMethods->getSnapshotXmin(snapshot->snapshotcsn, snapshot->xmin);
+    } else {
+        snapshot->snapshotcsn = snapxid->snapshotcsn;
+    }
     snapshot->curcid = GetCurrentCommandId(false);
 
     snapshot->active_count = 0;

src/gausskernel/storage/lmgr/lwlocknames.txt

@@ -138,3 +138,4 @@ GsStackLock 128
 ConfigFileLock 129
 DropArchiveSlotLock 130
 AboCacheLock 131
+GLTLRULock 132

src/include/access/glt.h

@@ -0,0 +1,119 @@
+/* -------------------------------------------------------------------------
+ *
+ * glt.h
+ * openGauss global logic time method definitions.
+ *
+ *
+ * Portions Copyright (c) 2020 Huawei Technologies Co.,Ltd.
+ * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/access/glt.h
+ *
+ * -------------------------------------------------------------------------
+ */
+#ifndef GLT_H
+#define GLT_H
+
+#include <map>
+#include "postgres.h"
+#include "knl/knl_variable.h"
+
+#include "access/transam.h"
+#include "access/xact.h"
+#include "access/twophase.h"
+
+using namespace std;
+
+extern const TransactionId GLT_ERROR_XID ; 
+
+typedef struct GLTMethods {
+    /* ------------------------------------------------------------------------
+     * Global Logic Time APIs
+     * ------------------------------------------------------------------------
+     */
+
+    /*
+     * get GLobal CSN for SetXact2CommitInProgress, shardingsphere will send
+     * the global CSN for commit before local transactions commit in distributed
+     * transaction. GLT will compare global csn and nextCommitSeqNo (local csn)
+     * , use the latest one.
+     */
+    CommitSeqNo (*getCommitCSNBeforeCommitInProgress)(CommitSeqNo csn, TransactionId xid);
+
+    /*
+     * get GLobal CSN to commit local transaction and advance local csn,
+     * shardingsphere will send the global CSN before local commit starts in
+     * distributed transacion. GLT will compare global csn and nextCommitSeqNo
+     * (local csn), use the latest one. GLT won't advance local csn if
+     * nextCommitSeqNo > global csn, because GLT hasn't been able to actively
+     * advance the global csn.
+     *
+     * @param xid: xid of the transaction to be committed
+     */
+    void (*getCommitCSNInXact)(TransactionId xid);
+
+    /*
+     * get GLobal CSN to perfrom a twophase commit and advance local csn ,
+     * shardingsphere will send the global CSN before local commit starts in
+     * distributed transacion. GLT will compare global csn and nextCommitSeqNo
+     * (local csn), use the latest one. GLT won't advance local csn if
+     * nextCommitSeqNo > global csn, because GLT hasn't been able to actively
+     * advance the global csn.
+     *
+     * @param xid: xid of the transaction to be committed
+     */
+    void (*getCommitCSNInTwophase)(TransactionId xid);
+
+    /*
+     * check a sql if is a glt command, shardingsphere send the global csn to
+     * dns by a dql in distribute transaction for  visibility judgment and
+     * transaction commit. If it's a glt command, dn will extract the global
+     * csn in sql and save it.
+     *
+     * @param query_string: a query string
+     */
+    bool (*checkIsGLTCommand)(const char *query_string);
+
+    /*
+     * get csn for snapshot csn. GLT will compare global csn and nextCommitSeqNo
+     * (local csn), use the latest one.
+     * this interface is reserved for future optimization.
+     *
+     * @param csn: the csn for judging visibility
+     */
+    CommitSeqNo (*getSnapshotCSN)();
+
+    /*
+     * get xmin by LRUCache or globalXmin
+     *
+     * @param csn: the snapshot csn
+     * @param xmin: the snapshot xmin
+     */
+    TransactionId (*getSnapshotXmin)(CommitSeqNo csn, TransactionId xmin);
+
+} GLTMethods;
+
+extern bool enable_glt;
+extern GLTMethods *gltMethods;
+
+/* methods if enable glt*/
+extern CommitSeqNo gltGetCommitCSNBeforeCommitInProgress(CommitSeqNo csn, TransactionId xid);
+extern void gltGetCommitCSNInXact(TransactionId xid);
+extern void gltGetCommitCSNInTwophase(TransactionId xid);
+extern bool gltCheckCommand(const char *query_string);
+extern CommitSeqNo gltGetSnapshotCSN();
+extern TransactionId gltGetSnapshotXmin(CommitSeqNo csn, TransactionId xmin);
+
+/* methods if not enable glt*/
+extern CommitSeqNo gltOffGetCommitCSNBeforeCommitInProgress(CommitSeqNo csn, TransactionId xid);
+extern void gltOffGetCommitCSNInXact(TransactionId xid);
+extern void gltOffGetCommitCSNInTwophase(TransactionId xid);
+extern bool gltOffCheckCommand(const char *query_string);
+extern CommitSeqNo gltOffGetSnapshotCSN();
+extern TransactionId gltOffGetSnapshotXmin(CommitSeqNo csn, TransactionId xmin);
+
+
+/* init gltMethods according to 'enable_glt' in postgres.conf */
+extern void gltInitMethod(bool flag);
+#endif

src/include/knl/knl_guc/knl_instance_attr_storage.h

@@ -119,6 +119,11 @@ typedef struct knl_instance_attr_dms {
     int scrlock_worker_count;
 } knl_instance_attr_dms;
 
+typedef struct knl_instance_attr_glt {
+    /* GLT switch */
+    bool enable_glt;
+} knl_instance_attr_glt;
+
 typedef struct knl_instance_attr_storage {
     bool wal_log_hints;
     bool EnableHotStandby;
@@ -186,6 +191,7 @@ typedef struct knl_instance_attr_storage {
     knl_instance_attr_nvm nvm_attr;
     knl_instance_attr_dss dss_attr;
     knl_instance_attr_dms dms_attr;
+    knl_instance_attr_glt glt_attr;
     int num_internal_lock_partitions[LWLOCK_PART_KIND];
     char* num_internal_lock_partitions_str;
     int wal_insert_status_entries_power;

src/include/storage/proc.h

@@ -226,7 +226,8 @@ struct PGPROC {
 
     /* commit sequence number send down */
     CommitSeqNo commitCSN;
-
+    CommitSeqNo gltSnapshotCSN;
+    CommitSeqNo gltCommitCSN;
     /* Support for group transaction status update. */
     bool clogGroupMember;                   /* true, if member of clog group */
     pg_atomic_uint32 clogGroupNext;         /* next clog group member */

src/test/regress/output/recovery_2pc_tools.source

@@ -256,6 +256,7 @@ select name,vartype,unit,min_val,max_val from pg_settings where name <> 'qunit_c
  enable_global_plancache                          | bool    |      |           | 
  enable_global_stats                              | bool    |      |           | 
  enable_global_syscache                           | bool    |      |           | 
+ enable_glt                                       | bool    |      |           | 
  enable_gtt_concurrent_truncate                   | bool    |      |           | 
  enable_hadoop_env                                | bool    |      |           | 
  enable_hashagg                                   | bool    |      |           |