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

Merge branch 'master' into velioglu/function_propagation

velioglu/tmpfuncprop
Burak Velioglu 2022-02-17 16:30:51 +03:00
parent 0bdb5d5ace 754d894375
commit 236d1cd1ef
No known key found for this signature in database
GPG Key ID: F6827E620F6549C6
36 changed files with 3018 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CONTRIBUTING.md
View File

@ -131,8 +131,6 @@ cd build
cmake ..
make -j5
sudo make install
# Optionally, you might instead want to use `sudo make install-all`
# since `multi_extension` regression test would fail due to missing downgrade scripts.
cd ../..
git clone https://github.com/citusdata/tools.git

4
ci/README.md
View File

@ -156,9 +156,9 @@ git merge "community/$PR_BRANCH"
familiar with the change.
5. You should rerun the `check-merge-to-enterprise` check on
`community/$PR_BRANCH`. You can use re-run from failed option in circle CI.
6. You can now merge the PR on enterprise. Be sure to NOT use "squash and merge",
6. You can now merge the PR on community. Be sure to NOT use "squash and merge",
but instead use the regular "merge commit" mode.
7. You can now merge the PR on community. Be sure to NOT use "squash and merge",
7. You can now merge the PR on enterprise. Be sure to NOT use "squash and merge",
but instead use the regular "merge commit" mode.
The subsequent PRs on community will be able to pass the

6
ci/check_enterprise_merge.sh
View File

@ -66,10 +66,10 @@ fi
git merge --abort
# If we have a conflict on enterprise merge on the master branch, we have a problem.
# Provide an error message to indicate that enterprise merge is needed.
# Provide an error message to indicate that enterprise merge is needed to fix this check.
if [[ $PR_BRANCH = master ]]; then
echo "ERROR: Master branch has merge conlicts with enterprise-master."
echo "Try re-running this job if you merged community PR before enterprise PR. Otherwise conflicts need to be resolved as a separate PR on enterprise."
echo "ERROR: Master branch has merge conflicts with enterprise-master."
echo "Try re-running this CI job after merging your changes into enterprise-master."
exit 1
fi

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

