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

Address reviews

onder_view
Burak Velioglu 2022-04-25 00:25:09 +03:00
parent 5efe6061e8
commit ff003bccd8
13 changed files with 120 additions and 224 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/backend/distributed/commands/dependencies.c
View File

@ -351,7 +351,7 @@ GetDependencyCreateDDLCommands(const ObjectAddress *dependency)
if (relKind == RELKIND_VIEW)
{
return CreateViewDDLCommandsIdempotent(dependency);
return CreateViewDDLCommandsIdempotent(dependency->objectId);
}
/* if this relation is not supported, break to the error at the end */

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

@ -186,7 +186,7 @@ static DistributeObjectOps Any_CreateFunction = {
.markDistributed = true,
};
static DistributeObjectOps Any_View = {
.deparse = DeparseViewStmt,
.deparse = NULL,
.qualify = NULL,
.preprocess = PreprocessViewStmt,
.postprocess = PostprocessViewStmt,

122
src/backend/distributed/commands/view.c
View File

@ -36,8 +36,10 @@
#include "utils/syscache.h"
static List * FilterNameListForDistributedViews(List *viewNamesList, bool missing_ok);
static void AppendQualifiedViewNameToCreateViewCommand(StringInfo buf, Oid viewOid);
static void AppendAliasesToCreateViewCommand(StringInfo createViewCommand, Oid viewOid);
static void AppendOptionsToCreateViewCommand(StringInfo createViewCommand, Oid viewOid);
static StringInfo CreateAlterViewOwnerCommand(Oid viewOid);
/*
* PreprocessViewStmt is called during the planning phase for CREATE OR REPLACE VIEW
@ -92,11 +94,6 @@ PostprocessViewStmt(Node *node, const char *queryString)
return NIL;
}
if (!HasAnyNodes())
{
return NIL;
}
ObjectAddress viewAddress = GetObjectAddressFromParseTree((Node *) stmt, false);
if (IsObjectAddressOwnedByExtension(&viewAddress, NULL))
@ -109,6 +106,22 @@ PostprocessViewStmt(Node *node, const char *queryString)
if (errMsg != NULL)
{
/*
* Don't need to give any warning/error messages if there is no worker nodes in
* the cluster as user's experience won't be affected on the single node even
* if the view won't be distributed.
*/
if (!HasAnyNodes())
{
return NIL;
}
/*
* If the view is already distributed, we should provide an error to not have
* different definition of view on coordinator and worker nodes. If the view
* is not distributed yet, we can create it locally to not affect user's local
* usage experience.
*/
if (IsObjectDistributed(&viewAddress))
{
RaiseDeferredError(errMsg, ERROR);
@ -122,11 +135,10 @@ PostprocessViewStmt(Node *node, const char *queryString)
EnsureDependenciesExistOnAllNodes(&viewAddress);
const char *sql = DeparseTreeNode((Node *) stmt);
List *commands = list_make3(DISABLE_DDL_PROPAGATION,
(void *) sql,
ENABLE_DDL_PROPAGATION);
List *commands = list_make1(DISABLE_DDL_PROPAGATION);
commands = list_concat(commands,
CreateViewDDLCommandsIdempotent(viewAddress.objectId));
commands = lappend(commands, ENABLE_DDL_PROPAGATION);
return NodeDDLTaskList(NON_COORDINATOR_NODES, commands);
}
@ -246,16 +258,10 @@ FilterNameListForDistributedViews(List *viewNamesList, bool missing_ok)
* executed on a node to recreate the view addressed by the viewAddress.
*/
List *
CreateViewDDLCommandsIdempotent(const ObjectAddress *viewAddress)
CreateViewDDLCommandsIdempotent(Oid viewOid)
{
StringInfo createViewCommand = makeStringInfo();
Oid viewOid = viewAddress->objectId;
char *viewName = get_rel_name(viewOid);
Oid schemaOid = get_rel_namespace(viewOid);
char *schemaName = get_namespace_name(schemaOid);
appendStringInfoString(createViewCommand, "CREATE OR REPLACE VIEW ");
AppendQualifiedViewNameToCreateViewCommand(createViewCommand, viewOid);
@ -263,21 +269,28 @@ CreateViewDDLCommandsIdempotent(const ObjectAddress *viewAddress)
AppendOptionsToCreateViewCommand(createViewCommand, viewOid);
AppendViewDefinitionToCreateViewCommand(createViewCommand, viewOid);
/* Add alter owner commmand */
StringInfo alterOwnerCommand = makeStringInfo();
char *viewOwnerName = TableOwner(viewOid);
char *qualifiedViewName = NameListToQuotedString(list_make2(makeString(schemaName),
makeString(viewName)));
appendStringInfo(alterOwnerCommand,
"ALTER VIEW %s OWNER TO %s", qualifiedViewName,
quote_identifier(viewOwnerName));
StringInfo alterOwnerCommand = CreateAlterViewOwnerCommand(viewOid);
return list_make2(createViewCommand->data,
alterOwnerCommand->data);
}
/*
* AppendQualifiedViewNameToCreateViewCommand adds the qualified view of the given view
* oid to the given create view command.
*/
static void
AppendQualifiedViewNameToCreateViewCommand(StringInfo buf, Oid viewOid)
{
char *viewName = get_rel_name(viewOid);
char *schemaName = get_namespace_name(get_rel_namespace(viewOid));
char *qualifiedViewName = quote_qualified_identifier(schemaName, viewName);
appendStringInfo(buf, "%s ", qualifiedViewName);
}
/*
* AppendAliasesToCreateViewCommand appends aliases to the create view
* command for the existing view.
@ -346,3 +359,60 @@ AppendOptionsToCreateViewCommand(StringInfo createViewCommand, Oid viewOid)
appendStringInfo(createViewCommand, "WITH (%s) ", relOptions);
}
}
/*
* AppendViewDefinitionToCreateViewCommand adds the definition of the given view to the
* given create view command.
*/
void
AppendViewDefinitionToCreateViewCommand(StringInfo buf, Oid viewOid)
{
/*
* Set search_path to NIL so that all objects outside of pg_catalog will be
* schema-prefixed.
*/
OverrideSearchPath *overridePath = GetOverrideSearchPath(CurrentMemoryContext);
overridePath->schemas = NIL;
overridePath->addCatalog = true;
PushOverrideSearchPath(overridePath);
/*
* Push the transaction snapshot to be able to get vief definition with pg_get_viewdef
*/
PushActiveSnapshot(GetTransactionSnapshot());
Datum viewDefinitionDatum = DirectFunctionCall1(pg_get_viewdef,
ObjectIdGetDatum(viewOid));
char *viewDefinition = TextDatumGetCString(viewDefinitionDatum);
PopActiveSnapshot();
PopOverrideSearchPath();
appendStringInfo(buf, "AS %s ", viewDefinition);
}
/*
* CreateAlterViewOwnerCommand creates the command to alter view owner command for the
* given view oid.
*/
static StringInfo
CreateAlterViewOwnerCommand(Oid viewOid)
{
/* Add alter owner commmand */
StringInfo alterOwnerCommand = makeStringInfo();
char *viewName = get_rel_name(viewOid);
Oid schemaOid = get_rel_namespace(viewOid);
char *schemaName = get_namespace_name(schemaOid);
char *viewOwnerName = TableOwner(viewOid);
char *qualifiedViewName = NameListToQuotedString(list_make2(makeString(schemaName),
makeString(viewName)));
appendStringInfo(alterOwnerCommand,
"ALTER VIEW %s OWNER TO %s", qualifiedViewName,
quote_identifier(viewOwnerName));
return alterOwnerCommand;
}

161
src/backend/distributed/deparser/deparse_view_stmts.c
View File

@ -22,170 +22,9 @@
#include "utils/builtins.h"
#include "utils/lsyscache.h"
static void AppendAliasesFromViewStmtToCreateViewCommand(StringInfo buf, ViewStmt *stmt);
static void AppendOptionsFromViewStmtToCreateViewCommand(StringInfo buf, ViewStmt *stmt);
static void AppendDropViewStmt(StringInfo buf, DropStmt *stmt);
static void AppendViewNameList(StringInfo buf, List *objects);
/*
* DeparseViewStmt deparses the given CREATE OR REPLACE VIEW statement
*/
char *
DeparseViewStmt(Node *node)
{
ViewStmt *stmt = castNode(ViewStmt, node);
StringInfo viewString = makeStringInfo();
Oid viewOid = RangeVarGetRelid(stmt->view, NoLock, false);
if (stmt->replace)
{
appendStringInfoString(viewString, "CREATE OR REPLACE VIEW ");
}
else
{
appendStringInfoString(viewString, "CREATE VIEW ");
}
AppendQualifiedViewNameToCreateViewCommand(viewString, viewOid);
AppendAliasesFromViewStmtToCreateViewCommand(viewString, stmt);
AppendOptionsFromViewStmtToCreateViewCommand(viewString, stmt);
/*
* Note that Postgres converts CREATE RECURSIVE VIEW commands to
* CREATE VIEW ... WITH RECURSIVE and pg_get_viewdef return it properly.
* So, we don't need to RECURSIVE views separately while obtaining the
* view creation command
*/
AppendViewDefinitionToCreateViewCommand(viewString, viewOid);
return viewString->data;
}
/*
* AppendQualifiedViewNameToCreateViewCommand adds the qualified view of the given view
* oid to the given create view command.
*/
void
AppendQualifiedViewNameToCreateViewCommand(StringInfo buf, Oid viewOid)
{
char *viewName = get_rel_name(viewOid);
char *schemaName = get_namespace_name(get_rel_namespace(viewOid));
char *qualifiedViewName = quote_qualified_identifier(schemaName, viewName);
appendStringInfo(buf, "%s ", qualifiedViewName);
}
/*
* AppendAliasesFromViewStmtToCreateViewCommand appends aliases (if exists) of the given view statement
* to the given create view command.
*/
static void
AppendAliasesFromViewStmtToCreateViewCommand(StringInfo buf, ViewStmt *stmt)
{
if (stmt->aliases == NIL)
{
return;
}
bool isFirstAlias = true;
ListCell *aliasItem;
foreach(aliasItem, stmt->aliases)
{
const char *columnAliasName = quote_identifier(strVal(lfirst(aliasItem)));
if (isFirstAlias)
{
appendStringInfoString(buf, "(");
isFirstAlias = false;
}
else
{
appendStringInfoString(buf, ",");
}
appendStringInfoString(buf, columnAliasName);
}
appendStringInfoString(buf, ") ");
}
/*
* AppendOptionsFromViewStmtToCreateViewCommand appends options (if exists) of the given view statement
* to the given create view command. Note that this function also handles
* WITH [CASCADED | LOCAL] CHECK OPTION part of the CREATE VIEW command.
*/
static void
AppendOptionsFromViewStmtToCreateViewCommand(StringInfo buf, ViewStmt *stmt)
{
if (list_length(stmt->options) == 0)
{
return;
}
bool isFirstOption = true;
ListCell *optionCell;
foreach(optionCell, stmt->options)
{
DefElem *option = (DefElem *) lfirst(optionCell);
if (isFirstOption)
{
appendStringInfoString(buf, "WITH (");
isFirstOption = false;
}
else
{
appendStringInfoString(buf, ",");
}
appendStringInfoString(buf, option->defname);
if (option->arg != NULL)
{
appendStringInfoString(buf, "=");
appendStringInfoString(buf, defGetString(option));
}
}
appendStringInfoString(buf, ") ");
}
/*
* AppendViewDefinitionToCreateViewCommand adds the definition of the given view to the
* given create view command.
*/
void
AppendViewDefinitionToCreateViewCommand(StringInfo buf, Oid viewOid)
{
/*
* Set search_path to NIL so that all objects outside of pg_catalog will be
* schema-prefixed.
*/
OverrideSearchPath *overridePath = GetOverrideSearchPath(CurrentMemoryContext);
overridePath->schemas = NIL;
overridePath->addCatalog = true;
PushOverrideSearchPath(overridePath);
/*
* Push the transaction snapshot to be able to get vief definition with pg_get_viewdef
*/
PushActiveSnapshot(GetTransactionSnapshot());
Datum viewDefinitionDatum = DirectFunctionCall1(pg_get_viewdef,
ObjectIdGetDatum(viewOid));
char *viewDefinition = TextDatumGetCString(viewDefinitionDatum);
PopActiveSnapshot();
PopOverrideSearchPath();
appendStringInfo(buf, "AS %s ", viewDefinition);
}
/*
* DeparseDropViewStmt deparses the given DROP VIEW statement.
*/

26
src/backend/distributed/metadata/dependency.c
View File

@ -165,7 +165,7 @@ static bool FollowAllDependencies(ObjectAddressCollector *collector,
DependencyDefinition *definition);
static void ApplyAddToDependencyList(ObjectAddressCollector *collector,
DependencyDefinition *definition);
static List * GetRelationRuleReferenceDependencyList(Oid relationId);
static List * GetViewRuleReferenceDependencyList(Oid relationId);
static List * ExpandCitusSupportedTypes(ObjectAddressCollector *collector,
ObjectAddress target);
static ViewDependencyNode * BuildViewDependencyGraph(Oid relationId, HTAB *nodeMap);
@ -1328,14 +1328,18 @@ ExpandCitusSupportedTypes(ObjectAddressCollector *collector, ObjectAddress targe
result = list_concat(result, indexDependencyList);
/*
* Get the dependencies of the rule for the given relation. PG keeps internal
* dependency between relation and rule. As it is stated on the PG doc, if
* Get the dependencies of the rule for the given view. PG keeps internal
* dependency between view and rule. As it is stated on the PG doc, if
* there is an internal dependency, NORMAL and AUTO dependencies of the
* dependent object behave much like they were dependencies of the referenced
* object.
*/
List *ruleRefDepList = GetRelationRuleReferenceDependencyList(relationId);
result = list_concat(result, ruleRefDepList);
char relKind = get_rel_relkind(relationId);
if (relKind == RELKIND_VIEW)
{
List *ruleRefDepList = GetViewRuleReferenceDependencyList(relationId);
result = list_concat(result, ruleRefDepList);
}
}
default:
@ -1349,15 +1353,15 @@ ExpandCitusSupportedTypes(ObjectAddressCollector *collector, ObjectAddress targe
/*
* GetRelationRuleReferenceDependencyList returns the dependencies of the relation's
* GetViewRuleReferenceDependencyList returns the dependencies of the view's
* internal rule dependencies.
*/
static List *
GetRelationRuleReferenceDependencyList(Oid relationId)
GetViewRuleReferenceDependencyList(Oid relationId)
{
List *dependencyTupleList = GetPgDependTuplesForDependingObjects(RelationRelationId,
relationId);
List *nonInternalRuleDependencies = NIL;
List *nonInternalDependenciesOfDependingRules = NIL;
HeapTuple depTup = NULL;
foreach_ptr(depTup, dependencyTupleList)
@ -1388,13 +1392,13 @@ GetRelationRuleReferenceDependencyList(Oid relationId)
continue;
}
nonInternalRuleDependencies = lappend(nonInternalRuleDependencies,
dependencyDef);
nonInternalDependenciesOfDependingRules =
lappend(nonInternalDependenciesOfDependingRules, dependencyDef);
}
}
}
return nonInternalRuleDependencies;
return nonInternalDependenciesOfDependingRules;
}

