external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Ads set statement

pull/7178/head
gindibay 2023-09-04 10:45:17 +03:00
parent 09ab1ac4bb
commit 78991cab7c
3 changed files with 288 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/backend/distributed/commands/distribute_object_ops.c
View File

@ -448,7 +448,7 @@ static DistributeObjectOps Database_Alter = {
.deparse = DeparseAlterDatabaseStmt,
.qualify = NULL,
.preprocess = PreprocessAlterDatabaseStmt,
.postprocess = PostprocessAlterDistributedObjectStmt,
.postprocess = NULL,
.objectType = OBJECT_DATABASE,
.operationType = DIST_OPS_ALTER,
.address = NULL,
@ -459,7 +459,7 @@ static DistributeObjectOps Database_Set = {
.deparse = DeparseAlterDatabaseSetStmt,
.qualify = NULL,
.preprocess = PreprocessAlterDatabaseSetStmt,
.postprocess = PostprocessAlterDistributedObjectStmt,
.postprocess = NULL,
.objectType = OBJECT_DATABASE,
.operationType = DIST_OPS_ALTER,
.address = NULL,
@ -468,7 +468,7 @@ static DistributeObjectOps Database_Set = {
static DistributeObjectOps Database_Rename = {
.deparse = DeparseAlterDatabaseRenameStmt,
.qualify = NULL,
.preprocess = PreprocessAlterDatabaseRenameStmt,
.preprocess = NULL,
.postprocess = NULL,
.objectType = OBJECT_DATABASE,
.operationType = DIST_OPS_ALTER,
@ -479,7 +479,7 @@ static DistributeObjectOps Database_Rename = {
static DistributeObjectOps Database_RefreshColl = {
.deparse = DeparseAlterDatabaseRefreshCollStmt,
.qualify = NULL,
.preprocess = PreprocessAlterDatabaseRefreshCollStmt,
.preprocess = NULL,//TODO: Add PostprocessAfter adding addresses
.postprocess = NULL,
.objectType = OBJECT_DATABASE,
.operationType = DIST_OPS_ALTER,

165
src/backend/distributed/deparser/citus_deparse_set_utils.c Normal file
View File

@ -0,0 +1,165 @@
#include "postgres.h"
#include "lib/stringinfo.h"
#include "nodes/nodes.h"
#include "parser/parse_type.h"
#include "utils/builtins.h"
#include "utils/guc.h"
static void
AppendVarSetValue2(StringInfo buf, VariableSetStmt *setStmt);
static void
AppendVariableSet2(StringInfo buf, VariableSetStmt *setStmt);
static void
AppendVarSetValue2(StringInfo buf, VariableSetStmt *setStmt)
{
ListCell *varArgCell = NULL;
ListCell *firstCell = list_head(setStmt->args);
Assert(setStmt->kind == VAR_SET_VALUE);
foreach(varArgCell, setStmt->args)
{
Node *varArgNode = lfirst(varArgCell);
A_Const *varArgConst = NULL;
TypeName *typeName = NULL;
if (IsA(varArgNode, A_Const))
{
varArgConst = (A_Const *) varArgNode;
}
else if (IsA(varArgNode, TypeCast))
{
TypeCast *varArgTypeCast = (TypeCast *) varArgNode;
varArgConst = castNode(A_Const, varArgTypeCast->arg);
typeName = varArgTypeCast->typeName;
}
else
{
elog(ERROR, "unrecognized node type: %d", varArgNode->type);
}
/* don't know how to start SET until we inspect first arg */
if (varArgCell != firstCell)
{
appendStringInfoChar(buf, ',');
}
else if (typeName != NULL)
{
appendStringInfoString(buf, " SET TIME ZONE");
}
else
{
appendStringInfo(buf, " SET %s =", quote_identifier(setStmt->name));
}
Node *value = (Node *) &varArgConst->val;
switch (value->type)
{
case T_Integer:
{
appendStringInfo(buf, " %d", intVal(value));
break;
}
case T_Float:
{
appendStringInfo(buf, " %s", strVal(value));
break;
}
case T_String:
{
if (typeName != NULL)
{
/*
* Must be a ConstInterval argument for TIME ZONE. Coerce
* to interval and back to normalize the value and account
* for any typmod.
*/
Oid typoid = InvalidOid;
int32 typmod = -1;
typenameTypeIdAndMod(NULL, typeName, &typoid, &typmod);
Assert(typoid == INTERVALOID);
Datum interval =
DirectFunctionCall3(interval_in,
CStringGetDatum(strVal(value)),
ObjectIdGetDatum(InvalidOid),
Int32GetDatum(typmod));
char *intervalout =
DatumGetCString(DirectFunctionCall1(interval_out,
interval));
appendStringInfo(buf, " INTERVAL '%s'", intervalout);
}
else
{
appendStringInfo(buf, " %s", quote_literal_cstr(strVal(
value)));
}
break;
}
default:
{
elog(ERROR, "Unexpected Value type in VAR_SET_VALUE arguments.");
break;
}
}
}
}
/*
* AppendVariableSet appends a string representing the VariableSetStmt to a buffer
*/
static void
AppendVariableSet2(StringInfo buf, VariableSetStmt *setStmt)
{
switch (setStmt->kind)
{
case VAR_SET_VALUE:
{
AppendVarSetValue2(buf, setStmt);
break;
}
case VAR_SET_CURRENT:
{
appendStringInfo(buf, " SET %s FROM CURRENT", quote_identifier(
setStmt->name));
break;
}
case VAR_SET_DEFAULT:
{
appendStringInfo(buf, " SET %s TO DEFAULT", quote_identifier(setStmt->name));
break;
}
case VAR_RESET:
{
appendStringInfo(buf, " RESET %s", quote_identifier(setStmt->name));
break;
}
case VAR_RESET_ALL:
{
appendStringInfoString(buf, " RESET ALL");
break;
}
/* VAR_SET_MULTI is a special case for SET TRANSACTION that should not occur here */
case VAR_SET_MULTI:
default:
{
ereport(ERROR, (errmsg("Unable to deparse SET statement")));
break;
}
}
}

121
src/backend/distributed/deparser/deparse_database_stmts.c
View File

@ -20,6 +20,7 @@
#include "distributed/deparser.h"
#include "distributed/citus_ruleutils.h"
#include "commands/defrem.h"
#include "distributed/deparser.h"
#include "distributed/log_utils.h"
@ -41,7 +42,6 @@ DeparseAlterDatabaseOwnerStmt(Node *node)
return str.data;
}
static void
AppendAlterDatabaseOwnerStmt(StringInfo buf, AlterOwnerStmt *stmt)
{
@ -53,7 +53,6 @@ AppendAlterDatabaseOwnerStmt(StringInfo buf, AlterOwnerStmt *stmt)
RoleSpecString(stmt->newowner, true));
}
static void
AppendGrantDatabases(StringInfo buf, GrantStmt *stmt)
{
@ -71,7 +70,6 @@ AppendGrantDatabases(StringInfo buf, GrantStmt *stmt)
}
}
static void
AppendGrantOnDatabaseStmt(StringInfo buf, GrantStmt *stmt)
{
@ -141,6 +139,11 @@ AppendAlterDatabaseSetStmt(StringInfo buf, AlterDatabaseSetStmt *stmt)
appendStringInfo(buf, ";");
}
static void
AppendDefElemConnLimit(StringInfo buf, DefElem *def)
{
appendStringInfo(buf, " CONNECTION LIMIT %ld",(long int) defGetNumeric(def));
}
static void
AppendAlterDatabaseStmt(StringInfo buf, AlterDatabaseStmt *stmt)
@ -150,12 +153,112 @@ AppendAlterDatabaseStmt(StringInfo buf, AlterDatabaseStmt *stmt)
if (stmt->options)
{
ListCell *cell = NULL;
appendStringInfo(buf, "WITH OPTION ");
appendStringInfo(buf, "WITH ");
foreach(cell, stmt->options)
{
DefElem *def = castNode(DefElem, lfirst(cell));
printf("test");
if (strcmp(def->defname, "is_template") == 0)
{
appendStringInfo(buf, "%s %s", quote_identifier(def->defname),
quote_literal_cstr(strVal(def->arg)));
}
else if (strcmp(def->defname, "connection_limit") == 0)
{
AppendDefElemConnLimit(buf, def);
}
else if (strcmp(def->defname, "allow_connections") == 0)
{
ereport(ERROR,
errmsg("ALLOW_CONNECTIONS is not supported"));
}
else{
ereport(ERROR,
errmsg("unrecognized AlterDatabaseStmt option: %s",
def->defname));
}
if (cell != list_tail(stmt->options))
{
appendStringInfo(buf, ", ");
}
}
}
appendStringInfo(buf, ";");
}
static void
AppendAlterDatabaseSetStmt(StringInfo buf, AlterDatabaseSetStmt *stmt)
{
appendStringInfo(buf, "ALTER DATABASE %s SET ", quote_identifier(stmt->dbname));
VariableSetStmt *varSetStmt = castNode(VariableSetStmt, stmt->setstmt);
if (varSetStmt->kind == VAR_SET_VALUE)
{
appendStringInfo(buf, "%s = %s", quote_identifier(varSetStmt->name),
quote_literal_cstr(strVal(varSetStmt->args)));
}
else if (varSetStmt->kind == VAR_RESET_ALL)
{
appendStringInfo(buf, "RESET ALL");
}
else if (varSetStmt->kind == VAR_RESET)
{
appendStringInfo(buf, "RESET %s", quote_identifier(varSetStmt->name));
}
else
{
ereport(ERROR,
errmsg("unrecognized AlterDatabaseSetStmt kind: %d",
varSetStmt->kind));
}
appendStringInfo(buf, ";");
}
static void
AppendDefElemConnLimit(StringInfo buf, DefElem *def)
{
appendStringInfo(buf, " CONNECTION LIMIT %ld",(long int) defGetNumeric(def));
}
static void
AppendAlterDatabaseStmt(StringInfo buf, AlterDatabaseStmt *stmt)
{
appendStringInfo(buf, "ALTER DATABASE %s ", quote_identifier(stmt->dbname));
if (stmt->options)
{
ListCell *cell = NULL;
appendStringInfo(buf, "WITH ");
foreach(cell, stmt->options)
{
DefElem *def = castNode(DefElem, lfirst(cell));
printf("test");
if (strcmp(def->defname, "is_template") == 0)
{
appendStringInfo(buf, "%s %s", quote_identifier(def->defname),
quote_literal_cstr(strVal(def->arg)));
}
else if (strcmp(def->defname, "connection_limit") == 0)
{
AppendDefElemConnLimit(buf, def);
}
else if (strcmp(def->defname, "allow_connections") == 0)
{
ereport(ERROR,
errmsg("ALLOW_CONNECTIONS is not supported"));
}
else{
ereport(ERROR,
errmsg("unrecognized AlterDatabaseStmt option: %s",
def->defname));
}
if (cell != list_tail(stmt->options))
{
appendStringInfo(buf, ", ");
@ -213,7 +316,6 @@ DeparseGrantOnDatabaseStmt(Node *node)
return str.data;
}
char *
DeparseAlterDatabaseStmt(Node *node)
{
@ -227,7 +329,6 @@ DeparseAlterDatabaseStmt(Node *node)
return str.data;
}
char *
DeparseAlterDatabaseSetStmt(Node *node)
{
@ -241,7 +342,6 @@ DeparseAlterDatabaseSetStmt(Node *node)
return str.data;
}
char *
DeparseAlterDatabaseRenameStmt(Node *node)
{
@ -250,13 +350,11 @@ DeparseAlterDatabaseRenameStmt(Node *node)
StringInfoData str;
initStringInfo(&str);
appendStringInfo(&str, "ALTER DATABASE %s RENAME TO %s;", quote_identifier(
stmt->subname), quote_identifier(stmt->newname));
appendStringInfo(&str, "ALTER DATABASE %s RENAME TO %s;", quote_identifier(stmt->subname), quote_identifier(stmt->newname));
return str.data;
}
char *
DeparseAlterDatabaseRefreshCollStmt(Node *node)
{
@ -265,8 +363,7 @@ DeparseAlterDatabaseRefreshCollStmt(Node *node)
StringInfoData str;
initStringInfo(&str);
appendStringInfo(&str, "ALTER DATABASE %s REFRESH COLLATION;", quote_identifier(
stmt->dbname));
appendStringInfo(&str, "ALTER DATABASE %s REFRESH COLLATION;", quote_identifier(stmt->dbname));
return str.data;
}