@ -241,6 +241,17 @@ GetDependencyCreateDDLCommands(const ObjectAddress *dependency)
return NIL;
}
/*
* Indices are created separately, however, they do show up in the dependency
* list for a table since they will have potentially their own dependencies.
* The commands will be added to both shards and metadata tables via the table
* creation commands.
*/
if (relKind == RELKIND_INDEX)
{
return NIL;
}
if (relKind == RELKIND_RELATION || relKind == RELKIND_PARTITIONED_TABLE ||
relKind == RELKIND_FOREIGN_TABLE)
{
@ -317,6 +328,11 @@ GetDependencyCreateDDLCommands(const ObjectAddress *dependency)
return DDLCommands;
}
case OCLASS_TSCONFIG:
{
return CreateTextSearchConfigDDLCommandsIdempotent(dependency);
}
case OCLASS_TYPE:
{
return CreateTypeDDLCommandsIdempotent(dependency);

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

@ -505,6 +505,62 @@ static DistributeObjectOps Sequence_Rename = {
.address = RenameSequenceStmtObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps TextSearchConfig_Alter = {
.deparse = DeparseAlterTextSearchConfigurationStmt,
.qualify = QualifyAlterTextSearchConfigurationStmt,
.preprocess = PreprocessAlterTextSearchConfigurationStmt,
.postprocess = NULL,
.address = AlterTextSearchConfigurationStmtObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps TextSearchConfig_AlterObjectSchema = {
.deparse = DeparseAlterTextSearchConfigurationSchemaStmt,
.qualify = QualifyAlterTextSearchConfigurationSchemaStmt,
.preprocess = PreprocessAlterTextSearchConfigurationSchemaStmt,
.postprocess = PostprocessAlterTextSearchConfigurationSchemaStmt,
.address = AlterTextSearchConfigurationSchemaStmtObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps TextSearchConfig_AlterOwner = {
.deparse = DeparseAlterTextSearchConfigurationOwnerStmt,
.qualify = QualifyAlterTextSearchConfigurationOwnerStmt,
.preprocess = PreprocessAlterTextSearchConfigurationOwnerStmt,
.postprocess = PostprocessAlterTextSearchConfigurationOwnerStmt,
.address = AlterTextSearchConfigurationOwnerObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps TextSearchConfig_Comment = {
.deparse = DeparseTextSearchConfigurationCommentStmt,
.qualify = QualifyTextSearchConfigurationCommentStmt,
.preprocess = PreprocessTextSearchConfigurationCommentStmt,
.postprocess = NULL,
.address = TextSearchConfigurationCommentObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps TextSearchConfig_Define = {
.deparse = DeparseCreateTextSearchStmt,
.qualify = NULL,
.preprocess = NULL,
.postprocess = PostprocessCreateTextSearchConfigurationStmt,
.address = CreateTextSearchConfigurationObjectAddress,
.markDistributed = true,
};
static DistributeObjectOps TextSearchConfig_Drop = {
.deparse = DeparseDropTextSearchConfigurationStmt,
.qualify = QualifyDropTextSearchConfigurationStmt,
.preprocess = PreprocessDropTextSearchConfigurationStmt,
.postprocess = NULL,
.address = NULL,
.markDistributed = false,
};
static DistributeObjectOps TextSearchConfig_Rename = {
.deparse = DeparseRenameTextSearchConfigurationStmt,
.qualify = QualifyRenameTextSearchConfigurationStmt,
.preprocess = PreprocessRenameTextSearchConfigurationStmt,
.postprocess = NULL,
.address = RenameTextSearchConfigurationStmtObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps Trigger_AlterObjectDepends = {
.deparse = NULL,
.qualify = NULL,
@ -811,6 +867,11 @@ GetDistributeObjectOps(Node *node)
return &Table_AlterObjectSchema;
}
case OBJECT_TSCONFIGURATION:
{
return &TextSearchConfig_AlterObjectSchema;
}
case OBJECT_TYPE:
{
return &Type_AlterObjectSchema;
@ -868,6 +929,11 @@ GetDistributeObjectOps(Node *node)
return &Statistics_AlterOwner;
}
case OBJECT_TSCONFIGURATION:
{
return &TextSearchConfig_AlterOwner;
}
case OBJECT_TYPE:
{
return &Type_AlterOwner;
@ -949,11 +1015,33 @@ GetDistributeObjectOps(Node *node)
return &Any_AlterTableMoveAll;
}
case T_AlterTSConfigurationStmt:
{
return &TextSearchConfig_Alter;
}
case T_ClusterStmt:
{
return &Any_Cluster;
}
case T_CommentStmt:
{
CommentStmt *stmt = castNode(CommentStmt, node);
switch (stmt->objtype)
{
case OBJECT_TSCONFIGURATION:
{
return &TextSearchConfig_Comment;
}
default:
{
return &NoDistributeOps;
}
}
}
case T_CompositeTypeStmt:
{
return &Any_CompositeType;
@ -1014,6 +1102,11 @@ GetDistributeObjectOps(Node *node)
return &Collation_Define;
}
case OBJECT_TSCONFIGURATION:
{
return &TextSearchConfig_Define;
}
default:
{
return &NoDistributeOps;
@ -1091,6 +1184,11 @@ GetDistributeObjectOps(Node *node)
return &Table_Drop;
}
case OBJECT_TSCONFIGURATION:
{
return &TextSearchConfig_Drop;
}
case OBJECT_TYPE:
{
return &Type_Drop;
@ -1190,6 +1288,11 @@ GetDistributeObjectOps(Node *node)
return &Statistics_Rename;
}
case OBJECT_TSCONFIGURATION:
{
return &TextSearchConfig_Rename;
}
case OBJECT_TYPE:
{
return &Type_Rename;

32
src/backend/distributed/commands/index.c
View File

@ -725,12 +725,6 @@ PostprocessIndexStmt(Node *node, const char *queryString)
{
IndexStmt *indexStmt = castNode(IndexStmt, node);
/* we are only processing CONCURRENT index statements */
if (!indexStmt->concurrent)
{
return NIL;
}
/* this logic only applies to the coordinator */
if (!IsCoordinator())
{
@ -747,14 +741,36 @@ PostprocessIndexStmt(Node *node, const char *queryString)
return NIL;
}
Oid indexRelationId = get_relname_relid(indexStmt->idxname, schemaId);
/* ensure dependencies of index exist on all nodes */
ObjectAddress address = { 0 };
ObjectAddressSet(address, RelationRelationId, indexRelationId);
EnsureDependenciesExistOnAllNodes(&address);
/* furtheron we are only processing CONCURRENT index statements */
if (!indexStmt->concurrent)
{
return NIL;
}
/*
* EnsureDependenciesExistOnAllNodes could have distributed objects that are required
* by this index. During the propagation process an active snapshout might be left as
* a side effect of inserting the local tuples via SPI. To not leak a snapshot like
* that we will pop any snapshot if we have any right before we commit.
*/
if (ActiveSnapshotSet())
{
PopActiveSnapshot();
}
/* commit the current transaction and start anew */
CommitTransactionCommand();
StartTransactionCommand();
/* get the affected relation and index */
Relation relation = table_openrv(indexStmt->relation, ShareUpdateExclusiveLock);
Oid indexRelationId = get_relname_relid(indexStmt->idxname,
schemaId);
Relation indexRelation = index_open(indexRelationId, RowExclusiveLock);
/* close relations but retain locks */

935
src/backend/distributed/commands/text_search.c Normal file
View File

@ -0,0 +1,935 @@
/*-------------------------------------------------------------------------
*
* text_search.c
* Commands for creating and altering TEXT SEARCH objects
*
* Copyright (c) Citus Data, Inc.
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "access/genam.h"
#include "access/xact.h"
#include "catalog/namespace.h"
#include "catalog/objectaddress.h"
#include "catalog/pg_ts_config.h"
#include "catalog/pg_ts_config_map.h"
#include "catalog/pg_ts_dict.h"
#include "catalog/pg_ts_parser.h"
#include "commands/comment.h"
#include "commands/extension.h"
#include "fmgr.h"
#include "nodes/makefuncs.h"
#include "tsearch/ts_cache.h"
#include "tsearch/ts_public.h"
#include "utils/fmgroids.h"
#include "utils/lsyscache.h"
#include "utils/syscache.h"
#include "distributed/commands.h"
#include "distributed/commands/utility_hook.h"
#include "distributed/deparser.h"
#include "distributed/listutils.h"
#include "distributed/metadata/distobject.h"
#include "distributed/metadata_sync.h"
#include "distributed/multi_executor.h"
#include "distributed/relation_access_tracking.h"
#include "distributed/worker_create_or_replace.h"
static List * GetDistributedTextSearchConfigurationNames(DropStmt *stmt);
static DefineStmt * GetTextSearchConfigDefineStmt(Oid tsconfigOid);
static List * GetTextSearchConfigCommentStmt(Oid tsconfigOid);
static List * get_ts_parser_namelist(Oid tsparserOid);
static List * GetTextSearchConfigMappingStmt(Oid tsconfigOid);
static List * GetTextSearchConfigOwnerStmts(Oid tsconfigOid);
static List * get_ts_dict_namelist(Oid tsdictOid);
static Oid get_ts_config_parser_oid(Oid tsconfigOid);
static char * get_ts_parser_tokentype_name(Oid parserOid, int32 tokentype);
/*
* PostprocessCreateTextSearchConfigurationStmt is called after the TEXT SEARCH
* CONFIGURATION has been created locally.
*
* Contrary to many other objects a text search configuration is often created as a copy
* of an existing configuration. After the copy there is no relation to the configuration
* that has been copied. This prevents our normal approach of ensuring dependencies to
* exist before forwarding a close ressemblance of the statement the user executed.
*
* Instead we recreate the object based on what we find in our own catalog, hence the
* amount of work we perform in the postprocess function, contrary to other objects.
*/
List *
PostprocessCreateTextSearchConfigurationStmt(Node *node, const char *queryString)
{
DefineStmt *stmt = castNode(DefineStmt, node);
Assert(stmt->kind == OBJECT_TSCONFIGURATION);
if (!ShouldPropagate())
{
return NIL;
}
/*
* If the create command is a part of a multi-statement transaction that is not in
* sequential mode, don't propagate. Instead we will rely on back filling.
*/
if (IsMultiStatementTransaction())
{
if (MultiShardConnectionType != SEQUENTIAL_CONNECTION)
{
return NIL;
}
}
EnsureCoordinator();
EnsureSequentialMode(OBJECT_TSCONFIGURATION);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt, false);
EnsureDependenciesExistOnAllNodes(&address);
/*
* TEXT SEARCH CONFIGURATION objects are more complex with their mappings and the
* possibility of copying from existing templates that we will require the idempotent
* recreation commands to be run for successful propagation
*/
List *commands = CreateTextSearchConfigDDLCommandsIdempotent(&address);
commands = lcons(DISABLE_DDL_PROPAGATION, commands);
commands = lappend(commands, ENABLE_DDL_PROPAGATION);
return NodeDDLTaskList(NON_COORDINATOR_NODES, commands);
}
List *
GetCreateTextSearchConfigStatements(const ObjectAddress *address)
{
Assert(address->classId == TSConfigRelationId);
List *stmts = NIL;
/* CREATE TEXT SEARCH CONFIGURATION ...*/
stmts = lappend(stmts, GetTextSearchConfigDefineStmt(address->objectId));
/* ALTER TEXT SEARCH CONFIGURATION ... OWNER TO ...*/
stmts = list_concat(stmts, GetTextSearchConfigOwnerStmts(address->objectId));
/* COMMENT ON TEXT SEARCH CONFIGURATION ... */
stmts = list_concat(stmts, GetTextSearchConfigCommentStmt(address->objectId));
/* ALTER TEXT SEARCH CONFIGURATION ... ADD MAPPING FOR ... WITH ... */
stmts = list_concat(stmts, GetTextSearchConfigMappingStmt(address->objectId));
return stmts;
}
/*
* CreateTextSearchConfigDDLCommandsIdempotent creates a list of ddl commands to recreate
* a TEXT SERACH CONFIGURATION object in an idempotent manner on workers.
*/
List *
CreateTextSearchConfigDDLCommandsIdempotent(const ObjectAddress *address)
{
List *stmts = GetCreateTextSearchConfigStatements(address);
List *sqls = DeparseTreeNodes(stmts);
return list_make1(WrapCreateOrReplaceList(sqls));
}
/*
* PreprocessDropTextSearchConfigurationStmt prepares the statements we need to send to
* the workers. After we have dropped the schema's locally they also got removed from
* pg_dist_object so it is important to do all distribution checks before the change is
* made locally.
*/
List *
PreprocessDropTextSearchConfigurationStmt(Node *node, const char *queryString,
ProcessUtilityContext processUtilityContext)
{
DropStmt *stmt = castNode(DropStmt, node);
Assert(stmt->removeType == OBJECT_TSCONFIGURATION);
if (!ShouldPropagate())
{
return NIL;
}
List *distributedObjects = GetDistributedTextSearchConfigurationNames(stmt);
if (list_length(distributedObjects) == 0)
{
/* no distributed objects to remove */
return NIL;
}
EnsureCoordinator();
EnsureSequentialMode(OBJECT_TSCONFIGURATION);
/*
* Temporarily replace the list of objects being dropped with only the list
* containing the distributed objects. After we have created the sql statement we
* restore the original list of objects to execute on locally.
*
* Because searchpaths on coordinator and workers might not be in sync we fully
* qualify the list before deparsing. This is safe because qualification doesn't
* change the original names in place, but insteads creates new ones.
*/
List *originalObjects = stmt->objects;
stmt->objects = distributedObjects;
QualifyTreeNode((Node *) stmt);
const char *dropStmtSql = DeparseTreeNode((Node *) stmt);
stmt->objects = originalObjects;
List *commands = list_make3(DISABLE_DDL_PROPAGATION,
(void *) dropStmtSql,
ENABLE_DDL_PROPAGATION);
return NodeDDLTaskList(NON_COORDINATOR_METADATA_NODES, commands);
}
/*
* GetDistributedTextSearchConfigurationNames iterates over all text search configurations
* dropped, and create a list containign all configurations that are distributed.
*/
static List *
GetDistributedTextSearchConfigurationNames(DropStmt *stmt)
{
List *objName = NULL;
List *distributedObjects = NIL;
foreach_ptr(objName, stmt->objects)
{
Oid tsconfigOid = get_ts_config_oid(objName, stmt->missing_ok);
if (!OidIsValid(tsconfigOid))
{
/* skip missing configuration names, they can't be dirstibuted */
continue;
}
ObjectAddress address = { 0 };
ObjectAddressSet(address, TSConfigRelationId, tsconfigOid);
if (!IsObjectDistributed(&address))
{
continue;
}
distributedObjects = lappend(distributedObjects, objName);
}
return distributedObjects;
}
/*
* PreprocessAlterTextSearchConfigurationStmt verifies if the configuration being altered
* is distributed in the cluster. If that is the case it will prepare the list of commands
* to send to the worker to apply the same changes remote.
*/
List *
PreprocessAlterTextSearchConfigurationStmt(Node *node, const char *queryString,
ProcessUtilityContext processUtilityContext)
{
AlterTSConfigurationStmt *stmt = castNode(AlterTSConfigurationStmt, node);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt, false);
if (!ShouldPropagateObject(&address))
{
return NIL;
}
EnsureCoordinator();
EnsureSequentialMode(OBJECT_TSCONFIGURATION);
QualifyTreeNode((Node *) stmt);
const char *alterStmtSql = DeparseTreeNode((Node *) stmt);
List *commands = list_make3(DISABLE_DDL_PROPAGATION,
(void *) alterStmtSql,
ENABLE_DDL_PROPAGATION);
return NodeDDLTaskList(NON_COORDINATOR_METADATA_NODES, commands);
}
/*
* PreprocessRenameTextSearchConfigurationStmt verifies if the configuration being altered
* is distributed in the cluster. If that is the case it will prepare the list of commands
* to send to the worker to apply the same changes remote.
*/
List *
PreprocessRenameTextSearchConfigurationStmt(Node *node, const char *queryString,
ProcessUtilityContext processUtilityContext)
{
RenameStmt *stmt = castNode(RenameStmt, node);
Assert(stmt->renameType == OBJECT_TSCONFIGURATION);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt, false);
if (!ShouldPropagateObject(&address))
{
return NIL;
}
EnsureCoordinator();
EnsureSequentialMode(OBJECT_TSCONFIGURATION);
QualifyTreeNode((Node *) stmt);
char *ddlCommand = DeparseTreeNode((Node *) stmt);
List *commands = list_make3(DISABLE_DDL_PROPAGATION,
(void *) ddlCommand,
ENABLE_DDL_PROPAGATION);
return NodeDDLTaskList(NON_COORDINATOR_METADATA_NODES, commands);
}
/*
* PreprocessAlterTextSearchConfigurationSchemaStmt verifies if the configuration being
* altered is distributed in the cluster. If that is the case it will prepare the list of
* commands to send to the worker to apply the same changes remote.
*/
List *
PreprocessAlterTextSearchConfigurationSchemaStmt(Node *node, const char *queryString,
ProcessUtilityContext
processUtilityContext)
{
AlterObjectSchemaStmt *stmt = castNode(AlterObjectSchemaStmt, node);
Assert(stmt->objectType == OBJECT_TSCONFIGURATION);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt,
stmt->missing_ok);
if (!ShouldPropagateObject(&address))
{
return NIL;
}
EnsureCoordinator();
EnsureSequentialMode(OBJECT_TSCONFIGURATION);
QualifyTreeNode((Node *) stmt);
const char *sql = DeparseTreeNode((Node *) stmt);
List *commands = list_make3(DISABLE_DDL_PROPAGATION,
(void *) sql,
ENABLE_DDL_PROPAGATION);
return NodeDDLTaskList(NON_COORDINATOR_METADATA_NODES, commands);
}
/*
* PostprocessAlterTextSearchConfigurationSchemaStmt is invoked after the schema has been
* changed locally. Since changing the schema could result in new dependencies being found
* for this object we re-ensure all the dependencies for the configuration do exist. This
* is solely to propagate the new schema (and all its dependencies) if it was not already
* distributed in the cluster.
*/
List *
PostprocessAlterTextSearchConfigurationSchemaStmt(Node *node, const char *queryString)
{
AlterObjectSchemaStmt *stmt = castNode(AlterObjectSchemaStmt, node);
Assert(stmt->objectType == OBJECT_TSCONFIGURATION);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt,
stmt->missing_ok);
if (!ShouldPropagateObject(&address))
{
return NIL;
}
/* dependencies have changed (schema) let's ensure they exist */
EnsureDependenciesExistOnAllNodes(&address);
return NIL;
}
/*
* PreprocessTextSearchConfigurationCommentStmt propagates any comment on a distributed
* configuration to the workers. Since comments for configurations are promenently shown
* when listing all text search configurations this is purely a cosmetic thing when
* running in MX.
*/
List *
PreprocessTextSearchConfigurationCommentStmt(Node *node, const char *queryString,
ProcessUtilityContext processUtilityContext)
{
CommentStmt *stmt = castNode(CommentStmt, node);
Assert(stmt->objtype == OBJECT_TSCONFIGURATION);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt, false);
if (!ShouldPropagateObject(&address))
{
return NIL;
}
EnsureCoordinator();
EnsureSequentialMode(OBJECT_TSCONFIGURATION);
QualifyTreeNode((Node *) stmt);
const char *sql = DeparseTreeNode((Node *) stmt);
List *commands = list_make3(DISABLE_DDL_PROPAGATION,
(void *) sql,
ENABLE_DDL_PROPAGATION);
return NodeDDLTaskList(NON_COORDINATOR_METADATA_NODES, commands);
}
/*
* PreprocessAlterTextSearchConfigurationOwnerStmt verifies if the configuration being
* altered is distributed in the cluster. If that is the case it will prepare the list of
* commands to send to the worker to apply the same changes remote.
*/
List *
PreprocessAlterTextSearchConfigurationOwnerStmt(Node *node, const char *queryString,
ProcessUtilityContext
processUtilityContext)
{
AlterOwnerStmt *stmt = castNode(AlterOwnerStmt, node);
Assert(stmt->objectType == OBJECT_TSCONFIGURATION);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt, false);
if (!ShouldPropagateObject(&address))
{
return NIL;
}
EnsureCoordinator();
EnsureSequentialMode(OBJECT_TSCONFIGURATION);
QualifyTreeNode((Node *) stmt);
char *sql = DeparseTreeNode((Node *) stmt);
List *commands = list_make3(DISABLE_DDL_PROPAGATION,
(void *) sql,
ENABLE_DDL_PROPAGATION);
return NodeDDLTaskList(NON_COORDINATOR_NODES, commands);
}
/*
* PostprocessAlterTextSearchConfigurationOwnerStmt is invoked after the owner has been
* changed locally. Since changing the owner could result in new dependencies being found
* for this object we re-ensure all the dependencies for the configuration do exist. This
* is solely to propagate the new owner (and all its dependencies) if it was not already
* distributed in the cluster.
*/
List *
PostprocessAlterTextSearchConfigurationOwnerStmt(Node *node, const char *queryString)
{
AlterOwnerStmt *stmt = castNode(AlterOwnerStmt, node);
Assert(stmt->objectType == OBJECT_TSCONFIGURATION);
ObjectAddress address = GetObjectAddressFromParseTree((Node *) stmt, false);
if (!ShouldPropagateObject(&address))
{
return NIL;
}
/* dependencies have changed (owner) let's ensure they exist */
EnsureDependenciesExistOnAllNodes(&address);
return NIL;
}
/*
* GetTextSearchConfigDefineStmt returns the DefineStmt for a TEXT SEARCH CONFIGURATION
* based on the configuration as defined in the catalog identified by tsconfigOid.
*
* This statement will only contain the parser, as all other properties for text search
* configurations are stored as mappings in a different catalog.
*/
static DefineStmt *
GetTextSearchConfigDefineStmt(Oid tsconfigOid)
{
HeapTuple tup = SearchSysCache1(TSCONFIGOID, ObjectIdGetDatum(tsconfigOid));
if (!HeapTupleIsValid(tup)) /* should not happen */
{
elog(ERROR, "cache lookup failed for text search configuration %u",
tsconfigOid);
}
Form_pg_ts_config config = (Form_pg_ts_config) GETSTRUCT(tup);
DefineStmt *stmt = makeNode(DefineStmt);
stmt->kind = OBJECT_TSCONFIGURATION;
stmt->defnames = get_ts_config_namelist(tsconfigOid);
List *parserNameList = get_ts_parser_namelist(config->cfgparser);
TypeName *parserTypeName = makeTypeNameFromNameList(parserNameList);
stmt->definition = list_make1(makeDefElem("parser", (Node *) parserTypeName, -1));
ReleaseSysCache(tup);
return stmt;
}
/*
* GetTextSearchConfigCommentStmt returns a list containing all entries to recreate a
* comment on the configuration identified by tsconfigOid. The list could be empty if
* there is no comment on a configuration.
*
* The reason for a list is for easy use when building a list of all statements to invoke
* to recreate the text search configuration. An empty list can easily be concatinated
* without inspection, contrary to a NULL ptr if we would return the CommentStmt struct.
*/
static List *
GetTextSearchConfigCommentStmt(Oid tsconfigOid)
{
char *comment = GetComment(tsconfigOid, TSConfigRelationId, 0);
if (!comment)
{
return NIL;
}
CommentStmt *stmt = makeNode(CommentStmt);
stmt->objtype = OBJECT_TSCONFIGURATION;
stmt->object = (Node *) get_ts_config_namelist(tsconfigOid);
stmt->comment = comment;
return list_make1(stmt);
}
/*
* GetTextSearchConfigMappingStmt returns a list of all mappings from token_types to
* dictionaries configured on a text search configuration identified by tsconfigOid.
*
* Many mappings can exist on a configuration which all require their own statement to
* recreate.
*/
static List *
GetTextSearchConfigMappingStmt(Oid tsconfigOid)
{
ScanKeyData mapskey = { 0 };
/* mapcfg = tsconfigOid */
ScanKeyInit(&mapskey,
Anum_pg_ts_config_map_mapcfg,
BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(tsconfigOid));
Relation maprel = table_open(TSConfigMapRelationId, AccessShareLock);
Relation mapidx = index_open(TSConfigMapIndexId, AccessShareLock);
SysScanDesc mapscan = systable_beginscan_ordered(maprel, mapidx, NULL, 1, &mapskey);
List *stmts = NIL;
AlterTSConfigurationStmt *stmt = NULL;
/*
* We iterate the config mappings on the index order filtered by mapcfg. Meaning we
* get equal maptokentype's in 1 run. By comparing the current tokentype to the last
* we know when we can create a new stmt and append the previous constructed one to
* the list.
*/
int lastTokType = -1;
/*
* We read all mappings filtered by config id, hence we only need to load the name
* once and can reuse for every statement.
*/
List *configName = get_ts_config_namelist(tsconfigOid);
Oid parserOid = get_ts_config_parser_oid(tsconfigOid);
HeapTuple maptup = NULL;
while ((maptup = systable_getnext_ordered(mapscan, ForwardScanDirection)) != NULL)
{
Form_pg_ts_config_map cfgmap = (Form_pg_ts_config_map) GETSTRUCT(maptup);
if (lastTokType != cfgmap->maptokentype)
{
/* creating a new statement, appending the previous one (if existing) */
if (stmt != NULL)
{
stmts = lappend(stmts, stmt);
}
stmt = makeNode(AlterTSConfigurationStmt);
stmt->cfgname = configName;
stmt->kind = ALTER_TSCONFIG_ADD_MAPPING;
stmt->tokentype = list_make1(makeString(
get_ts_parser_tokentype_name(parserOid,
cfgmap->
maptokentype)));
lastTokType = cfgmap->maptokentype;
}
stmt->dicts = lappend(stmt->dicts, get_ts_dict_namelist(cfgmap->mapdict));
}
/*
* If we have ran atleast 1 iteration above we have the last stmt not added to the
* stmts list.
*/
if (stmt != NULL)
{
stmts = lappend(stmts, stmt);
stmt = NULL;
}
systable_endscan_ordered(mapscan);
index_close(mapidx, NoLock);
table_close(maprel, NoLock);
return stmts;
}
/*
* GetTextSearchConfigOwnerStmts returns a potentially empty list of statements to change
* the ownership of a TEXT SEARCH CONFIGURATION object.
*
* The list is for convenienve when building a full list of statements to recreate the
* configuration.
*/
static List *
GetTextSearchConfigOwnerStmts(Oid tsconfigOid)
{
HeapTuple tup = SearchSysCache1(TSCONFIGOID, ObjectIdGetDatum(tsconfigOid));
if (!HeapTupleIsValid(tup)) /* should not happen */
{
elog(ERROR, "cache lookup failed for text search configuration %u",
tsconfigOid);
}
Form_pg_ts_config config = (Form_pg_ts_config) GETSTRUCT(tup);
AlterOwnerStmt *stmt = makeNode(AlterOwnerStmt);
stmt->objectType = OBJECT_TSCONFIGURATION;
stmt->object = (Node *) get_ts_config_namelist(tsconfigOid);
stmt->newowner = GetRoleSpecObjectForUser(config->cfgowner);
ReleaseSysCache(tup);
return list_make1(stmt);
}
/*
* get_ts_config_namelist based on the tsconfigOid this function creates the namelist that
* identifies the configuration in a fully qualified manner, irregardless of the schema
* existing on the search_path.
*/
List *
get_ts_config_namelist(Oid tsconfigOid)
{
HeapTuple tup = SearchSysCache1(TSCONFIGOID, ObjectIdGetDatum(tsconfigOid));
if (!HeapTupleIsValid(tup)) /* should not happen */
{
elog(ERROR, "cache lookup failed for text search configuration %u",
tsconfigOid);
}
Form_pg_ts_config config = (Form_pg_ts_config) GETSTRUCT(tup);
char *schema = get_namespace_name(config->cfgnamespace);
char *configName = pstrdup(NameStr(config->cfgname));
List *names = list_make2(makeString(schema), makeString(configName));
ReleaseSysCache(tup);
return names;
}
/*
* get_ts_dict_namelist based on the tsdictOid this function creates the namelist that
* identifies the dictionary in a fully qualified manner, irregardless of the schema
* existing on the search_path.
*/
static List *
get_ts_dict_namelist(Oid tsdictOid)
{
HeapTuple tup = SearchSysCache1(TSDICTOID, ObjectIdGetDatum(tsdictOid));
if (!HeapTupleIsValid(tup)) /* should not happen */
{
elog(ERROR, "cache lookup failed for text search dictionary %u", tsdictOid);
}
Form_pg_ts_dict dict = (Form_pg_ts_dict) GETSTRUCT(tup);
char *schema = get_namespace_name(dict->dictnamespace);
char *dictName = pstrdup(NameStr(dict->dictname));
List *names = list_make2(makeString(schema), makeString(dictName));
ReleaseSysCache(tup);
return names;
}
/*
* get_ts_config_parser_oid based on the tsconfigOid this function returns the Oid of the
* parser used in the configuration.
*/
static Oid
get_ts_config_parser_oid(Oid tsconfigOid)
{
HeapTuple tup = SearchSysCache1(TSCONFIGOID, ObjectIdGetDatum(tsconfigOid));
if (!HeapTupleIsValid(tup)) /* should not happen */
{
elog(ERROR, "cache lookup failed for text search configuration %u", tsconfigOid);
}
Form_pg_ts_config config = (Form_pg_ts_config) GETSTRUCT(tup);
Oid parserOid = config->cfgparser;
ReleaseSysCache(tup);
return parserOid;
}
/*
* get_ts_parser_tokentype_name returns the name of the token as known to the parser by
* its tokentype identifier. The parser used to resolve the token name is identified by
* parserOid and should be the same that emitted the tokentype to begin with.
*/
static char *
get_ts_parser_tokentype_name(Oid parserOid, int32 tokentype)
{
TSParserCacheEntry *parserCache = lookup_ts_parser_cache(parserOid);
if (!OidIsValid(parserCache->lextypeOid))
{
elog(ERROR, "method lextype isn't defined for text search parser %u", parserOid);
}
/* take lextypes from parser */
LexDescr *tokenlist = (LexDescr *) DatumGetPointer(
OidFunctionCall1(parserCache->lextypeOid, Int32GetDatum(0)));
/* and find the one with lexid = tokentype */
int tokenIndex = 0;
while (tokenlist && tokenlist[tokenIndex].lexid)
{
if (tokenlist[tokenIndex].lexid == tokentype)
{
return pstrdup(tokenlist[tokenIndex].alias);
}
tokenIndex++;
}
/* we haven't found the token */
ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("token type \"%d\" does not exist in parser", tokentype)));
}
/*
* get_ts_parser_namelist based on the tsparserOid this function creates the namelist that
* identifies the parser in a fully qualified manner, irregardless of the schema existing
* on the search_path.
*/
static List *
get_ts_parser_namelist(Oid tsparserOid)
{
HeapTuple tup = SearchSysCache1(TSPARSEROID, ObjectIdGetDatum(tsparserOid));
if (!HeapTupleIsValid(tup)) /* should not happen */
{
elog(ERROR, "cache lookup failed for text search parser %u",
tsparserOid);
}
Form_pg_ts_parser parser = (Form_pg_ts_parser) GETSTRUCT(tup);
char *schema = get_namespace_name(parser->prsnamespace);
char *parserName = pstrdup(NameStr(parser->prsname));
List *names = list_make2(makeString(schema), makeString(parserName));
ReleaseSysCache(tup);
return names;
}
/*
* CreateTextSearchConfigurationObjectAddress resolves the ObjectAddress for the object
* being created. If missing_pk is false the function will error, explaining to the user
* the text search configuration described in the statement doesn't exist.
*/
ObjectAddress
CreateTextSearchConfigurationObjectAddress(Node *node, bool missing_ok)
{
DefineStmt *stmt = castNode(DefineStmt, node);
Assert(stmt->kind == OBJECT_TSCONFIGURATION);
Oid objid = get_ts_config_oid(stmt->defnames, missing_ok);
ObjectAddress address = { 0 };
ObjectAddressSet(address, TSConfigRelationId, objid);
return address;
}
/*
* RenameTextSearchConfigurationStmtObjectAddress resolves the ObjectAddress for the TEXT
* SEARCH CONFIGURATION being renamed. Optionally errors if the configuration does not
* exist based on the missing_ok flag passed in by the caller.
*/
ObjectAddress
RenameTextSearchConfigurationStmtObjectAddress(Node *node, bool missing_ok)
{
RenameStmt *stmt = castNode(RenameStmt, node);
Assert(stmt->renameType == OBJECT_TSCONFIGURATION);
Oid objid = get_ts_config_oid(castNode(List, stmt->object), missing_ok);
ObjectAddress address = { 0 };
ObjectAddressSet(address, TSConfigRelationId, objid);
return address;
}
/*
* AlterTextSearchConfigurationStmtObjectAddress resolves the ObjectAddress for the TEXT
* SEARCH CONFIGURATION being altered. Optionally errors if the configuration does not
* exist based on the missing_ok flag passed in by the caller.
*/
ObjectAddress
AlterTextSearchConfigurationStmtObjectAddress(Node *node, bool missing_ok)
{
AlterTSConfigurationStmt *stmt = castNode(AlterTSConfigurationStmt, node);
Oid objid = get_ts_config_oid(stmt->cfgname, missing_ok);
ObjectAddress address = { 0 };
ObjectAddressSet(address, TSConfigRelationId, objid);
return address;
}
/*
* AlterTextSearchConfigurationSchemaStmtObjectAddress resolves the ObjectAddress for the
* TEXT SEARCH CONFIGURATION being moved to a different schema. Optionally errors if the
* configuration does not exist based on the missing_ok flag passed in by the caller.
*
* This can be called, either before or after the move of schema has been executed, hence
* the triple checking before the error might be thrown. Errors for non-existing schema's
* in edgecases will be raised by postgres while executing the move.
*/
ObjectAddress
AlterTextSearchConfigurationSchemaStmtObjectAddress(Node *node, bool missing_ok)
{
AlterObjectSchemaStmt *stmt = castNode(AlterObjectSchemaStmt, node);
Assert(stmt->objectType == OBJECT_TSCONFIGURATION);
Oid objid = get_ts_config_oid(castNode(List, stmt->object), true);
if (!OidIsValid(objid))
{
/*
* couldn't find the text search configuration, might have already been moved to
* the new schema, we construct a new sequence name that uses the new schema to
* search in.
*/
char *schemaname = NULL;
char *config_name = NULL;
DeconstructQualifiedName(castNode(List, stmt->object), &schemaname, &config_name);
char *newSchemaName = stmt->newschema;
List *names = list_make2(makeString(newSchemaName), makeString(config_name));
objid = get_ts_config_oid(names, true);
if (!missing_ok && !OidIsValid(objid))
{
/*
* if the text search config id is still invalid we couldn't find it, error
* with the same message postgres would error with if missing_ok is false
* (not ok to miss)
*/
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_OBJECT),
errmsg("text search configuration \"%s\" does not exist",
NameListToString(castNode(List, stmt->object)))));
}
}
ObjectAddress sequenceAddress = { 0 };
ObjectAddressSet(sequenceAddress, TSConfigRelationId, objid);
return sequenceAddress;
}
/*
* TextSearchConfigurationCommentObjectAddress resolves the ObjectAddress for the TEXT
* SEARCH CONFIGURATION on which the comment is placed. Optionally errors if the
* configuration does not exist based on the missing_ok flag passed in by the caller.
*/
ObjectAddress
TextSearchConfigurationCommentObjectAddress(Node *node, bool missing_ok)
{
CommentStmt *stmt = castNode(CommentStmt, node);
Assert(stmt->objtype == OBJECT_TSCONFIGURATION);
Oid objid = get_ts_config_oid(castNode(List, stmt->object), missing_ok);
ObjectAddress address = { 0 };
ObjectAddressSet(address, TSConfigRelationId, objid);
return address;
}
/*
* AlterTextSearchConfigurationOwnerObjectAddress resolves the ObjectAddress for the TEXT
* SEARCH CONFIGURATION for which the owner is changed. Optionally errors if the
* configuration does not exist based on the missing_ok flag passed in by the caller.
*/
ObjectAddress
AlterTextSearchConfigurationOwnerObjectAddress(Node *node, bool missing_ok)
{
AlterOwnerStmt *stmt = castNode(AlterOwnerStmt, node);
Relation relation = NULL;
Assert(stmt->objectType == OBJECT_TSCONFIGURATION);
return get_object_address(stmt->objectType, stmt->object, &relation, AccessShareLock,
missing_ok);
}
/*
* GenerateBackupNameForTextSearchConfiguration generates a safe name that is not in use
* already that can be used to rename an existing TEXT SEARCH CONFIGURATION to allow the
* configuration with a specific name to be created, even if this would not have been
* possible due to name collisions.
*/
char *
GenerateBackupNameForTextSearchConfiguration(const ObjectAddress *address)
{
Assert(address->classId == TSConfigRelationId);
List *names = get_ts_config_namelist(address->objectId);
RangeVar *rel = makeRangeVarFromNameList(names);
char *newName = palloc0(NAMEDATALEN);
char suffix[NAMEDATALEN] = { 0 };
char *baseName = rel->relname;
int baseLength = strlen(baseName);
int count = 0;
while (true)
{
int suffixLength = SafeSnprintf(suffix, NAMEDATALEN - 1, "(citus_backup_%d)",
count);
/* trim the base name at the end to leave space for the suffix and trailing \0 */
baseLength = Min(baseLength, NAMEDATALEN - suffixLength - 1);
/* clear newName before copying the potentially trimmed baseName and suffix */
memset(newName, 0, NAMEDATALEN);
strncpy_s(newName, NAMEDATALEN, baseName, baseLength);
strncpy_s(newName + baseLength, NAMEDATALEN - baseLength, suffix,
suffixLength);
rel->relname = newName;
List *newNameList = MakeNameListFromRangeVar(rel);
Oid tsconfigOid = get_ts_config_oid(newNameList, true);
if (!OidIsValid(tsconfigOid))
{
return newName;
}
count++;
}
}

