Merge pull request cloudflarearchive#1 from Miniwoffer/master

Added support for flags

Author: buster1253

README.md

@@ -3,37 +3,51 @@
 Thanks to the Blackhole Foreign Data Wrapper for PostgreSQL for providing
 a good template of how to build FDWs for PostgreSQL.
 
-##To Build:
+## To Build:
+```
     make
     (sudo) make install
-
+```
 > A note about transactional support:
 > This FDW supports Kyoto Tycoon transactions via LUA scripts.
 Therefore, KT needs to be compiled with LUA support. To do this,
 run
->
->     ./configure --enable-lua
->
+> ```sh
+> ./configure --enable-lua
+> ```
 > when compiling KT.
 >
 > Furthermore the ktserver has to be started with tranactions.lua loaded:
-
->     ./ktserver -scr <fdw directory>/transactions.lua
-
+> ```sh
+> ./ktserver -scr <fdw directory>/transactions.lua
+> ```
 > If you want to disable transactions remove -DUSE_TRANSACTIONS from the
 > Makefile in the FDW.
 
-##To Test:
-    make installcheck
-
-##Usage:
-
-    CREATE SERVER <server name> FOREIGN DATA WRAPPER kt_fdw OPTIONS
-    (host '127.0.0.1', port '1978', timeout '-1');
-    (the above options are the defaults)
-
-    CREATE USER MAPPING FOR PUBLIC SERVER kt_server;
-
-    CREATE FOREIGN TABLE <table name> (key TEXT, value TEXT) SERVER <server name>;
-
+## To Test:
+```sh
+make installcheck
+```
+
+## Flags
+### kt_set
+```pgsql
+insert into <table name> values ('key', 'value', 'kt_set');
+```
+Makes insert overwrite existing keys in kyototycoon.
+## Usage:
+```pgsql
+CREATE SERVER <server name> FOREIGN DATA WRAPPER kt_fdw OPTIONS
+(host '127.0.0.1', port '1978', timeout '-1');
+(the above options are the defaults)
+
+CREATE USER MAPPING FOR PUBLIC SERVER kt_server;
+
+CREATE FOREIGN TABLE <table name> (key TEXT, value BYTEA, flags TEXT) SERVER <server name>;
+```
 Now you can Select, Update, Delete and Insert!
+
+## TODO
+- [ ] Give the user the option not to use flags
+- [ ] Support more flags
+- [ ] Write new tests

src/kt_fdw.c

@@ -212,6 +212,7 @@ typedef struct {
     Relation rel;
     FmgrInfo *key_info;
     FmgrInfo *value_info;
+    FmgrInfo *flags_info;
     AttrNumber key_junk_no;
 } KtFdwModifyState;
 
@@ -282,7 +283,6 @@ static bool isValidOption(const char *option, Oid context)
 #ifdef DEBUG
     elog(NOTICE, "isValidOption %s", option);
 #endif
