Merge pull request #7 from jsc0218/dev_gc

jsc0218 · web-flow · commit 89e41341cf2b · 2020-01-14T11:27:43.000-05:00
fix the parallel issue in utility functions
diff --git a/src/kv_fdw.c b/src/kv_fdw.c
@@ -1,6 +1,5 @@
 
 #include "kv_fdw.h"
-#include "kv_storage.h"
 
 #include <pthread.h>
 
@@ -54,6 +53,7 @@ static void GetForeignRelSize(PlannerInfo *root,
      * min & max will call GetForeignRelSize & GetForeignPaths multiple times,
      * we should open & close db multiple times.
      */
+    printf("\n-----------------%s open----------------------\n", __func__);
     ptr = OpenRequest(foreignTableId, ptr);
 
     /* TODO better estimation */
@@ -146,6 +146,88 @@ static ForeignScan *GetForeignPlan(PlannerInfo *root,
                             NULL);
 }
 
+#ifdef VidarDB
+static void GetKeyRangeQual(Plan *plan,
+                            TupleDesc tupleDescriptor,
+                            TableReadState *readState) {
+    if (!plan) {
+        return;
+    }
+
+    int qualSize = list_length(plan->qual);
+    if (qualSize < 2) {
+        return;
+    }
+
+    bool startFlag = false;
+    bool limitFlag = false;
+    StringInfo startInfo = makeStringInfo();
+    StringInfo limitInfo = makeStringInfo();
+
+    ListCell *lc;
+    foreach (lc, plan->qual) {
+        Expr *state = lfirst(lc);
+        if (!state || !IsA(state, OpExpr)) {
+            continue;
+        }
+        OpExpr *op = (OpExpr *) state;
+        if (list_length(op->args) != 2) {
+            continue;
+        }
+
+        Node *left = list_nth(op->args, 0);
+        if (!IsA(left, Var)) {
+            continue;
+        }
+
+        Node *right = list_nth(op->args, 1);
+        if (!IsA(right, Const)) {
+            continue;
+        }
+
+        Index varattno = ((Var *) left)->varattno;
+        if (varattno != 1) {
+            continue;
+        }
+
+        HeapTuple opertup = SearchSysCache1(OPEROID, ObjectIdGetDatum(op->opno));
+        if (!HeapTupleIsValid(opertup)) {
+            ereport(ERROR, (errmsg("cache lookup failed for operator %u", op->opno)));
+        }
+        Form_pg_operator operform = (Form_pg_operator) GETSTRUCT(opertup);
+        char *oprname = NameStr(operform->oprname);
+        if (strncmp(oprname, ">=", NAMEDATALEN) == 0) {
+            startFlag = true;
+            Const *constNode = ((Const *) right);
+            Datum datum = constNode->constvalue;
+            TypeCacheEntry *typeEntry = lookup_type_cache(constNode->consttype, 0);
+            datum = ShortVarlena(datum, typeEntry->typlen, typeEntry->typstorage);
+            SerializeAttribute(tupleDescriptor, varattno-1, datum, startInfo);
+            ReleaseSysCache(opertup);
+        }else if (strncmp(oprname, "<=", NAMEDATALEN) == 0) {
+            limitFlag = true;
+            Const *constNode = ((Const *) right);
+            Datum datum = constNode->constvalue;
+            TypeCacheEntry *typeEntry = lookup_type_cache(constNode->consttype, 0);
+            datum = ShortVarlena(datum, typeEntry->typlen, typeEntry->typstorage);
+            SerializeAttribute(tupleDescriptor, varattno-1, datum, limitInfo);
+            ReleaseSysCache(opertup);
+        }else{
+            ReleaseSysCache(opertup);
+            continue;
+        }
+        
+        if (startFlag && limitFlag) {
+            readState->isRangeQueryUsed = true;
+            readState->rangeSpec.start = startInfo->data;
+            readState->rangeSpec.startLen = startInfo->len;
+            readState->rangeSpec.limit = limitInfo->data;
+            readState->rangeSpec.limitLen = limitInfo->len;
+        }
+    }
+}
+#endif
+
 static void GetKeyBasedQual(Node *node,
                             TupleDesc tupleDescriptor,
                             TableReadState *readState) {
@@ -242,9 +324,13 @@ static void BeginForeignScan(ForeignScanState *scanState, int executorFlags) {
         return;
     }
 
+    //printf("\n-----------------Plan Type: %d----------------------\n", scanState->ss.ps.plan->type);
+ 
     ListCell *lc;
     foreach (lc, scanState->ss.ps.plan->qual) {
         Expr *state = lfirst(lc);
+        //printf("\n-----------------Qual Type: %d----------------------\n", state->type);
+
         GetKeyBasedQual((Node *) state,
                         scanState->ss.ss_currentRelation->rd_att,
                         readState);
@@ -255,8 +341,18 @@ static void BeginForeignScan(ForeignScanState *scanState, int executorFlags) {
     }
 
     if (!readState->isKeyBased) {
-        Oid relationId = RelationGetRelid(scanState->ss.ss_currentRelation);
-        GetIterRequest(relationId, ptr);
+        #ifdef VidarDB
+        GetKeyRangeQual(scanState->ss.ps.plan, 
+                scanState->ss.ss_currentRelation->rd_att,
+                readState);
+        #endif
+        //if(readState->isRangeQueryUsed) {
+
+        //}else
+        //{
+            Oid relationId = RelationGetRelid(scanState->ss.ss_currentRelation);
+            GetIterRequest(relationId, ptr);   
+        //}
     }
 }
 
@@ -359,6 +455,9 @@ static TupleTableSlot *IterateForeignScan(ForeignScanState *scanState) {
             found = GetRequest(relationId, ptr, k, kLen, &v, &vLen);
             readState->done = true;
         }
+    /*} else if (readState->isRangeQueryUsed)
+    {*/
+    
     } else {
         found = NextRequest(relationId, ptr, &k, &kLen, &v, &vLen);
     }