14
src/backend/distributed/commands/truncate.c
View File

@ -267,13 +267,17 @@ ErrorIfUnsupportedTruncateStmt(TruncateStmt *truncateStatement)
ErrorIfIllegallyChangingKnownShard(relationId);
if (IsCitusTable(relationId) && IsForeignTable(relationId))
/*
* We allow truncating foreign tables that are added to metadata
* only on the coordinator, as user mappings are not propagated.
*/
if (IsForeignTable(relationId) &&
IsCitusTableType(relationId, CITUS_LOCAL_TABLE) &&
!IsCoordinator())
{
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("truncating distributed foreign tables is "
"currently unsupported"),
errhint("Consider undistributing table before TRUNCATE, "
"and then distribute or add to metadata again")));
errmsg("truncating foreign tables that are added to metadata "
"can only be excuted on the coordinator")));
}
}
}

18
src/backend/distributed/deparser/deparse.c
View File

@ -17,6 +17,7 @@
#include "distributed/commands.h"
#include "distributed/deparser.h"
#include "distributed/listutils.h"
/*
* DeparseTreeNode aims to be the inverse of postgres' ParseTreeNode. Currently with
@ -35,3 +36,20 @@ DeparseTreeNode(Node *stmt)
return ops->deparse(stmt);
}
/*
* DeparseTreeNodes deparses all stmts in the list from the statement datastructure into
* sql statements.
*/
List *
DeparseTreeNodes(List *stmts)
{
List *sqls = NIL;
Node *stmt = NULL;
foreach_ptr(stmt, stmts)
{
sqls = lappend(sqls, DeparseTreeNode(stmt));
}
return sqls;
}

377
src/backend/distributed/deparser/deparse_text_search.c Normal file
View File

@ -0,0 +1,377 @@
/*-------------------------------------------------------------------------
*
* deparse_text_search.c
* All routines to deparse text search statements.
* This file contains all entry points specific for text search statement deparsing.
*
* Copyright (c) Citus Data, Inc.
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "catalog/namespace.h"
#include "utils/builtins.h"
#include "distributed/citus_ruleutils.h"
#include "distributed/deparser.h"
#include "distributed/listutils.h"
static void AppendDefElemList(StringInfo buf, List *defelms);
static void AppendStringInfoTokentypeList(StringInfo buf, List *tokentypes);
static void AppendStringInfoDictnames(StringInfo buf, List *dicts);
/*
* DeparseCreateTextSearchStmt returns the sql for a DefineStmt defining a TEXT SEARCH
* CONFIGURATION
*
* Although the syntax is mutually exclusive on the two arguments that can be passed in
* the deparser will syntactically correct multiple definitions if provided. *
*/
char *
DeparseCreateTextSearchStmt(Node *node)
{
DefineStmt *stmt = castNode(DefineStmt, node);
StringInfoData buf = { 0 };
initStringInfo(&buf);
const char *identifier = NameListToQuotedString(stmt->defnames);
appendStringInfo(&buf, "CREATE TEXT SEARCH CONFIGURATION %s ", identifier);
appendStringInfoString(&buf, "(");
AppendDefElemList(&buf, stmt->definition);
appendStringInfoString(&buf, ");");
return buf.data;
}
/*
* AppendDefElemList specialization to append a comma separated list of definitions to a
* define statement.
*
* Currently only supports String and TypeName entries. Will error on others.
*/
static void
AppendDefElemList(StringInfo buf, List *defelems)
{
DefElem *defelem = NULL;
bool first = true;
foreach_ptr(defelem, defelems)
{
if (!first)
{
appendStringInfoString(buf, ", ");
}
first = false;
/* extract identifier from defelem */
const char *identifier = NULL;
switch (nodeTag(defelem->arg))
{
case T_String:
{
identifier = quote_identifier(strVal(defelem->arg));
break;
}
case T_TypeName:
{
TypeName *typeName = castNode(TypeName, defelem->arg);
identifier = NameListToQuotedString(typeName->names);
break;
}
default:
{
ereport(ERROR, (errmsg("unexpected argument during deparsing of "
"TEXT SEARCH CONFIGURATION definition")));
}
}
/* stringify */
appendStringInfo(buf, "%s = %s", defelem->defname, identifier);
}
}
/*
* DeparseDropTextSearchConfigurationStmt returns the sql representation for a DROP TEXT
* SEARCH CONFIGURATION ... statment. Supports dropping multiple configurations at once.
*/
char *
DeparseDropTextSearchConfigurationStmt(Node *node)
{
DropStmt *stmt = castNode(DropStmt, node);
Assert(stmt->removeType == OBJECT_TSCONFIGURATION);
StringInfoData buf = { 0 };
initStringInfo(&buf);
appendStringInfoString(&buf, "DROP TEXT SEARCH CONFIGURATION ");
List *nameList = NIL;
bool first = true;
foreach_ptr(nameList, stmt->objects)
{
if (!first)
{
appendStringInfoString(&buf, ", ");
}
first = false;
appendStringInfoString(&buf, NameListToQuotedString(nameList));
}
if (stmt->behavior == DROP_CASCADE)
{
appendStringInfoString(&buf, " CASCADE");
}
appendStringInfoString(&buf, ";");
return buf.data;
}
/*
* DeparseRenameTextSearchConfigurationStmt returns the sql representation of a ALTER TEXT
* SEARCH CONFIGURATION ... RENAME TO ... statement.
*/
char *
DeparseRenameTextSearchConfigurationStmt(Node *node)
{
RenameStmt *stmt = castNode(RenameStmt, node);
Assert(stmt->renameType == OBJECT_TSCONFIGURATION);
StringInfoData buf = { 0 };
initStringInfo(&buf);
char *identifier = NameListToQuotedString(castNode(List, stmt->object));
appendStringInfo(&buf, "ALTER TEXT SEARCH CONFIGURATION %s RENAME TO %s;",
identifier, quote_identifier(stmt->newname));
return buf.data;
}
/*
* DeparseAlterTextSearchConfigurationStmt returns the ql representation of any generic
* ALTER TEXT SEARCH CONFIGURATION .... statement. The statements supported include:
* - ALTER TEXT SEARCH CONFIGURATIONS ... ADD MAPPING FOR [, ...] WITH [, ...]
* - ALTER TEXT SEARCH CONFIGURATIONS ... ALTER MAPPING FOR [, ...] WITH [, ...]
* - ALTER TEXT SEARCH CONFIGURATIONS ... ALTER MAPPING REPLACE ... WITH ...
* - ALTER TEXT SEARCH CONFIGURATIONS ... ALTER MAPPING FOR [, ...] REPLACE ... WITH ...
* - ALTER TEXT SEARCH CONFIGURATIONS ... DROP MAPPING [ IF EXISTS ] FOR ...
*/
char *
DeparseAlterTextSearchConfigurationStmt(Node *node)
{
AlterTSConfigurationStmt *stmt = castNode(AlterTSConfigurationStmt, node);
StringInfoData buf = { 0 };
initStringInfo(&buf);
char *identifier = NameListToQuotedString(castNode(List, stmt->cfgname));
appendStringInfo(&buf, "ALTER TEXT SEARCH CONFIGURATION %s", identifier);
switch (stmt->kind)
{
case ALTER_TSCONFIG_ADD_MAPPING:
{
appendStringInfoString(&buf, " ADD MAPPING FOR ");
AppendStringInfoTokentypeList(&buf, stmt->tokentype);
appendStringInfoString(&buf, " WITH ");
AppendStringInfoDictnames(&buf, stmt->dicts);
break;
}
case ALTER_TSCONFIG_ALTER_MAPPING_FOR_TOKEN:
{
appendStringInfoString(&buf, " ALTER MAPPING FOR ");
AppendStringInfoTokentypeList(&buf, stmt->tokentype);
appendStringInfoString(&buf, " WITH ");
AppendStringInfoDictnames(&buf, stmt->dicts);
break;
}
case ALTER_TSCONFIG_REPLACE_DICT:
case ALTER_TSCONFIG_REPLACE_DICT_FOR_TOKEN:
{
appendStringInfoString(&buf, " ALTER MAPPING");
if (list_length(stmt->tokentype) > 0)
{
appendStringInfoString(&buf, " FOR ");
AppendStringInfoTokentypeList(&buf, stmt->tokentype);
}
if (list_length(stmt->dicts) != 2)
{
elog(ERROR, "unexpected number of dictionaries while deparsing ALTER "
"TEXT SEARCH CONFIGURATION ... ALTER MAPPING [FOR ...] REPLACE "
"statement.");
}
appendStringInfo(&buf, " REPLACE %s",
NameListToQuotedString(linitial(stmt->dicts)));
appendStringInfo(&buf, " WITH %s",
NameListToQuotedString(lsecond(stmt->dicts)));
break;
}
case ALTER_TSCONFIG_DROP_MAPPING:
{
appendStringInfoString(&buf, " DROP MAPPING");
if (stmt->missing_ok)
{
appendStringInfoString(&buf, " IF EXISTS");
}
appendStringInfoString(&buf, " FOR ");
AppendStringInfoTokentypeList(&buf, stmt->tokentype);
break;
}
default:
{
elog(ERROR, "unable to deparse unsupported ALTER TEXT SEARCH STATEMENT");
}
}
appendStringInfoString(&buf, ";");
return buf.data;
}
/*
* DeparseAlterTextSearchConfigurationSchemaStmt returns the sql statement representing
* ALTER TEXT SEARCH CONFIGURATION ... SET SCHEMA ... statements.
*/
char *
DeparseAlterTextSearchConfigurationSchemaStmt(Node *node)
{
AlterObjectSchemaStmt *stmt = castNode(AlterObjectSchemaStmt, node);
Assert(stmt->objectType == OBJECT_TSCONFIGURATION);
StringInfoData buf = { 0 };
initStringInfo(&buf);
appendStringInfo(&buf, "ALTER TEXT SEARCH CONFIGURATION %s SET SCHEMA %s;",
NameListToQuotedString(castNode(List, stmt->object)),
quote_identifier(stmt->newschema));
return buf.data;
}
/*
* DeparseTextSearchConfigurationCommentStmt returns the sql statement representing
* COMMENT ON TEXT SEARCH CONFIGURATION ... IS ...
*/
char *
DeparseTextSearchConfigurationCommentStmt(Node *node)
{
CommentStmt *stmt = castNode(CommentStmt, node);
Assert(stmt->objtype == OBJECT_TSCONFIGURATION);
StringInfoData buf = { 0 };
initStringInfo(&buf);
appendStringInfo(&buf, "COMMENT ON TEXT SEARCH CONFIGURATION %s IS ",
NameListToQuotedString(castNode(List, stmt->object)));
if (stmt->comment == NULL)
{
appendStringInfoString(&buf, "NULL");
}
else
{
appendStringInfoString(&buf, quote_literal_cstr(stmt->comment));
}
appendStringInfoString(&buf, ";");
return buf.data;
}
/*
* AppendStringInfoTokentypeList specializes in adding a comma separated list of
* token_tyoe's to TEXT SEARCH CONFIGURATION commands
*/
static void
AppendStringInfoTokentypeList(StringInfo buf, List *tokentypes)
{
Value *tokentype = NULL;
bool first = true;
foreach_ptr(tokentype, tokentypes)
{
if (nodeTag(tokentype) != T_String)
{
elog(ERROR,
"unexpected tokentype for deparsing in text search configuration");
}
if (!first)
{
appendStringInfoString(buf, ", ");
}
first = false;
appendStringInfoString(buf, strVal(tokentype));
}
}
/*
* AppendStringInfoDictnames specializes in appending a comma separated list of
* dictionaries to TEXT SEARCH CONFIGURATION commands.
*/
static void
AppendStringInfoDictnames(StringInfo buf, List *dicts)
{
List *dictNames = NIL;
bool first = true;
foreach_ptr(dictNames, dicts)
{
if (!first)
{
appendStringInfoString(buf, ", ");
}
first = false;
char *dictIdentifier = NameListToQuotedString(dictNames);
appendStringInfoString(buf, dictIdentifier);
}
}
/*
* DeparseAlterTextSearchConfigurationOwnerStmt returns the sql statement representing
* ALTER TEXT SEARCH CONFIGURATION ... ONWER TO ... commands.
*/
char *
DeparseAlterTextSearchConfigurationOwnerStmt(Node *node)
{
AlterOwnerStmt *stmt = castNode(AlterOwnerStmt, node);
Assert(stmt->objectType == OBJECT_TSCONFIGURATION);
StringInfoData buf = { 0 };
initStringInfo(&buf);
appendStringInfo(&buf, "ALTER TEXT SEARCH CONFIGURATION %s OWNER TO %s;",
NameListToQuotedString(castNode(List, stmt->object)),
RoleSpecString(stmt->newowner, true));
return buf.data;
}