-    
     for (opt = valid_options; opt->optname; opt++){
         if (context == opt->optcontext && strcmp(opt->optname, option) == 0) {
             return true;
@@ -950,8 +950,8 @@ ktIterateForeignScan(ForeignScanState *node)
 
     if (found)
     {
-	dvalues = (Datum *) palloc(2 * sizeof(Datum));
-	nulls = (bool *) palloc(2 * sizeof(bool));
+	dvalues = (Datum *) palloc(3 * sizeof(Datum));
+	nulls = (bool *) palloc(3 * sizeof(bool));
 	for (i = 0; i < 2 ; i++)
 	{
 		nulls[i] = false;
@@ -981,6 +981,8 @@ ktIterateForeignScan(ForeignScanState *node)
 				nulls[i] = true;
 		}
 	}
+        /* set flags to null */
+        nulls[2] = true;
 
 	tuple = heap_form_tuple(estate->attinmeta->tupdesc, dvalues, nulls);
 	ExecStoreTuple(tuple, slot, InvalidBuffer, false);
@@ -1179,6 +1181,7 @@ ktBeginForeignModify(ModifyTableState *mtstate,
     fmstate->rel = rel;
     fmstate->key_info = (FmgrInfo *) palloc0(sizeof(FmgrInfo));
     fmstate->value_info = (FmgrInfo *) palloc0(sizeof(FmgrInfo));
+    fmstate->flags_info = (FmgrInfo *) palloc0(sizeof(FmgrInfo));
 
 
     if (operation == CMD_UPDATE || operation == CMD_DELETE) {
@@ -1210,6 +1213,16 @@ ktBeginForeignModify(ModifyTableState *mtstate,
     getTypeBinaryOutputInfo(attr->atttypid, &typefnoid, &isvarlena);
     fmgr_info(typefnoid, fmstate->value_info);
 
+#if PG_VERSION_NUM >= 110000
+    attr = &RelationGetDescr(rel)->attrs[2];
+#else
+    attr = RelationGetDescr(rel)->attrs[2];
+#endif
+    Assert(!attr->attisdropped);
+
+    getTypeBinaryOutputInfo(attr->atttypid, &typefnoid, &isvarlena);
+    fmgr_info(typefnoid, fmstate->flags_info);
+
     initTableOptions(&(fmstate->opt));
     getTableOptions(RelationGetRelid(rel), &(fmstate->opt));
 
@@ -1254,7 +1267,8 @@ ktExecForeignInsert(EState *estate,
 
     Datum value;
     bool isnull;
-    bytea *bkey, *bval;
+    bool isSet = false;
+    bytea *bkey, *bval, *sval;
     KtFdwModifyState *fmstate = (KtFdwModifyState *) rinfo->ri_FdwState;
 
     elog(DEBUG1,"entering function %s",__func__);
@@ -1271,9 +1285,24 @@ ktExecForeignInsert(EState *estate,
 
     bval = SendFunctionCall(fmstate->value_info, value);
 
-    if(!ktaddl(fmstate->db, VARDATA(bkey), VARSIZE(bkey) - VARHDRSZ,
-                            VARDATA(bval), VARSIZE(bval) - VARHDRSZ))
-        elog(ERROR, "Error from kt: %s", ktgeterror(fmstate->db));
+    value = slot_getattr(planSlot, 3, &isnull);
+    if(!isnull) {
+      sval = SendFunctionCall(fmstate->flags_info, value);
+      if(memcmp(VARDATA(sval), "kt_set", VARSIZE(sval) - VARHDRSZ) == 0) {
+        isSet = true;
+      }
+    }
+    if(isSet) {
+        if(!ktsetl(fmstate->db, VARDATA(bkey), VARSIZE(bkey) - VARHDRSZ,
+                            VARDATA(bval), VARSIZE(bval) - VARHDRSZ)) {
+          elog(ERROR, "Error from kt: %s", ktgeterror(fmstate->db));
+        }
+    } else {
+      if(!ktaddl(fmstate->db, VARDATA(bkey), VARSIZE(bkey) - VARHDRSZ,
+                            VARDATA(bval), VARSIZE(bval) - VARHDRSZ)) {
+          elog(ERROR, "%d, Error from kt: %s", ktgeterror(fmstate->db));
+        }
+    }
 
     return slot;
 }

src/ktlangc.cc

@@ -194,6 +194,15 @@ bool ktreplacel(KTDB*db, const char * key, size_t keylen,
     return pdb->replace(key, keylen, value, vallen);
 }
 
+bool ktsetl(KTDB*db, const char * key, size_t keylen,
+                         const char * value, size_t vallen)
+{
+    _assert_(db);
+    RemoteDB* pdb = (RemoteDB*)db;
+
+    return pdb->set(key, keylen, value, vallen);
+}
+
 bool ktremove(KTDB*db, const char * key)
 {
     _assert_(db);