2
src/include/distributed/commands.h
View File

@ -660,7 +660,7 @@ extern List * PostprocessViewStmt(Node *node, const char *queryString);
extern ObjectAddress ViewStmtObjectAddress(Node *node, bool missing_ok);
extern List * PreprocessDropViewStmt(Node *node, const char *queryString,
ProcessUtilityContext processUtilityContext);
extern List * CreateViewDDLCommandsIdempotent(const ObjectAddress *viewAddress);
extern List * CreateViewDDLCommandsIdempotent(Oid viewOid);
extern char * DeparseViewStmt(Node *node);
extern char * DeparseDropViewStmt(Node *node);

1
src/include/distributed/deparser.h
View File

@ -147,7 +147,6 @@ extern ObjectAddress RenameAttributeStmtObjectAddress(Node *stmt, bool missing_o
/* forward declarations for deparse_view_stmts.c */
extern void QualifyDropViewStmt(Node *node);
extern void AppendViewDefinitionToCreateViewCommand(StringInfo buf, Oid viewOid);
extern void AppendQualifiedViewNameToCreateViewCommand(StringInfo buf, Oid viewOid);
/* forward declarations for deparse_function_stmts.c */
extern char * DeparseDropFunctionStmt(Node *stmt);

19
src/test/regress/expected/local_dist_join_mixed.out
View File

@ -1604,22 +1604,5 @@ DEBUG: Plan XXX query after replacing subqueries and CTEs: SELECT count(*) AS c
101
(1 row)
SET client_min_messages TO ERROR;
DROP SCHEMA local_dist_join_mixed CASCADE;
DEBUG: switching to sequential query execution mode
DETAIL: A command for a distributed schema is run. To make sure subsequent commands see the schema correctly we need to make sure to use only one connection for all future commands
NOTICE: drop cascades to 7 other objects
DETAIL: drop cascades to table distributed
drop cascades to table reference
drop cascades to table local
drop cascades to table unlogged_local
drop cascades to materialized view mat_view
drop cascades to view local_regular_view
drop cascades to view dist_regular_view
DEBUG: drop cascades to view local_dist_join_mixed.dist_regular_view
DETAIL: from localhost:xxxxx
CONTEXT: SQL statement "SELECT master_remove_distributed_table_metadata_from_workers(v_obj.objid, v_obj.schema_name, v_obj.object_name)"
PL/pgSQL function citus_drop_trigger() line XX at PERFORM
DEBUG: drop cascades to view local_dist_join_mixed.dist_regular_view
DETAIL: from localhost:xxxxx
CONTEXT: SQL statement "SELECT master_remove_distributed_table_metadata_from_workers(v_obj.objid, v_obj.schema_name, v_obj.object_name)"
PL/pgSQL function citus_drop_trigger() line XX at PERFORM

3
src/test/regress/expected/resync_metadata_with_sequences.out
View File

@ -140,9 +140,8 @@ INSERT INTO sensors_news VALUES (DEFAULT, DEFAULT, '2021-01-01') RETURNING *;
(1 row)
\c - - - :master_port
SET client_min_messages TO ERROR;
SELECT 1 FROM citus_activate_node('localhost', :worker_1_port);
NOTICE: Replicating postgres objects to node localhost:xxxxx
DETAIL: There are 102 objects to replicate, depending on your environment this might take a while
?column?
---------------------------------------------------------------------
1

1
src/test/regress/input/multi_copy.source
View File

@ -508,6 +508,7 @@ INSERT INTO pg_catalog.pg_dist_object(classid, objid, objsubid) values('pg_class
INSERT INTO pg_catalog.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, 'table_to_distribute'::regclass::oid, 0);
INSERT INTO pg_catalog.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, 'second_dustbunnies'::regclass::oid, 0);
SET client_min_messages TO ERROR;
SELECT 1 FROM master_activate_node('localhost', :worker_1_port);
RESET client_min_messages;
RESET citus.shard_replication_factor;