278
src/backend/distributed/deparser/qualify_text_search_stmts.c Normal file
View File

@ -0,0 +1,278 @@
/*-------------------------------------------------------------------------
*
* qualify_text_search_stmts.c
* Functions specialized in fully qualifying all text search statements. These
* functions are dispatched from qualify.c
*
* Fully qualifying text search statements consists of adding the schema name
* to the subject of the types as well as any other branch of the parsetree.
*
* Goal would be that the deparser functions for these statements can
* serialize the statement without any external lookups.
*
* Copyright (c) Citus Data, Inc.
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "access/htup_details.h"
#include "catalog/namespace.h"
#include "catalog/pg_ts_config.h"
#include "catalog/pg_ts_dict.h"
#include "utils/lsyscache.h"
#include "utils/syscache.h"
#include "distributed/deparser.h"
#include "distributed/listutils.h"
static Oid get_ts_config_namespace(Oid tsconfigOid);
static Oid get_ts_dict_namespace(Oid tsdictOid);
/*
* QualifyDropTextSearchConfigurationStmt adds any missing schema names to text search
* configurations being dropped. All configurations are expected to exists before fully
* qualifying the statement. Errors will be raised for objects not existing. Non-existing
* objects are expected to not be distributed.
*/
void
QualifyDropTextSearchConfigurationStmt(Node *node)
{
DropStmt *stmt = castNode(DropStmt, node);
Assert(stmt->removeType == OBJECT_TSCONFIGURATION);
List *qualifiedObjects = NIL;
List *objName = NIL;
foreach_ptr(objName, stmt->objects)
{
char *schemaName = NULL;
char *tsconfigName = NULL;
DeconstructQualifiedName(objName, &schemaName, &tsconfigName);
if (!schemaName)
{
Oid tsconfigOid = get_ts_config_oid(objName, false);
Oid namespaceOid = get_ts_config_namespace(tsconfigOid);
schemaName = get_namespace_name(namespaceOid);
objName = list_make2(makeString(schemaName),
makeString(tsconfigName));
}
qualifiedObjects = lappend(qualifiedObjects, objName);
}
stmt->objects = qualifiedObjects;
}
/*
* QualifyAlterTextSearchConfigurationStmt adds the schema name (if missing) to the name
* of the text search configurations, as well as the dictionaries referenced.
*/
void
QualifyAlterTextSearchConfigurationStmt(Node *node)
{
AlterTSConfigurationStmt *stmt = castNode(AlterTSConfigurationStmt, node);
char *schemaName = NULL;
char *objName = NULL;
DeconstructQualifiedName(stmt->cfgname, &schemaName, &objName);
/* fully qualify the cfgname being altered */
if (!schemaName)
{
Oid tsconfigOid = get_ts_config_oid(stmt->cfgname, false);
Oid namespaceOid = get_ts_config_namespace(tsconfigOid);
schemaName = get_namespace_name(namespaceOid);
stmt->cfgname = list_make2(makeString(schemaName),
makeString(objName));
}
/* fully qualify the dicts */
bool useNewDicts = false;
List *dicts = NULL;
List *dictName = NIL;
foreach_ptr(dictName, stmt->dicts)
{
DeconstructQualifiedName(dictName, &schemaName, &objName);
/* fully qualify the cfgname being altered */
if (!schemaName)
{
Oid dictOid = get_ts_dict_oid(dictName, false);
Oid namespaceOid = get_ts_dict_namespace(dictOid);
schemaName = get_namespace_name(namespaceOid);
useNewDicts = true;
dictName = list_make2(makeString(schemaName), makeString(objName));
}
dicts = lappend(dicts, dictName);
}
if (useNewDicts)
{
/* swap original dicts with the new list */
stmt->dicts = dicts;
}
else
{
/* we don't use the new list, everything was already qualified, free-ing */
list_free(dicts);
}
}
/*
* QualifyRenameTextSearchConfigurationStmt adds the schema name (if missing) to the
* configuration being renamed. The new name will kept be without schema name since this
* command cannot be used to change the schema of a configuration.
*/
void
QualifyRenameTextSearchConfigurationStmt(Node *node)
{
RenameStmt *stmt = castNode(RenameStmt, node);
Assert(stmt->renameType == OBJECT_TSCONFIGURATION);
char *schemaName = NULL;
char *objName = NULL;
DeconstructQualifiedName(castNode(List, stmt->object), &schemaName, &objName);
/* fully qualify the cfgname being altered */
if (!schemaName)
{
Oid tsconfigOid = get_ts_config_oid(castNode(List, stmt->object), false);
Oid namespaceOid = get_ts_config_namespace(tsconfigOid);
schemaName = get_namespace_name(namespaceOid);
stmt->object = (Node *) list_make2(makeString(schemaName),
makeString(objName));
}
}
/*
* QualifyAlterTextSearchConfigurationSchemaStmt adds the schema name (if missing) for the
* text search being moved to a new schema.
*/
void
QualifyAlterTextSearchConfigurationSchemaStmt(Node *node)
{
AlterObjectSchemaStmt *stmt = castNode(AlterObjectSchemaStmt, node);
Assert(stmt->objectType == OBJECT_TSCONFIGURATION);
char *schemaName = NULL;
char *objName = NULL;
DeconstructQualifiedName(castNode(List, stmt->object), &schemaName, &objName);
if (!schemaName)
{
Oid tsconfigOid = get_ts_config_oid(castNode(List, stmt->object), false);
Oid namespaceOid = get_ts_config_namespace(tsconfigOid);
schemaName = get_namespace_name(namespaceOid);
stmt->object = (Node *) list_make2(makeString(schemaName),
makeString(objName));
}
}
/*
* QualifyTextSearchConfigurationCommentStmt adds the schema name (if missing) to the
* configuration name on which the comment is created.
*/
void
QualifyTextSearchConfigurationCommentStmt(Node *node)
{
CommentStmt *stmt = castNode(CommentStmt, node);
Assert(stmt->objtype == OBJECT_TSCONFIGURATION);
char *schemaName = NULL;
char *objName = NULL;
DeconstructQualifiedName(castNode(List, stmt->object), &schemaName, &objName);
if (!schemaName)
{
Oid tsconfigOid = get_ts_config_oid(castNode(List, stmt->object), false);
Oid namespaceOid = get_ts_config_namespace(tsconfigOid);
schemaName = get_namespace_name(namespaceOid);
stmt->object = (Node *) list_make2(makeString(schemaName),
makeString(objName));
}
}
/*
* QualifyAlterTextSearchConfigurationOwnerStmt adds the schema name (if missing) to the
* configuration for which the owner is changing.
*/
void
QualifyAlterTextSearchConfigurationOwnerStmt(Node *node)
{
AlterOwnerStmt *stmt = castNode(AlterOwnerStmt, node);
Assert(stmt->objectType == OBJECT_TSCONFIGURATION);
char *schemaName = NULL;
char *objName = NULL;
DeconstructQualifiedName(castNode(List, stmt->object), &schemaName, &objName);
if (!schemaName)
{
Oid tsconfigOid = get_ts_config_oid(castNode(List, stmt->object), false);
Oid namespaceOid = get_ts_config_namespace(tsconfigOid);
schemaName = get_namespace_name(namespaceOid);
stmt->object = (Node *) list_make2(makeString(schemaName),
makeString(objName));
}
}
/*
* get_ts_config_namespace returns the oid of the namespace which is housing the text
* search configuration identified by tsconfigOid.
*/
static Oid
get_ts_config_namespace(Oid tsconfigOid)
{
HeapTuple tup = SearchSysCache1(TSCONFIGOID, ObjectIdGetDatum(tsconfigOid));
if (HeapTupleIsValid(tup))
{
Form_pg_ts_config cfgform = (Form_pg_ts_config) GETSTRUCT(tup);
Oid namespaceOid = cfgform->cfgnamespace;
ReleaseSysCache(tup);
return namespaceOid;
}
return InvalidOid;
}
/*
* get_ts_dict_namespace returns the oid of the namespace which is housing the text
* search dictionary identified by tsdictOid.
*/
static Oid
get_ts_dict_namespace(Oid tsdictOid)
{
HeapTuple tup = SearchSysCache1(TSDICTOID, ObjectIdGetDatum(tsdictOid));
if (HeapTupleIsValid(tup))
{
Form_pg_ts_dict cfgform = (Form_pg_ts_dict) GETSTRUCT(tup);
Oid namespaceOid = cfgform->dictnamespace;
ReleaseSysCache(tup);
return namespaceOid;
}
return InvalidOid;
}

5
src/backend/distributed/executor/multi_executor.c
View File