diff --git a/src/kv_fdw.h b/src/kv_fdw.h
@@ -2,6 +2,7 @@
 #ifndef KV_FDW_H_
 #define KV_FDW_H_
 
+#include "kv_storage.h"
 
 #include <stdbool.h>
 #include <semaphore.h>
@@ -73,6 +74,16 @@ typedef struct TableReadState {
     bool isKeyBased;
     bool done;
     StringInfo key;
+    
+#ifdef VidarDB
+    bool isRangeQueryUsed;
+    RangeSpec rangeSpec;
+    void **rangeReadOptions;
+    char* valArray;
+    uint32* valLens;
+    uint32 valArraySize;
+    uint32 iterCount;
+#endif
 } TableReadState;
 
 /*
@@ -97,6 +108,9 @@ typedef enum FuncName {
     GET,
     PUT,
     DELETE,
+    #ifdef VidarDB
+    RANGEQUERY,
+    #endif
     TERMINATE
 } FuncName;
 
@@ -155,6 +169,13 @@ extern void DeleteRequest(Oid relationId,
                           char* key,
                           uint32 keyLen);
 
+#ifdef VidarDB
+extern void RangeQueryRequest(Oid relationId,
+                              SharedMem *ptr,
+                              void** readOptions,
+                              RangeSpec rangeSpec
+                              );
+#endif
 
 /* global variables */
 
diff --git a/src/kv_shm.c b/src/kv_shm.c
@@ -328,6 +328,10 @@ static void KVWorkerMain(int argc, char *argv[]) {
             case DELETE:
                 DeleteResponse(buf + sizeof(responseId));
                 break;
+            #ifdef VidarDB
+            case RANGEQUERY:
+                RangeQueryResponse(buff + sizeof(int));
+            #endif    
             default:
                 ereport(ERROR, (errmsg("%s failed in switch", __func__)));
         }
@@ -427,6 +431,7 @@ static void OpenResponse(char *area) {
     char *pos = strrchr(path, '/');
     Oid relationId = atoi(pos + 1);
     bool found;
+    
     KVHashEntry *entry = hash_search(kvTableHash, &relationId, HASH_ENTER, &found);
     if (!found) {
         entry->relationId = relationId;
@@ -952,3 +957,13 @@ static void DeleteResponse(char *area) {
         }
     }
 }
+
+#ifdef VidarDB
+void RangeQueryRequest(Oid relationId, SharedMem *ptr, void** readOptions, RangeSpec rangeSpec) {
+
+}
+
+static void RangeQueryResponse(char *area) {
+
+}
+#endif
diff --git a/src/kv_storage.cc b/src/kv_storage.cc
@@ -79,4 +79,44 @@ bool Delete(void* db, char* key, uint32 keyLen) {
     return s.ok();
 }
 