3
src/test/regress/output/multi_copy.source
View File

@ -634,9 +634,8 @@ INSERT INTO pg_catalog.pg_dist_object(classid, objid, objsubid) values('pg_class
INSERT INTO pg_catalog.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, 'super_packed_numbers_hash'::regclass::oid, 0);
INSERT INTO pg_catalog.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, 'table_to_distribute'::regclass::oid, 0);
INSERT INTO pg_catalog.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, 'second_dustbunnies'::regclass::oid, 0);
SET client_min_messages TO ERROR;
SELECT 1 FROM master_activate_node('localhost', :worker_1_port);
NOTICE: Replicating postgres objects to node localhost:57637
DETAIL: There are 118 objects to replicate, depending on your environment this might take a while
?column?
---------------------------------------------------------------------
1

1
src/test/regress/sql/local_dist_join_mixed.sql
View File

@ -408,4 +408,5 @@ JOIN
USING
(id);
SET client_min_messages TO ERROR;
DROP SCHEMA local_dist_join_mixed CASCADE;

1
src/test/regress/sql/resync_metadata_with_sequences.sql
View File

@ -55,6 +55,7 @@ INSERT INTO sensors VALUES (DEFAULT, DEFAULT, '2010-01-01') RETURNING *;
INSERT INTO sensors_news VALUES (DEFAULT, DEFAULT, '2021-01-01') RETURNING *;
\c - - - :master_port
SET client_min_messages TO ERROR;
SELECT 1 FROM citus_activate_node('localhost', :worker_1_port);