@ -770,6 +770,11 @@ GetObjectTypeString(ObjectType objType)
return "schema";
}
case OBJECT_TSCONFIGURATION:
{
return "text search configuration";
}
case OBJECT_TYPE:
{
return "type";

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

@ -124,6 +124,7 @@ typedef struct ViewDependencyNode
static List * GetRelationSequenceDependencyList(Oid relationId);
static List * GetRelationTriggerFunctionDependencyList(Oid relationId);
static List * GetRelationStatsSchemaDependencyList(Oid relationId);
static List * GetRelationIndicesDependencyList(Oid relationId);
static DependencyDefinition * CreateObjectAddressDependencyDef(Oid classId, Oid objectId);
static List * CreateObjectAddressDependencyDefList(Oid classId, List *objectIdList);
static ObjectAddress DependencyDefinitionObjectAddress(DependencyDefinition *definition);
@ -670,6 +671,11 @@ SupportedDependencyByCitus(const ObjectAddress *address)
return true;
}
case OCLASS_TSCONFIG:
{
return true;
}
case OCLASS_TYPE:
{
switch (get_typtype(address->objectId))
@ -717,7 +723,8 @@ SupportedDependencyByCitus(const ObjectAddress *address)
relKind == RELKIND_RELATION ||
relKind == RELKIND_PARTITIONED_TABLE ||
relKind == RELKIND_FOREIGN_TABLE ||
relKind == RELKIND_SEQUENCE)
relKind == RELKIND_SEQUENCE ||
relKind == RELKIND_INDEX)
{
return true;
}
@ -1103,6 +1110,17 @@ ExpandCitusSupportedTypes(ObjectAddressCollector *collector, ObjectAddress targe
List *sequenceDependencyList = GetRelationSequenceDependencyList(relationId);
result = list_concat(result, sequenceDependencyList);
/*
* Tables could have indexes. Indexes themself could have dependencies that
* need to be propagated. eg. TEXT SEARCH CONFIGRUATIONS. Here we add the
* addresses of all indices to the list of objects to vist, as to make sure we
* create all objects required by the indices before we create the table
* including indices.
*/
List *indexDependencyList = GetRelationIndicesDependencyList(relationId);
result = list_concat(result, indexDependencyList);
}
default:
@ -1146,6 +1164,28 @@ GetRelationStatsSchemaDependencyList(Oid relationId)
}
/*
* CollectIndexOids implements PGIndexProcessor to create a list of all index oids
*/
static void
CollectIndexOids(Form_pg_index formPgIndex, List **oids, int flags)
{
*oids = lappend_oid(*oids, formPgIndex->indexrelid);
}
/*
* GetRelationIndicesDependencyList creates a list of ObjectAddressDependencies for the
* indexes on a given relation.
*/
static List *
GetRelationIndicesDependencyList(Oid relationId)
{
List *indexIds = ExecuteFunctionOnEachTableIndex(relationId, CollectIndexOids, 0);
return CreateObjectAddressDependencyDefList(RelationRelationId, indexIds);
}
/*
* GetRelationTriggerFunctionDependencyList returns a list of DependencyDefinition
* objects for the functions that triggers of the relation with relationId depends.

15
src/backend/distributed/metadata/distobject.c
View File

@ -405,6 +405,21 @@ GetDistributedObjectAddressList(void)
}
/*
* GetRoleSpecObjectForUser creates a RoleSpec object for the given roleOid.
*/
RoleSpec *
GetRoleSpecObjectForUser(Oid roleOid)
{
RoleSpec *roleSpec = makeNode(RoleSpec);
roleSpec->roletype = OidIsValid(roleOid) ? ROLESPEC_CSTRING : ROLESPEC_PUBLIC;
roleSpec->rolename = OidIsValid(roleOid) ? GetUserNameFromId(roleOid, false) : NULL;
roleSpec->location = -1;
return roleSpec;
}
/*
* UpdateDistributedObjectColocationId gets an old and a new colocationId
* and updates the colocationId of all tuples in citus.pg_dist_object which

19
src/backend/distributed/metadata/metadata_sync.c
View File

@ -102,7 +102,6 @@ static GrantStmt * GenerateGrantStmtForRights(ObjectType objectType,
bool withGrantOption);
static List * GetObjectsForGrantStmt(ObjectType objectType, Oid objectId);
static AccessPriv * GetAccessPrivObjectForGrantStmt(char *permission);
static RoleSpec * GetRoleSpecObjectForGrantStmt(Oid roleOid);
static List * GenerateGrantOnSchemaQueriesFromAclItem(Oid schemaOid,
AclItem *aclItem);
static void SetLocalEnableMetadataSync(bool state);
@ -1782,7 +1781,7 @@ GenerateGrantStmtForRights(ObjectType objectType,
stmt->objtype = objectType;
stmt->objects = GetObjectsForGrantStmt(objectType, objectId);
stmt->privileges = list_make1(GetAccessPrivObjectForGrantStmt(permission));
stmt->grantees = list_make1(GetRoleSpecObjectForGrantStmt(roleOid));
stmt->grantees = list_make1(GetRoleSpecObjectForUser(roleOid));
stmt->grant_option = withGrantOption;
return stmt;
@ -1831,22 +1830,6 @@ GetAccessPrivObjectForGrantStmt(char *permission)
}
/*
* GetRoleSpecObjectForGrantStmt creates a RoleSpec object for the given roleOid.
* It will be used when creating GrantStmt objects.
*/
static RoleSpec *
GetRoleSpecObjectForGrantStmt(Oid roleOid)
{
RoleSpec *roleSpec = makeNode(RoleSpec);
roleSpec->roletype = OidIsValid(roleOid) ? ROLESPEC_CSTRING : ROLESPEC_PUBLIC;
roleSpec->rolename = OidIsValid(roleOid) ? GetUserNameFromId(roleOid, false) : NULL;
roleSpec->location = -1;
return roleSpec;
}
/*
* SetLocalEnableMetadataSync sets the enable_metadata_sync locally
*/

1
src/backend/distributed/metadata/pg_get_object_address_12_13_14.c
View File

@ -410,6 +410,7 @@ ErrorIfCurrentUserCanNotDistributeObject(ObjectType type, ObjectAddress *addr,
case OBJECT_FUNCTION:
case OBJECT_PROCEDURE:
case OBJECT_AGGREGATE:
case OBJECT_TSCONFIGURATION:
case OBJECT_TYPE:
case OBJECT_FOREIGN_SERVER:
case OBJECT_SEQUENCE:

1
src/backend/distributed/sql/citus--10.2-4--11.0-1.sql
View File

@ -18,6 +18,7 @@
#include "udfs/get_global_active_transactions/11.0-1.sql"
#include "udfs/citus_worker_stat_activity/11.0-1.sql"
#include "udfs/worker_create_or_replace_object/11.0-1.sql"
CREATE VIEW citus.citus_worker_stat_activity AS
SELECT * FROM pg_catalog.citus_worker_stat_activity();

8
src/backend/distributed/sql/citus--8.3-1--9.0-1.sql
View File

@ -21,13 +21,7 @@ ALTER FUNCTION citus.restore_isolation_tester_func SET SCHEMA citus_internal;
GRANT USAGE ON SCHEMA citus TO public;
#include "udfs/pg_dist_shard_placement_trigger_func/9.0-1.sql"
CREATE OR REPLACE FUNCTION pg_catalog.worker_create_or_replace_object(statement text)
RETURNS bool
LANGUAGE C STRICT
AS 'MODULE_PATHNAME', $$worker_create_or_replace_object$$;
COMMENT ON FUNCTION pg_catalog.worker_create_or_replace_object(statement text)
IS 'takes a sql CREATE statement, before executing the create it will check if an object with that name already exists and safely replaces that named object with the new object';
#include "udfs/worker_create_or_replace_object/9.0-1.sql"
CREATE OR REPLACE FUNCTION pg_catalog.master_unmark_object_distributed(classid oid, objid oid, objsubid int)
RETURNS void

3
src/backend/distributed/sql/downgrades/citus--11.0-1--10.2-4.sql
View File

@ -208,4 +208,7 @@ SELECT * FROM pg_catalog.citus_worker_stat_activity();
ALTER VIEW citus.citus_worker_stat_activity SET SCHEMA pg_catalog;
GRANT SELECT ON pg_catalog.citus_worker_stat_activity TO PUBLIC;
DROP FUNCTION pg_catalog.worker_create_or_replace_object(text[]);
#include "../udfs/worker_create_or_replace_object/9.0-1.sql"
RESET search_path;

15
src/backend/distributed/sql/udfs/worker_create_or_replace_object/11.0-1.sql generated Normal file
View File

@ -0,0 +1,15 @@
CREATE OR REPLACE FUNCTION pg_catalog.worker_create_or_replace_object(statement text)
RETURNS bool
LANGUAGE C STRICT
AS 'MODULE_PATHNAME', $$worker_create_or_replace_object$$;
COMMENT ON FUNCTION pg_catalog.worker_create_or_replace_object(statement text)
IS 'takes a sql CREATE statement, before executing the create it will check if an object with that name already exists and safely replaces that named object with the new object';
CREATE OR REPLACE FUNCTION pg_catalog.worker_create_or_replace_object(statements text[])
RETURNS bool
LANGUAGE C STRICT
AS 'MODULE_PATHNAME', $$worker_create_or_replace_object_array$$;
COMMENT ON FUNCTION pg_catalog.worker_create_or_replace_object(statements text[])
IS 'takes a lost of sql statements, before executing these it will check if the object already exists in that exact state otherwise replaces that named object with the new object';

6
src/backend/distributed/sql/udfs/worker_create_or_replace_object/9.0-1.sql generated Normal file
View File

@ -0,0 +1,6 @@
CREATE OR REPLACE FUNCTION pg_catalog.worker_create_or_replace_object(statement text)
RETURNS bool
LANGUAGE C STRICT
AS 'MODULE_PATHNAME', $$worker_create_or_replace_object$$;
COMMENT ON FUNCTION pg_catalog.worker_create_or_replace_object(statement text)
IS 'takes a sql CREATE statement, before executing the create it will check if an object with that name already exists and safely replaces that named object with the new object';

15
src/backend/distributed/sql/udfs/worker_create_or_replace_object/latest.sql Normal file
View File

@ -0,0 +1,15 @@
CREATE OR REPLACE FUNCTION pg_catalog.worker_create_or_replace_object(statement text)
RETURNS bool
LANGUAGE C STRICT
AS 'MODULE_PATHNAME', $$worker_create_or_replace_object$$;
COMMENT ON FUNCTION pg_catalog.worker_create_or_replace_object(statement text)
IS 'takes a sql CREATE statement, before executing the create it will check if an object with that name already exists and safely replaces that named object with the new object';
CREATE OR REPLACE FUNCTION pg_catalog.worker_create_or_replace_object(statements text[])
RETURNS bool
LANGUAGE C STRICT
AS 'MODULE_PATHNAME', $$worker_create_or_replace_object_array$$;
COMMENT ON FUNCTION pg_catalog.worker_create_or_replace_object(statements text[])
IS 'takes a lost of sql statements, before executing these it will check if the object already exists in that exact state otherwise replaces that named object with the new object';

4
src/backend/distributed/utils/resource_lock.c
View File

@ -1012,8 +1012,8 @@ CitusRangeVarCallbackForLockTable(const RangeVar *rangeVar, Oid relationId,
return;
}
/* we only allow tables and views to be locked */
if (!RegularTable(relationId))
/* we only allow tables, views and foreign tables to be locked */
if (!RegularTable(relationId) && !IsForeignTable(relationId))
{
ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("\"%s\" is not a table", rangeVar->relname)));

236
src/backend/distributed/worker/worker_create_or_replace.c
View File