+#ifdef VidarDB
+bool RangeQuery(void* db, void** readOptions, RangeSpec range, char** valArray, uint32** valLens, uint32* valArraySize) {
+    Range r(Slice(range.start, range.startLen), Slice(range.limit, range.limitLen));
+    ReadOptions *options = static_cast<ReadOptions*>(*readOptions);
+    if (options == nullptr)
+    {
+        options = (ReadOptions*) palloc0(sizeof(ReadOptions));
+    }
+    options->max_result_num = MAXRESULTNUM;
+    *readOptions = options;
+    std::vector<std::string> res;
+    Status s; 
+    bool ret = static_cast<DB*>(db)->RangeQuery(*options, r, res, &s);
+
+    if (!s.ok()) {
+        *valArraySize = 0;
+        return false;
+    }
+    
+    *valArraySize = res.size();
+    if( *valArraySize > 0) {
+        uint32 *tmpLens = (uint32*) palloc0(*valArraySize * sizeof(uint32));
+        *valLens = tmpLens;
+        uint32 total = 0;
+        for (auto it = res.begin(); it != res.end(); ++it) {
+            *tmpLens = (*it).size();
+            total += (*it).size();
+            tmpLens++;
+        }
+        char *tmpVal = (char*) palloc0(total);
+        *valArray = tmpVal;
+        for (auto it = res.begin(); it != res.end(); ++it) {
+            memcpy(tmpVal, (*it).c_str(), (*it).size());
+            tmpVal = tmpVal + (*it).size();
+        }
+    }
+    return ret;
+}
+#endif
+
 }
diff --git a/src/kv_storage.h b/src/kv_storage.h
@@ -28,6 +28,17 @@ bool Get(void* db, char* key, uint32 keyLen, char** val, uint32* valLen);
 bool Put(void* db, char* key, uint32 keyLen, char* val, uint32 valLen);
 bool Delete(void* db, char* key, uint32 keyLen);
 
+#ifdef VidarDB
+#define MAXRESULTNUM 10000
+typedef struct RangeSpec {
+    char* start;
+    uint32 startLen;
+    char* limit;
+    uint32 limitLen;
+} RangeSpec;
+
+bool RangeQuery(void* db, void** readOptions, RangeSpec range, char** valArray, uint32** valLens, uint32* valArraySize);
+#endif
 
 #if defined(__cplusplus)
 }
diff --git a/src/kv_utility.c b/src/kv_utility.c
@@ -30,6 +30,8 @@
         PREVIOUS_UTILITY(plannedStmt, queryString, context, paramListInfo, \
                          queryEnvironment, destReceiver, completionTag)
 
+/* */
+static SharedMem *ptr = NULL;
 
 /*
  * SQL functions
@@ -200,7 +202,7 @@ Datum kv_ddl_event_end_trigger(PG_FUNCTION_ARGS) {
              * directory for it during database creation time.
              */
             KVCreateDatabaseDirectory(MyDatabaseId);
-
+            
             StringInfo kvPath = makeStringInfo();
             appendStringInfo(kvPath,
                              "%s/%s/%u/%u",
@@ -470,8 +472,10 @@ static uint64 KVCopyIntoTable(const CopyStmt *copyStmt,
                                                        ALLOCSET_DEFAULT_SIZES);
 
     Oid relationId = RelationGetRelid(relation);
-    KVFdwOptions *fdwOptions = KVGetOptions(relationId);
-    void *db = Open(fdwOptions->filename);
+    //KVFdwOptions *fdwOptions = KVGetOptions(relationId);
+    
+    //void *db = Open(fdwOptions->filename);
+    ptr = OpenRequest(relationId, ptr);
 
     TupleDesc tupleDescriptor = RelationGetDescr(relation);
     uint32 count = tupleDescriptor->natts;
@@ -514,9 +518,12 @@ static uint64 KVCopyIntoTable(const CopyStmt *copyStmt,
             }
             memcpy(value->data, buffer->data, bufLen);
 
-            if (!Put(db, key->data, key->len, value->data, value->len)) {
-                ereport(ERROR, (errmsg("error from Copy")));
-            }
+            //if (!Put(db, key->data, key->len, value->data, value->len)) {
+            //    ereport(ERROR, (errmsg("error from Copy")));
+            //}
+
+
+            PutRequest(relationId, ptr, key->data, key->len, value->data, value->len);
             rowCount++;
         }
 
@@ -527,7 +534,8 @@ static uint64 KVCopyIntoTable(const CopyStmt *copyStmt,
 
     /* end read/write sessions and close the relation */
     EndCopyFrom(copyState);
-    Close(db);
+    //Close(db);
+    CloseRequest(relationId, ptr);
     heap_close(relation, ShareUpdateExclusiveLock);
 
     return rowCount;
@@ -657,7 +665,7 @@ static void KVProcessUtility(PlannedStmt *plannedStmt,
 
         CopyStmt *copyStmt = (CopyStmt *) parseTree;
         if (KVCopyTableStatement(copyStmt)) {
-
+            
             uint64 rowCount = 0;
             if (copyStmt->is_from) {
                 rowCount = KVCopyIntoTable(copyStmt, queryString);