@ -13,8 +13,10 @@
#include "catalog/dependency.h"
#include "catalog/pg_collation.h"
#include "catalog/pg_proc.h"
#include "catalog/pg_ts_config.h"
#include "catalog/pg_type.h"
#include "fmgr.h"
#include "funcapi.h"
#include "nodes/makefuncs.h"
#include "nodes/nodes.h"
#include "parser/parse_type.h"
@ -28,13 +30,17 @@
#include "distributed/commands.h"
#include "distributed/commands/utility_hook.h"
#include "distributed/deparser.h"
#include "distributed/listutils.h"
#include "distributed/metadata/distobject.h"
#include "distributed/worker_create_or_replace.h"
#include "distributed/worker_protocol.h"
static const char * CreateStmtByObjectAddress(const ObjectAddress *address);
static List * CreateStmtListByObjectAddress(const ObjectAddress *address);
static bool CompareStringList(List *list1, List *list2);
PG_FUNCTION_INFO_V1(worker_create_or_replace_object);
PG_FUNCTION_INFO_V1(worker_create_or_replace_object_array);
static bool WorkerCreateOrReplaceObject(List *sqlStatements);
/*
@ -51,6 +57,37 @@ WrapCreateOrReplace(const char *sql)
}
/*
* WrapCreateOrReplaceList takes a list of sql commands and wraps it in a call to citus'
* udf to create or replace the existing object based on its create commands.
*/
char *
WrapCreateOrReplaceList(List *sqls)
{
StringInfoData textArrayLitteral = { 0 };
initStringInfo(&textArrayLitteral);
appendStringInfoString(&textArrayLitteral, "ARRAY[");
const char *sql = NULL;
bool first = true;
foreach_ptr(sql, sqls)
{
if (!first)
{
appendStringInfoString(&textArrayLitteral, ", ");
}
appendStringInfoString(&textArrayLitteral, quote_literal_cstr(sql));
first = false;
}
appendStringInfoString(&textArrayLitteral, "]::text[]");
StringInfoData buf = { 0 };
initStringInfo(&buf);
appendStringInfo(&buf, CREATE_OR_REPLACE_COMMAND, textArrayLitteral.data);
return buf.data;
}
/*
* worker_create_or_replace_object(statement text)
*
@ -73,35 +110,102 @@ Datum
worker_create_or_replace_object(PG_FUNCTION_ARGS)
{
text *sqlStatementText = PG_GETARG_TEXT_P(0);
const char *sqlStatement = text_to_cstring(sqlStatementText);
Node *parseTree = ParseTreeNode(sqlStatement);
char *sqlStatement = text_to_cstring(sqlStatementText);
List *sqlStatements = list_make1(sqlStatement);
PG_RETURN_BOOL(WorkerCreateOrReplaceObject(sqlStatements));
}
/*
* worker_create_or_replace_object(statements text[])
*
* function is called, by the coordinator, with a CREATE statement for an object. This
* function implements the CREATE ... IF NOT EXISTS functionality for objects that do not
* have this functionality or where their implementation is not sufficient.
*
* Besides checking if an object of said name exists it tries to compare the object to be
* created with the one in the local catalog. If there is a difference the one in the local
* catalog will be renamed after which the statement can be executed on this worker to
* create the object. If more statements are provided, all are compared in order with the
* statements generated on the worker. This works assuming a) both citus versions are the
* same, b) the objects are exactly the same.
*
* Renaming has two purposes
* - free the identifier for creation
* - non destructive if there is data store that would be destroyed if the object was
* used in a table on this node, eg. types. If the type would be dropped with a cascade
* it would drop any column holding user data for this type.
*/
Datum
worker_create_or_replace_object_array(PG_FUNCTION_ARGS)
{
List *sqlStatements = NIL;
Datum *textArray = NULL;
int length = 0;
deconstruct_array(PG_GETARG_ARRAYTYPE_P(0), TEXTOID, -1, false, 'i', &textArray,
NULL, &length);
for (int i = 0; i < length; i++)
{
sqlStatements = lappend(sqlStatements, TextDatumGetCString(textArray[i]));
}
if (list_length(sqlStatements) < 1)
{
ereport(ERROR, (errmsg("expected atleast 1 statement to be provided")));
}
PG_RETURN_BOOL(WorkerCreateOrReplaceObject(sqlStatements));
}
/*
* WorkerCreateOrReplaceObject implements the logic used by both variants of
* worker_create_or_replace_object to either create the object or coming to the conclusion
* the object already exists in the correct state.
*
* Returns true if the object has been created, false if it was already in the exact state
* it was asked for.
*/
static bool
WorkerCreateOrReplaceObject(List *sqlStatements)
{
/*
* since going to the drop statement might require some resolving we will do a check
* if the type actually exists instead of adding the IF EXISTS keyword to the
* statement.
* To check which object we are changing we find the object address from the first
* statement passed into the UDF. Later we will check if all object addresses are the
* same.
*
* Although many of the objects will only have one statement in this call, more
* complex objects might come with a list of statements. We assume they all are on the
* same subject.
*/
Node *parseTree = ParseTreeNode(linitial(sqlStatements));
ObjectAddress address = GetObjectAddressFromParseTree(parseTree, true);
if (ObjectExists(&address))
{
const char *localSqlStatement = CreateStmtByObjectAddress(&address);
/*
* Object with name from statement is already found locally, check if states are
* identical. If objects differ we will rename the old object (non- destructively)
* as to make room to create the new object according to the spec sent.
*/
if (strcmp(sqlStatement, localSqlStatement) == 0)
/*
* Based on the local catalog we generate the list of commands we would send to
* recreate our version of the object. This we can compare to what the coordinator
* sent us. If they match we don't do anything.
*/
List *localSqlStatements = CreateStmtListByObjectAddress(&address);
if (CompareStringList(sqlStatements, localSqlStatements))
{
/*
* TODO string compare is a poor man's comparison, but calling equal on the
* parsetree's returns false because there is extra information list character
* position of some sort
*/
/*
* parseTree sent by the coordinator is the same as we would create for our
* object, therefore we can omit the create statement locally and not create
* the object as it already exists.
* statements sent by the coordinator are the same as we would create for our
* object, therefore we can omit the statements locally and not create the
* object as it already exists in the correct shape.
*
* We let the coordinator know we didn't create the object.
*/
PG_RETURN_BOOL(false);
return false;
}
char *newName = GenerateBackupNameForCollision(&address);
@ -113,12 +217,47 @@ worker_create_or_replace_object(PG_FUNCTION_ARGS)
NULL, None_Receiver, NULL);
}
/* apply create statement locally */
ProcessUtilityParseTree(parseTree, sqlStatement, PROCESS_UTILITY_QUERY, NULL,
None_Receiver, NULL);
/* apply all statement locally */
char *sqlStatement = NULL;
foreach_ptr(sqlStatement, sqlStatements)
{
parseTree = ParseTreeNode(sqlStatement);
ProcessUtilityParseTree(parseTree, sqlStatement, PROCESS_UTILITY_QUERY, NULL,
None_Receiver, NULL);
/* TODO verify all statements are about exactly 1 subject, mostly a sanity check
* to prevent unintentional use of this UDF, needs to come after the local
* execution to be able to actually resolve the ObjectAddress of the newly created
* object */
}
/* type has been created */
PG_RETURN_BOOL(true);
return true;
}
static bool
CompareStringList(List *list1, List *list2)
{
if (list_length(list1) != list_length(list2))
{
return false;
}
ListCell *cell1 = NULL;
ListCell *cell2 = NULL;
forboth(cell1, list1, cell2, list2)
{
const char *str1 = lfirst(cell1);
const char *str2 = lfirst(cell2);
if (strcmp(str1, str2) != 0)
{
return false;
}
}
return true;
}
@ -130,24 +269,38 @@ worker_create_or_replace_object(PG_FUNCTION_ARGS)
* therefore you cannot equal this tree against parsed statement. Instead it can be
* deparsed to do a string comparison.
*/
static const char *
CreateStmtByObjectAddress(const ObjectAddress *address)
static List *
CreateStmtListByObjectAddress(const ObjectAddress *address)
{
switch (getObjectClass(address))
{
case OCLASS_COLLATION:
{
return CreateCollationDDL(address->objectId);
return list_make1(CreateCollationDDL(address->objectId));
}
case OCLASS_PROC:
{
return GetFunctionDDLCommand(address->objectId, false);
return list_make1(GetFunctionDDLCommand(address->objectId, false));
}
case OCLASS_TSCONFIG:
{
/*
* We do support TEXT SEARCH CONFIGURATION, however, we can't recreate the
* object in 1 command. Since the returned text is compared to the create
* statement sql we always want the sql to be different compared to the
* canonical creation sql we return here, hence we return an empty string, as
* that should never match the sql we have passed in for the creation.
*/
List *stmts = GetCreateTextSearchConfigStatements(address);
return DeparseTreeNodes(stmts);
}
case OCLASS_TYPE:
{
return DeparseTreeNode(CreateTypeStmtByObjectAddress(address));
return list_make1(DeparseTreeNode(CreateTypeStmtByObjectAddress(address)));
}
default:
@ -179,6 +332,11 @@ GenerateBackupNameForCollision(const ObjectAddress *address)
return GenerateBackupNameForProcCollision(address);
}
case OCLASS_TSCONFIG:
{
return GenerateBackupNameForTextSearchConfiguration(address);
}
case OCLASS_TYPE:
{
return GenerateBackupNameForTypeCollision(address);
@ -256,6 +414,25 @@ CreateRenameTypeStmt(const ObjectAddress *address, char *newName)
}
/*
* CreateRenameTextSearchStmt creates a rename statement for a text search configuration
* based on its ObjectAddress. The rename statement will rename the existing object on its
* address to the value provided in newName.
*/
static RenameStmt *
CreateRenameTextSearchStmt(const ObjectAddress *address, char *newName)
{
Assert(address->classId == TSConfigRelationId);
RenameStmt *stmt = makeNode(RenameStmt);
stmt->renameType = OBJECT_TSCONFIGURATION;
stmt->object = (Node *) get_ts_config_namelist(address->objectId);
stmt->newname = newName;
return stmt;
}
/*
* CreateRenameTypeStmt creates a rename statement for a type based on its ObjectAddress.
* The rename statement will rename the existing object on its address to the value
@ -325,6 +502,11 @@ CreateRenameStatement(const ObjectAddress *address, char *newName)
return CreateRenameProcStmt(address, newName);
}
case OCLASS_TSCONFIG:
{
return CreateRenameTextSearchStmt(address, newName);
}
case OCLASS_TYPE:
{
return CreateRenameTypeStmt(address, newName);

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

@ -465,6 +465,54 @@ extern Oid GetSequenceOid(Oid relationId, AttrNumber attnum);
extern bool ConstrTypeUsesIndex(ConstrType constrType);
/* text_search.c - forward declarations */
extern List * PostprocessCreateTextSearchConfigurationStmt(Node *node,
const char *queryString);
extern List * GetCreateTextSearchConfigStatements(const ObjectAddress *address);
extern List * CreateTextSearchConfigDDLCommandsIdempotent(const ObjectAddress *address);
extern List * PreprocessDropTextSearchConfigurationStmt(Node *node,
const char *queryString,
ProcessUtilityContext
processUtilityContext);
extern List * PreprocessAlterTextSearchConfigurationStmt(Node *node,
const char *queryString,
ProcessUtilityContext
processUtilityContext);
extern List * PreprocessRenameTextSearchConfigurationStmt(Node *node,
const char *queryString,
ProcessUtilityContext
processUtilityContext);
extern List * PreprocessAlterTextSearchConfigurationSchemaStmt(Node *node,
const char *queryString,
ProcessUtilityContext
processUtilityContext);
extern List * PostprocessAlterTextSearchConfigurationSchemaStmt(Node *node,
const char *queryString);
extern List * PreprocessTextSearchConfigurationCommentStmt(Node *node,
const char *queryString,
ProcessUtilityContext
processUtilityContext);
extern List * PreprocessAlterTextSearchConfigurationOwnerStmt(Node *node,
const char *queryString,
ProcessUtilityContext
processUtilityContext);
extern List * PostprocessAlterTextSearchConfigurationOwnerStmt(Node *node,
const char *queryString);
extern ObjectAddress CreateTextSearchConfigurationObjectAddress(Node *node,
bool missing_ok);
extern ObjectAddress RenameTextSearchConfigurationStmtObjectAddress(Node *node,
bool missing_ok);
extern ObjectAddress AlterTextSearchConfigurationStmtObjectAddress(Node *node,
bool missing_ok);
extern ObjectAddress AlterTextSearchConfigurationSchemaStmtObjectAddress(Node *node,
bool missing_ok);
extern ObjectAddress TextSearchConfigurationCommentObjectAddress(Node *node,
bool missing_ok);
extern ObjectAddress AlterTextSearchConfigurationOwnerObjectAddress(Node *node,
bool missing_ok);
extern char * GenerateBackupNameForTextSearchConfiguration(const ObjectAddress *address);
extern List * get_ts_config_namelist(Oid tsconfigOid);
/* truncate.c - forward declarations */
extern void PreprocessTruncateStatement(TruncateStmt *truncateStatement);

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

@ -31,6 +31,7 @@ extern void AssertObjectTypeIsFunctional(ObjectType type);
extern void QualifyTreeNode(Node *stmt);
extern char * DeparseTreeNode(Node *stmt);
extern List * DeparseTreeNodes(List *stmts);
/* forward declarations for deparse_attribute_stmts.c */
extern char * DeparseRenameAttributeStmt(Node *);
@ -59,6 +60,15 @@ extern char * DeparseAlterTableStmt(Node *node);
extern void QualifyAlterTableSchemaStmt(Node *stmt);
/* foward declarations fro deparse_text_search.c */
extern char * DeparseCreateTextSearchStmt(Node *node);
extern char * DeparseDropTextSearchConfigurationStmt(Node *node);
extern char * DeparseRenameTextSearchConfigurationStmt(Node *node);
extern char * DeparseAlterTextSearchConfigurationStmt(Node *node);
extern char * DeparseAlterTextSearchConfigurationSchemaStmt(Node *node);
extern char * DeparseTextSearchConfigurationCommentStmt(Node *node);
extern char * DeparseAlterTextSearchConfigurationOwnerStmt(Node *node);
/* forward declarations for deparse_schema_stmts.c */
extern char * DeparseCreateSchemaStmt(Node *node);
extern char * DeparseDropSchemaStmt(Node *node);
@ -140,6 +150,14 @@ extern char * DeparseAlterExtensionStmt(Node *stmt);
/* forward declarations for deparse_database_stmts.c */
extern char * DeparseAlterDatabaseOwnerStmt(Node *node);
/* forward declatations for depatse_text_search_stmts.c */
extern void QualifyDropTextSearchConfigurationStmt(Node *node);
extern void QualifyAlterTextSearchConfigurationStmt(Node *node);
extern void QualifyRenameTextSearchConfigurationStmt(Node *node);
extern void QualifyAlterTextSearchConfigurationSchemaStmt(Node *node);
extern void QualifyTextSearchConfigurationCommentStmt(Node *node);
extern void QualifyAlterTextSearchConfigurationOwnerStmt(Node *node);
/* forward declarations for deparse_sequence_stmts.c */
extern char * DeparseDropSequenceStmt(Node *node);
extern char * DeparseRenameSequenceStmt(Node *node);

2
src/include/distributed/metadata/distobject.h
View File

@ -30,8 +30,8 @@ extern bool IsObjectAddressOwnedByExtension(const ObjectAddress *target,
ObjectAddress *extensionAddress);
extern ObjectAddress PgGetObjectAddress(char *ttype, ArrayType *namearr,
ArrayType *argsarr);
extern List * GetDistributedObjectAddressList(void);
extern RoleSpec * GetRoleSpecObjectForUser(Oid roleOid);
extern void UpdateDistributedObjectColocationId(uint32 oldColocationId, uint32
newColocationId);
#endif /* CITUS_METADATA_DISTOBJECT_H */

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

@ -19,6 +19,7 @@
#define CREATE_OR_REPLACE_COMMAND "SELECT worker_create_or_replace_object(%s);"
extern char * WrapCreateOrReplace(const char *sql);
extern char * WrapCreateOrReplaceList(List *sqls);
extern char * GenerateBackupNameForCollision(const ObjectAddress *address);
extern RenameStmt * CreateRenameStatement(const ObjectAddress *address, char *newName);

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

@ -1013,9 +1013,10 @@ SELECT * FROM multi_extension.print_extension_changes();
| function citus_shard_indexes_on_worker() SETOF record
| function citus_shards_on_worker() SETOF record
| function create_distributed_function(regprocedure,text,text,boolean) void
| function worker_create_or_replace_object(text[]) boolean
| function worker_drop_sequence_dependency(text) void
| function worker_drop_shell_table(text) void
(15 rows)
(16 rows)
DROP TABLE multi_extension.prev_objects, multi_extension.extension_diff;
-- show running version

69
src/test/regress/expected/pg14.out
View File

@ -1272,3 +1272,72 @@ SELECT count(*) FROM
set client_min_messages to error;
drop schema pg14 cascade;
create schema pg14;
set search_path to pg14;
select 1 from citus_add_node('localhost',:master_port,groupid=>0);
?column?
---------------------------------------------------------------------
1
(1 row)
-- test adding foreign table to metadata with the guc
-- will test truncating foreign tables later
CREATE TABLE foreign_table_test (id integer NOT NULL, data text, a bigserial);
INSERT INTO foreign_table_test VALUES (1, 'text_test');
SELECT citus_add_local_table_to_metadata('foreign_table_test');
citus_add_local_table_to_metadata
---------------------------------------------------------------------
(1 row)
CREATE EXTENSION postgres_fdw;
CREATE SERVER foreign_server
FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (host 'localhost', port :'master_port', dbname 'regression');
CREATE USER MAPPING FOR CURRENT_USER
SERVER foreign_server
OPTIONS (user 'postgres');
CREATE FOREIGN TABLE foreign_table (
id integer NOT NULL,
data text,
a bigserial
)
SERVER foreign_server
OPTIONS (schema_name 'pg14', table_name 'foreign_table_test');
SELECT citus_add_local_table_to_metadata('foreign_table');
citus_add_local_table_to_metadata
---------------------------------------------------------------------
(1 row)
SELECT count(*) FROM foreign_table;
count
---------------------------------------------------------------------
1
(1 row)
TRUNCATE foreign_table;
\c - - - :worker_1_port
set search_path to pg14;
-- verify the foreign table is truncated
SELECT count(*) FROM pg14.foreign_table;
count
---------------------------------------------------------------------
0
(1 row)
-- should error out
TRUNCATE foreign_table;
ERROR: truncating foreign tables that are added to metadata can only be excuted on the coordinator
\c - - - :master_port
-- cleanup
set client_min_messages to error;
drop extension postgres_fdw cascade;
drop schema pg14 cascade;
reset client_min_messages;
select 1 from citus_remove_node('localhost',:master_port);
?column?
---------------------------------------------------------------------
1
(1 row)

489
src/test/regress/expected/text_search.out Normal file
View File

@ -0,0 +1,489 @@
CREATE SCHEMA text_search;
CREATE SCHEMA text_search2;
SET search_path TO text_search;
-- create a new configruation from scratch
CREATE TEXT SEARCH CONFIGURATION my_text_search_config ( parser = default );
CREATE TABLE t1(id int, name text);
CREATE INDEX t1_search_name ON t1 USING gin (to_tsvector('text_search.my_text_search_config'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t1', 'name');
create_distributed_table
---------------------------------------------------------------------
(1 row)
DROP TABLE t1;
DROP TEXT SEARCH CONFIGURATION my_text_search_config;
-- try to create table and index in 1 transaction
BEGIN;
CREATE TEXT SEARCH CONFIGURATION my_text_search_config ( parser = default );
CREATE TABLE t1(id int, name text);
CREATE INDEX t1_search_name ON t1 USING gin (to_tsvector('text_search.my_text_search_config'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t1', 'name');
create_distributed_table
---------------------------------------------------------------------
(1 row)
ABORT;
-- try again, should not fail with my_text_search_config being retained on the worker
BEGIN;
CREATE TEXT SEARCH CONFIGURATION my_text_search_config ( parser = default );
COMMENT ON TEXT SEARCH CONFIGURATION my_text_search_config IS 'on demand propagation of text search object with a comment';
CREATE TABLE t1(id int, name text);
CREATE INDEX t1_search_name ON t1 USING gin (to_tsvector('text_search.my_text_search_config'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t1', 'name');
create_distributed_table
---------------------------------------------------------------------
(1 row)
SELECT * FROM run_command_on_workers($$
SELECT obj_description('text_search.my_text_search_config'::regconfig);
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | on demand propagation of text search object with a comment
localhost | 57638 | t | on demand propagation of text search object with a comment
(2 rows)
-- verify that changing anything on a managed TEXT SEARCH CONFIGURATION fails after parallel execution
COMMENT ON TEXT SEARCH CONFIGURATION my_text_search_config IS 'this comment can''t be set right now';
ERROR: cannot run text search configuration command because there was a parallel operation on a distributed table in the transaction
DETAIL: When running command on/for a distributed text search configuration, Citus needs to perform all operations over a single connection per node to ensure consistency.
HINT: Try re-running the transaction with "SET LOCAL citus.multi_shard_modify_mode TO 'sequential';"
ABORT;
-- create an index on an already distributed table
BEGIN;
CREATE TEXT SEARCH CONFIGURATION my_text_search_config2 ( parser = default );
COMMENT ON TEXT SEARCH CONFIGURATION my_text_search_config2 IS 'on demand propagation of text search object with a comment 2';
CREATE TABLE t1(id int, name text);
SELECT create_distributed_table('t1', 'name');
create_distributed_table
---------------------------------------------------------------------
(1 row)
CREATE INDEX t1_search_name ON t1 USING gin (to_tsvector('text_search.my_text_search_config2'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT * FROM run_command_on_workers($$
SELECT obj_description('text_search.my_text_search_config2'::regconfig);
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | on demand propagation of text search object with a comment 2
localhost | 57638 | t | on demand propagation of text search object with a comment 2
(2 rows)
ABORT;
-- should be able to create a configuration based on a copy of an existing configuration
CREATE TEXT SEARCH CONFIGURATION french_noaccent ( COPY = french );
CREATE TABLE t2(id int, name text);
CREATE INDEX t2_search_name ON t2 USING gin (to_tsvector('text_search.french_noaccent'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t2', 'id');
create_distributed_table
---------------------------------------------------------------------
(1 row)
-- spot check that french_noaccent copied settings from french
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'comment tu t''appelle') WHERE alias = 'asciiword' LIMIT 1;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | (asciiword,french_stem)
localhost | 57638 | t | (asciiword,french_stem)
(2 rows)
-- makes no sense, however we expect that the dictionary for the first token changes accordingly
ALTER TEXT SEARCH CONFIGURATION french_noaccent ALTER MAPPING FOR asciiword WITH dutch_stem;
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'comment tu t''appelle') WHERE alias = 'asciiword' LIMIT 1;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | (asciiword,dutch_stem)
localhost | 57638 | t | (asciiword,dutch_stem)
(2 rows)
-- do the same but we will replace all french dictionaries
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciihword' LIMIT 1;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | (asciihword,french_stem)
localhost | 57638 | t | (asciihword,french_stem)
(2 rows)
ALTER TEXT SEARCH CONFIGURATION french_noaccent ALTER MAPPING REPLACE french_stem WITH dutch_stem;
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciihword' LIMIT 1;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | (asciihword,dutch_stem)
localhost | 57638 | t | (asciihword,dutch_stem)
(2 rows)
-- once more but now back via yet a different DDL command
ALTER TEXT SEARCH CONFIGURATION french_noaccent ALTER MAPPING FOR asciihword REPLACE dutch_stem WITH french_stem;
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciihword' LIMIT 1;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | (asciihword,french_stem)
localhost | 57638 | t | (asciihword,french_stem)
(2 rows)
-- drop a mapping
ALTER TEXT SEARCH CONFIGURATION french_noaccent DROP MAPPING FOR asciihword;
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciihword' LIMIT 1;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | (asciihword,)
localhost | 57638 | t | (asciihword,)
(2 rows)
-- also with exists, doesn't change anything, but should not error
ALTER TEXT SEARCH CONFIGURATION french_noaccent DROP MAPPING IF EXISTS FOR asciihword;
NOTICE: mapping for token type "asciihword" does not exist, skipping
-- Comment on a text search configuration
COMMENT ON TEXT SEARCH CONFIGURATION french_noaccent IS 'a text configuration that is butcherd to test all edge cases';
SELECT * FROM run_command_on_workers($$
SELECT obj_description('text_search.french_noaccent'::regconfig);
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | a text configuration that is butcherd to test all edge cases
localhost | 57638 | t | a text configuration that is butcherd to test all edge cases
(2 rows)
-- Remove a comment
COMMENT ON TEXT SEARCH CONFIGURATION french_noaccent IS NULL;
SELECT * FROM run_command_on_workers($$
SELECT obj_description('text_search.french_noaccent'::regconfig);
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t |
localhost | 57638 | t |
(2 rows)
-- verify adding 2 dictionaries for two tokes at once
ALTER TEXT SEARCH CONFIGURATION french_noaccent DROP MAPPING IF EXISTS FOR asciiword, asciihword;
NOTICE: mapping for token type "asciihword" does not exist, skipping
ALTER TEXT SEARCH CONFIGURATION french_noaccent ADD MAPPING FOR asciiword, asciihword WITH french_stem, dutch_stem;
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionaries) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciiword' LIMIT 1;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | (asciiword,"{french_stem,dutch_stem}")
localhost | 57638 | t | (asciiword,"{french_stem,dutch_stem}")
(2 rows)
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionaries) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciihword' LIMIT 1;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | (asciihword,"{french_stem,dutch_stem}")
localhost | 57638 | t | (asciihword,"{french_stem,dutch_stem}")
(2 rows)
--verify we can drop cascade a configuration that is in use
-- verify it is in use
DROP TEXT SEARCH CONFIGURATION text_search.french_noaccent;
ERROR: cannot drop text search configuration french_noaccent because other objects depend on it
DETAIL: index t2_search_name depends on text search configuration french_noaccent
HINT: Use DROP ... CASCADE to drop the dependent objects too.
-- drop cascade
DROP TEXT SEARCH CONFIGURATION text_search.french_noaccent CASCADE;
NOTICE: drop cascades to index t2_search_name
-- verify the configuration is dropped from the workers
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.french_noaccent'::regconfig; $$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | f | ERROR: text search configuration "text_search.french_noaccent" does not exist
localhost | 57638 | f | ERROR: text search configuration "text_search.french_noaccent" does not exist
(2 rows)
SET client_min_messages TO 'warning';
SELECT * FROM run_command_on_workers($$CREATE ROLE text_search_owner;$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | CREATE ROLE
localhost | 57638 | t | CREATE ROLE
(2 rows)
CREATE ROLE text_search_owner;
RESET client_min_messages;
CREATE TEXT SEARCH CONFIGURATION changed_owner ( PARSER = default );
SELECT * FROM run_command_on_workers($$
SELECT cfgowner::regrole
FROM pg_ts_config
WHERE oid = 'text_search.changed_owner'::regconfig;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | postgres
localhost | 57638 | t | postgres
(2 rows)
ALTER TEXT SEARCH CONFIGURATION changed_owner OWNER TO text_search_owner;
SELECT * FROM run_command_on_workers($$
SELECT cfgowner::regrole
FROM pg_ts_config
WHERE oid = 'text_search.changed_owner'::regconfig;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | text_search_owner
localhost | 57638 | t | text_search_owner
(2 rows)
-- redo test with propagating object after it was created and changed of owner
SET citus.enable_ddl_propagation TO off;
CREATE TEXT SEARCH CONFIGURATION changed_owner2 ( PARSER = default );
ALTER TEXT SEARCH CONFIGURATION changed_owner2 OWNER TO text_search_owner;
RESET citus.enable_ddl_propagation;
-- verify object doesn't exist before propagating
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.changed_owner2'::regconfig; $$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | f | ERROR: text search configuration "text_search.changed_owner2" does not exist
localhost | 57638 | f | ERROR: text search configuration "text_search.changed_owner2" does not exist
(2 rows)
-- distribute configuration
CREATE TABLE t3(id int, name text);
CREATE INDEX t3_search_name ON t3 USING gin (to_tsvector('text_search.changed_owner2'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t3', 'name');
create_distributed_table
---------------------------------------------------------------------
(1 row)
-- verify config owner
SELECT * FROM run_command_on_workers($$
SELECT cfgowner::regrole
FROM pg_ts_config
WHERE oid = 'text_search.changed_owner2'::regconfig;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | text_search_owner
localhost | 57638 | t | text_search_owner
(2 rows)
-- rename tests
CREATE TEXT SEARCH CONFIGURATION change_name ( PARSER = default );
SELECT * FROM run_command_on_workers($$ -- verify the name exists on the worker
SELECT 'text_search.change_name'::regconfig;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | text_search.change_name
localhost | 57638 | t | text_search.change_name
(2 rows)
ALTER TEXT SEARCH CONFIGURATION change_name RENAME TO changed_name;
SELECT * FROM run_command_on_workers($$ -- verify the name exists on the worker
SELECT 'text_search.changed_name'::regconfig;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | text_search.changed_name
localhost | 57638 | t | text_search.changed_name
(2 rows)
-- test move of schema
CREATE TEXT SEARCH CONFIGURATION change_schema ( PARSER = default );
SELECT * FROM run_command_on_workers($$ -- verify the name exists on the worker
SELECT 'text_search.change_schema'::regconfig;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | text_search.change_schema
localhost | 57638 | t | text_search.change_schema
(2 rows)
ALTER TEXT SEARCH CONFIGURATION change_schema SET SCHEMA text_search2;
SELECT * FROM run_command_on_workers($$ -- verify the name exists on the worker
SELECT 'text_search2.change_schema'::regconfig;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | text_search2.change_schema
localhost | 57638 | t | text_search2.change_schema
(2 rows)
-- verify we get an error that the configuration change_schema is not found, even though the object address will be
-- found in its new schema, and is distributed
ALTER TEXT SEARCH CONFIGURATION change_schema SET SCHEMA text_search2;
ERROR: text search configuration "change_schema" does not exist
-- should tell us that text_search.does_not_exist does not exist, covers a complex edgecase
-- in resolving the object address
ALTER TEXT SEARCH CONFIGURATION text_search.does_not_exist SET SCHEMA text_search2;
ERROR: text search configuration "text_search.does_not_exist" does not exist
-- verify edgecases in deparsers
CREATE TEXT SEARCH CONFIGURATION config1 ( PARSER = default );
CREATE TEXT SEARCH CONFIGURATION config2 ( PARSER = default );
SET citus.enable_ddl_propagation TO off;
CREATE TEXT SEARCH CONFIGURATION config3 ( PARSER = default );
RESET citus.enable_ddl_propagation;
-- verify config1, config2 exist on workers, config3 not
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config1'::regconfig; $$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | text_search.config1
localhost | 57638 | t | text_search.config1
(2 rows)
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config2'::regconfig; $$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | text_search.config2
localhost | 57638 | t | text_search.config2
(2 rows)
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config3'::regconfig; $$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | f | ERROR: text search configuration "text_search.config3" does not exist
localhost | 57638 | f | ERROR: text search configuration "text_search.config3" does not exist
(2 rows)
-- DROP all config's, only 1&2 are distributed, they should propagate well to remotes
DROP TEXT SEARCH CONFIGURATION config1, config2, config3;
-- verify all existing ones have been removed (checking config3 for consistency)
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config1'::regconfig; $$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | f | ERROR: text search configuration "text_search.config1" does not exist
localhost | 57638 | f | ERROR: text search configuration "text_search.config1" does not exist
(2 rows)
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config2'::regconfig; $$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | f | ERROR: text search configuration "text_search.config2" does not exist
localhost | 57638 | f | ERROR: text search configuration "text_search.config2" does not exist
(2 rows)
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config3'::regconfig; $$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | f | ERROR: text search configuration "text_search.config3" does not exist
localhost | 57638 | f | ERROR: text search configuration "text_search.config3" does not exist
(2 rows)
-- verify they are all removed locally
SELECT 'text_search.config1'::regconfig;
ERROR: text search configuration "text_search.config1" does not exist
SELECT 'text_search.config2'::regconfig;
ERROR: text search configuration "text_search.config2" does not exist
SELECT 'text_search.config3'::regconfig;
ERROR: text search configuration "text_search.config3" does not exist
-- verify that indexes created concurrently that would propagate a TEXT SEARCH CONFIGURATION object
SET citus.enable_ddl_propagation TO off;
CREATE TEXT SEARCH CONFIGURATION concurrent_index_config ( PARSER = default );
RESET citus.enable_ddl_propagation;
-- verify it doesn't exist on the workers
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.concurrent_index_config'::regconfig; $$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | f | ERROR: text search configuration "text_search.concurrent_index_config" does not exist
localhost | 57638 | f | ERROR: text search configuration "text_search.concurrent_index_config" does not exist
(2 rows)
-- create distributed table that then concurrently would have an index created.
CREATE TABLE t4(id int, name text);
SELECT create_distributed_table('t4', 'name');
create_distributed_table
---------------------------------------------------------------------
(1 row)
CREATE INDEX CONCURRENTLY t4_search_name ON t4 USING gin (to_tsvector('text_search.concurrent_index_config'::regconfig, (COALESCE(name, ''::character varying))::text));
-- now the configuration should be on the worker, and the above index creation shouldn't have failed.
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.concurrent_index_config'::regconfig; $$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | text_search.concurrent_index_config
localhost | 57638 | t | text_search.concurrent_index_config
(2 rows)
-- verify the objid is correctly committed locally due to the somewhat convoluted commit and new transaction starting when creating an index concurrently
SELECT pg_catalog.pg_identify_object_as_address(classid, objid, objsubid)
FROM citus.pg_dist_object
WHERE classid = 3602 AND objid = 'text_search.concurrent_index_config'::regconfig::oid;
pg_identify_object_as_address
---------------------------------------------------------------------
("text search configuration","{text_search,concurrent_index_config}",{})
(1 row)
-- verify old text search configurations get renamed if they are not the same as the newly propagated configuration.
-- We do this by creating configurations on the workers as a copy from a different existing catalog.
SELECT * FROM run_command_on_workers($$
set citus.enable_metadata_sync TO off;
CREATE TEXT SEARCH CONFIGURATION text_search.manually_created_wrongly ( copy = dutch );
reset citus.enable_metadata_sync;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | SET
localhost | 57638 | t | SET
(2 rows)
CREATE TEXT SEARCH CONFIGURATION text_search.manually_created_wrongly ( copy = french );
-- now we expect manually_created_wrongly(citus_backup_XXX) to show up when querying the configurations
SELECT * FROM run_command_on_workers($$
SELECT array_agg(cfgname) FROM pg_ts_config WHERE cfgname LIKE 'manually_created_wrongly%';
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | {manually_created_wrongly(citus_backup_0),manually_created_wrongly}
localhost | 57638 | t | {manually_created_wrongly(citus_backup_0),manually_created_wrongly}
(2 rows)
-- verify the objects get reused appropriately when the specification is the same
SELECT * FROM run_command_on_workers($$
set citus.enable_metadata_sync TO off;
CREATE TEXT SEARCH CONFIGURATION text_search.manually_created_correct ( copy = french );
reset citus.enable_metadata_sync;
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | SET
localhost | 57638 | t | SET
(2 rows)
CREATE TEXT SEARCH CONFIGURATION text_search.manually_created_correct ( copy = french );
-- now we don't expect manually_created_correct(citus_backup_XXX) to show up when querying the configurations as the
-- original one is reused
SELECT * FROM run_command_on_workers($$
SELECT array_agg(cfgname) FROM pg_ts_config WHERE cfgname LIKE 'manually_created_correct%';
$$) ORDER BY 1,2;
nodename | nodeport | success | result
---------------------------------------------------------------------
localhost | 57637 | t | {manually_created_correct}
localhost | 57638 | t | {manually_created_correct}
(2 rows)
CREATE SCHEMA "Text Search Requiring Quote's";
CREATE TEXT SEARCH CONFIGURATION "Text Search Requiring Quote's"."Quoted Config Name" ( parser = default );
CREATE TABLE t5(id int, name text);
CREATE INDEX t5_search_name ON t5 USING gin (to_tsvector('"Text Search Requiring Quote''s"."Quoted Config Name"'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t5', 'name');
create_distributed_table
---------------------------------------------------------------------
(1 row)
SET client_min_messages TO 'warning';
DROP SCHEMA text_search, text_search2, "Text Search Requiring Quote's" CASCADE;
DROP ROLE text_search_owner;

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

@ -203,6 +203,7 @@ ORDER BY 1;
function worker_cleanup_job_schema_cache()
function worker_create_or_alter_role(text,text,text)
function worker_create_or_replace_object(text)
function worker_create_or_replace_object(text[])
function worker_create_schema(bigint,text)
function worker_create_truncate_trigger(regclass)
function worker_drop_distributed_table(text)
@ -267,5 +268,5 @@ ORDER BY 1;
view citus_worker_stat_activity
view pg_dist_shard_placement
view time_partitions
(251 rows)
(252 rows)

2
src/test/regress/isolation_schedule
View File

@ -19,6 +19,7 @@ test: isolation_cluster_management
# the following tests depend on the distributed
# transactionId, so should not be parallelized
# and no tests should be added before these
test: isolation_metadata_sync_vs_all
test: isolation_distributed_transaction_id
test: isolation_dump_global_wait_edges
test: isolation_citus_dist_activity
@ -92,7 +93,6 @@ test: isolation_metadata_sync_deadlock
test: isolation_replicated_dist_on_mx
test: isolation_replicate_reference_tables_to_coordinator
test: isolation_multiuser_locking
test: isolation_metadata_sync_vs_all
# MXless tests
test: isolation_check_mx

2
src/test/regress/multi_1_schedule
View File

@ -315,7 +315,7 @@ test: ssl_by_default
# TODO: After deprecating parameterless create_distributed_function combine
# distributed_functions and function_propagation tests
# ---------
test: distributed_types distributed_types_conflict disable_object_propagation distributed_types_xact_add_enum_value
test: distributed_types distributed_types_conflict disable_object_propagation distributed_types_xact_add_enum_value text_search
test: check_mx
test: distributed_functions distributed_functions_conflict
test: distributed_collations

43
src/test/regress/sql/pg14.sql
View File

@ -642,3 +642,46 @@ SELECT count(*) FROM
set client_min_messages to error;
drop schema pg14 cascade;
create schema pg14;
set search_path to pg14;
select 1 from citus_add_node('localhost',:master_port,groupid=>0);
-- test adding foreign table to metadata with the guc
-- will test truncating foreign tables later
CREATE TABLE foreign_table_test (id integer NOT NULL, data text, a bigserial);
INSERT INTO foreign_table_test VALUES (1, 'text_test');
SELECT citus_add_local_table_to_metadata('foreign_table_test');
CREATE EXTENSION postgres_fdw;
CREATE SERVER foreign_server
FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (host 'localhost', port :'master_port', dbname 'regression');
CREATE USER MAPPING FOR CURRENT_USER
SERVER foreign_server
OPTIONS (user 'postgres');
CREATE FOREIGN TABLE foreign_table (
id integer NOT NULL,
data text,
a bigserial
)
SERVER foreign_server
OPTIONS (schema_name 'pg14', table_name 'foreign_table_test');
SELECT citus_add_local_table_to_metadata('foreign_table');
SELECT count(*) FROM foreign_table;
TRUNCATE foreign_table;
\c - - - :worker_1_port
set search_path to pg14;
-- verify the foreign table is truncated
SELECT count(*) FROM pg14.foreign_table;
-- should error out
TRUNCATE foreign_table;
\c - - - :master_port
-- cleanup
set client_min_messages to error;
drop extension postgres_fdw cascade;
drop schema pg14 cascade;
reset client_min_messages;
select 1 from citus_remove_node('localhost',:master_port);

263
src/test/regress/sql/text_search.sql Normal file
View File

@ -0,0 +1,263 @@
CREATE SCHEMA text_search;
CREATE SCHEMA text_search2;
SET search_path TO text_search;
-- create a new configruation from scratch
CREATE TEXT SEARCH CONFIGURATION my_text_search_config ( parser = default );
CREATE TABLE t1(id int, name text);
CREATE INDEX t1_search_name ON t1 USING gin (to_tsvector('text_search.my_text_search_config'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t1', 'name');
DROP TABLE t1;
DROP TEXT SEARCH CONFIGURATION my_text_search_config;
-- try to create table and index in 1 transaction
BEGIN;
CREATE TEXT SEARCH CONFIGURATION my_text_search_config ( parser = default );
CREATE TABLE t1(id int, name text);
CREATE INDEX t1_search_name ON t1 USING gin (to_tsvector('text_search.my_text_search_config'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t1', 'name');
ABORT;
-- try again, should not fail with my_text_search_config being retained on the worker
BEGIN;
CREATE TEXT SEARCH CONFIGURATION my_text_search_config ( parser = default );
COMMENT ON TEXT SEARCH CONFIGURATION my_text_search_config IS 'on demand propagation of text search object with a comment';
CREATE TABLE t1(id int, name text);
CREATE INDEX t1_search_name ON t1 USING gin (to_tsvector('text_search.my_text_search_config'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t1', 'name');
SELECT * FROM run_command_on_workers($$
SELECT obj_description('text_search.my_text_search_config'::regconfig);
$$) ORDER BY 1,2;
-- verify that changing anything on a managed TEXT SEARCH CONFIGURATION fails after parallel execution
COMMENT ON TEXT SEARCH CONFIGURATION my_text_search_config IS 'this comment can''t be set right now';
ABORT;
-- create an index on an already distributed table
BEGIN;
CREATE TEXT SEARCH CONFIGURATION my_text_search_config2 ( parser = default );
COMMENT ON TEXT SEARCH CONFIGURATION my_text_search_config2 IS 'on demand propagation of text search object with a comment 2';
CREATE TABLE t1(id int, name text);
SELECT create_distributed_table('t1', 'name');
CREATE INDEX t1_search_name ON t1 USING gin (to_tsvector('text_search.my_text_search_config2'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT * FROM run_command_on_workers($$
SELECT obj_description('text_search.my_text_search_config2'::regconfig);
$$) ORDER BY 1,2;
ABORT;
-- should be able to create a configuration based on a copy of an existing configuration
CREATE TEXT SEARCH CONFIGURATION french_noaccent ( COPY = french );
CREATE TABLE t2(id int, name text);
CREATE INDEX t2_search_name ON t2 USING gin (to_tsvector('text_search.french_noaccent'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t2', 'id');
-- spot check that french_noaccent copied settings from french
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'comment tu t''appelle') WHERE alias = 'asciiword' LIMIT 1;
$$) ORDER BY 1,2;
-- makes no sense, however we expect that the dictionary for the first token changes accordingly
ALTER TEXT SEARCH CONFIGURATION french_noaccent ALTER MAPPING FOR asciiword WITH dutch_stem;
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'comment tu t''appelle') WHERE alias = 'asciiword' LIMIT 1;
$$) ORDER BY 1,2;
-- do the same but we will replace all french dictionaries
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciihword' LIMIT 1;
$$) ORDER BY 1,2;
ALTER TEXT SEARCH CONFIGURATION french_noaccent ALTER MAPPING REPLACE french_stem WITH dutch_stem;
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciihword' LIMIT 1;
$$) ORDER BY 1,2;
-- once more but now back via yet a different DDL command
ALTER TEXT SEARCH CONFIGURATION french_noaccent ALTER MAPPING FOR asciihword REPLACE dutch_stem WITH french_stem;
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciihword' LIMIT 1;
$$) ORDER BY 1,2;
-- drop a mapping
ALTER TEXT SEARCH CONFIGURATION french_noaccent DROP MAPPING FOR asciihword;
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionary) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciihword' LIMIT 1;
$$) ORDER BY 1,2;
-- also with exists, doesn't change anything, but should not error
ALTER TEXT SEARCH CONFIGURATION french_noaccent DROP MAPPING IF EXISTS FOR asciihword;
-- Comment on a text search configuration
COMMENT ON TEXT SEARCH CONFIGURATION french_noaccent IS 'a text configuration that is butcherd to test all edge cases';
SELECT * FROM run_command_on_workers($$
SELECT obj_description('text_search.french_noaccent'::regconfig);
$$) ORDER BY 1,2;
-- Remove a comment
COMMENT ON TEXT SEARCH CONFIGURATION french_noaccent IS NULL;
SELECT * FROM run_command_on_workers($$
SELECT obj_description('text_search.french_noaccent'::regconfig);
$$) ORDER BY 1,2;
-- verify adding 2 dictionaries for two tokes at once
ALTER TEXT SEARCH CONFIGURATION french_noaccent DROP MAPPING IF EXISTS FOR asciiword, asciihword;
ALTER TEXT SEARCH CONFIGURATION french_noaccent ADD MAPPING FOR asciiword, asciihword WITH french_stem, dutch_stem;
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionaries) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciiword' LIMIT 1;
$$) ORDER BY 1,2;
SELECT * FROM run_command_on_workers($$
SELECT ROW(alias,dictionaries) FROM ts_debug('text_search.french_noaccent', 'un chou-fleur') WHERE alias = 'asciihword' LIMIT 1;
$$) ORDER BY 1,2;
--verify we can drop cascade a configuration that is in use
-- verify it is in use
DROP TEXT SEARCH CONFIGURATION text_search.french_noaccent;
-- drop cascade
DROP TEXT SEARCH CONFIGURATION text_search.french_noaccent CASCADE;
-- verify the configuration is dropped from the workers
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.french_noaccent'::regconfig; $$) ORDER BY 1,2;
SET client_min_messages TO 'warning';
SELECT * FROM run_command_on_workers($$CREATE ROLE text_search_owner;$$) ORDER BY 1,2;
CREATE ROLE text_search_owner;
RESET client_min_messages;
CREATE TEXT SEARCH CONFIGURATION changed_owner ( PARSER = default );
SELECT * FROM run_command_on_workers($$
SELECT cfgowner::regrole
FROM pg_ts_config
WHERE oid = 'text_search.changed_owner'::regconfig;
$$) ORDER BY 1,2;
ALTER TEXT SEARCH CONFIGURATION changed_owner OWNER TO text_search_owner;
SELECT * FROM run_command_on_workers($$
SELECT cfgowner::regrole
FROM pg_ts_config
WHERE oid = 'text_search.changed_owner'::regconfig;
$$) ORDER BY 1,2;
-- redo test with propagating object after it was created and changed of owner
SET citus.enable_ddl_propagation TO off;
CREATE TEXT SEARCH CONFIGURATION changed_owner2 ( PARSER = default );
ALTER TEXT SEARCH CONFIGURATION changed_owner2 OWNER TO text_search_owner;
RESET citus.enable_ddl_propagation;
-- verify object doesn't exist before propagating
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.changed_owner2'::regconfig; $$) ORDER BY 1,2;
-- distribute configuration
CREATE TABLE t3(id int, name text);
CREATE INDEX t3_search_name ON t3 USING gin (to_tsvector('text_search.changed_owner2'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t3', 'name');
-- verify config owner
SELECT * FROM run_command_on_workers($$
SELECT cfgowner::regrole
FROM pg_ts_config
WHERE oid = 'text_search.changed_owner2'::regconfig;
$$) ORDER BY 1,2;
-- rename tests
CREATE TEXT SEARCH CONFIGURATION change_name ( PARSER = default );
SELECT * FROM run_command_on_workers($$ -- verify the name exists on the worker
SELECT 'text_search.change_name'::regconfig;
$$) ORDER BY 1,2;
ALTER TEXT SEARCH CONFIGURATION change_name RENAME TO changed_name;
SELECT * FROM run_command_on_workers($$ -- verify the name exists on the worker
SELECT 'text_search.changed_name'::regconfig;
$$) ORDER BY 1,2;
-- test move of schema
CREATE TEXT SEARCH CONFIGURATION change_schema ( PARSER = default );
SELECT * FROM run_command_on_workers($$ -- verify the name exists on the worker
SELECT 'text_search.change_schema'::regconfig;
$$) ORDER BY 1,2;
ALTER TEXT SEARCH CONFIGURATION change_schema SET SCHEMA text_search2;
SELECT * FROM run_command_on_workers($$ -- verify the name exists on the worker
SELECT 'text_search2.change_schema'::regconfig;
$$) ORDER BY 1,2;
-- verify we get an error that the configuration change_schema is not found, even though the object address will be
-- found in its new schema, and is distributed
ALTER TEXT SEARCH CONFIGURATION change_schema SET SCHEMA text_search2;
-- should tell us that text_search.does_not_exist does not exist, covers a complex edgecase
-- in resolving the object address
ALTER TEXT SEARCH CONFIGURATION text_search.does_not_exist SET SCHEMA text_search2;
-- verify edgecases in deparsers
CREATE TEXT SEARCH CONFIGURATION config1 ( PARSER = default );
CREATE TEXT SEARCH CONFIGURATION config2 ( PARSER = default );
SET citus.enable_ddl_propagation TO off;
CREATE TEXT SEARCH CONFIGURATION config3 ( PARSER = default );
RESET citus.enable_ddl_propagation;
-- verify config1, config2 exist on workers, config3 not
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config1'::regconfig; $$) ORDER BY 1,2;
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config2'::regconfig; $$) ORDER BY 1,2;
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config3'::regconfig; $$) ORDER BY 1,2;
-- DROP all config's, only 1&2 are distributed, they should propagate well to remotes
DROP TEXT SEARCH CONFIGURATION config1, config2, config3;
-- verify all existing ones have been removed (checking config3 for consistency)
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config1'::regconfig; $$) ORDER BY 1,2;
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config2'::regconfig; $$) ORDER BY 1,2;
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.config3'::regconfig; $$) ORDER BY 1,2;
-- verify they are all removed locally
SELECT 'text_search.config1'::regconfig;
SELECT 'text_search.config2'::regconfig;
SELECT 'text_search.config3'::regconfig;
-- verify that indexes created concurrently that would propagate a TEXT SEARCH CONFIGURATION object
SET citus.enable_ddl_propagation TO off;
CREATE TEXT SEARCH CONFIGURATION concurrent_index_config ( PARSER = default );
RESET citus.enable_ddl_propagation;
-- verify it doesn't exist on the workers
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.concurrent_index_config'::regconfig; $$) ORDER BY 1,2;
-- create distributed table that then concurrently would have an index created.
CREATE TABLE t4(id int, name text);
SELECT create_distributed_table('t4', 'name');
CREATE INDEX CONCURRENTLY t4_search_name ON t4 USING gin (to_tsvector('text_search.concurrent_index_config'::regconfig, (COALESCE(name, ''::character varying))::text));
-- now the configuration should be on the worker, and the above index creation shouldn't have failed.
SELECT * FROM run_command_on_workers($$ SELECT 'text_search.concurrent_index_config'::regconfig; $$) ORDER BY 1,2;
-- verify the objid is correctly committed locally due to the somewhat convoluted commit and new transaction starting when creating an index concurrently
SELECT pg_catalog.pg_identify_object_as_address(classid, objid, objsubid)
FROM citus.pg_dist_object
WHERE classid = 3602 AND objid = 'text_search.concurrent_index_config'::regconfig::oid;
-- verify old text search configurations get renamed if they are not the same as the newly propagated configuration.
-- We do this by creating configurations on the workers as a copy from a different existing catalog.
SELECT * FROM run_command_on_workers($$
set citus.enable_metadata_sync TO off;
CREATE TEXT SEARCH CONFIGURATION text_search.manually_created_wrongly ( copy = dutch );
reset citus.enable_metadata_sync;
$$) ORDER BY 1,2;
CREATE TEXT SEARCH CONFIGURATION text_search.manually_created_wrongly ( copy = french );
-- now we expect manually_created_wrongly(citus_backup_XXX) to show up when querying the configurations
SELECT * FROM run_command_on_workers($$
SELECT array_agg(cfgname) FROM pg_ts_config WHERE cfgname LIKE 'manually_created_wrongly%';
$$) ORDER BY 1,2;
-- verify the objects get reused appropriately when the specification is the same
SELECT * FROM run_command_on_workers($$
set citus.enable_metadata_sync TO off;
CREATE TEXT SEARCH CONFIGURATION text_search.manually_created_correct ( copy = french );
reset citus.enable_metadata_sync;
$$) ORDER BY 1,2;
CREATE TEXT SEARCH CONFIGURATION text_search.manually_created_correct ( copy = french );
-- now we don't expect manually_created_correct(citus_backup_XXX) to show up when querying the configurations as the
-- original one is reused
SELECT * FROM run_command_on_workers($$
SELECT array_agg(cfgname) FROM pg_ts_config WHERE cfgname LIKE 'manually_created_correct%';
$$) ORDER BY 1,2;
CREATE SCHEMA "Text Search Requiring Quote's";
CREATE TEXT SEARCH CONFIGURATION "Text Search Requiring Quote's"."Quoted Config Name" ( parser = default );
CREATE TABLE t5(id int, name text);
CREATE INDEX t5_search_name ON t5 USING gin (to_tsvector('"Text Search Requiring Quote''s"."Quoted Config Name"'::regconfig, (COALESCE(name, ''::character varying))::text));
SELECT create_distributed_table('t5', 'name');
SET client_min_messages TO 'warning';
DROP SCHEMA text_search, text_search2, "Text Search Requiring Quote's" CASCADE;
DROP ROLE text_search_owner;