diff --git a/src/backend/distributed/commands/foreign_constraint.c b/src/backend/distributed/commands/foreign_constraint.c index 536db0fb674..e60bd2d4d6e 100644 --- a/src/backend/distributed/commands/foreign_constraint.c +++ b/src/backend/distributed/commands/foreign_constraint.c @@ -59,6 +59,19 @@ typedef bool (*CheckRelationFunc)(Oid); +/* + * GUC controlling whether foreign keys between distributed tables require the + * two relations to be colocated. Defaults to true (enforce). When set to false, + * the "relations are not colocated" error is downgraded to a warning so that a + * foreign key can be created even if the referencing/referenced tables have + * (transiently) landed in different colocation groups -- e.g. during a + * mixed-version upgrade before citus_finish_citus_upgrade() has run. Note the + * foreign key cannot actually be enforced across non-colocated shards, so this + * should only be used as a temporary migration aid. + */ +bool EnforceForeignKeyColocation = true; + + /* Local functions forward declarations */ static void EnsureReferencingTableNotReplicated(Oid referencingTableId); static void EnsureSupportedFKeyOnDistKey(Form_pg_constraint constraintForm); @@ -335,7 +348,8 @@ ErrorIfUnsupportedForeignConstraintExists(Relation relation, char referencingDis referencingColocationId == INVALID_COLOCATION_ID || referencingColocationId != referencedColocationId)) { - ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + int colocationErrorLevel = EnforceForeignKeyColocation ? ERROR : WARNING; + ereport(colocationErrorLevel, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot create foreign key constraint since " "relations are not colocated or not referencing " "a reference table"), diff --git a/src/backend/distributed/metadata/metadata_sync.c b/src/backend/distributed/metadata/metadata_sync.c index fbd781f0dd4..40405cc9697 100644 --- a/src/backend/distributed/metadata/metadata_sync.c +++ b/src/backend/distributed/metadata/metadata_sync.c @@ -99,6 +99,32 @@ char *EnableManualMetadataChangesForUser = ""; int MetadataSyncTransMode = METADATA_SYNC_TRANSACTIONAL; +/* + * When true (default), metadata-sync commands sent to workers reference the + * internal helper functions via the citus_internal schema (e.g. + * citus_internal.add_partition_metadata), which was introduced in Citus 13.1. + * When false, the older pg_catalog.citus_internal_* names (e.g. + * pg_catalog.citus_internal_add_partition_metadata) are emitted instead so a + * coordinator can sync metadata to workers running a Citus version that predates + * the citus_internal schema (e.g. 12.1). + */ +bool UseCitusInternalSchema = true; + + +/* + * CitusInternalSchemaPrefix returns the prefix used to qualify the internal + * metadata helper functions in commands propagated to workers. Appending the + * bare function name (e.g. "add_partition_metadata") to this prefix yields a + * fully-qualified callable: either "citus_internal.add_partition_metadata" or + * "pg_catalog.citus_internal_add_partition_metadata" depending on the + * citus.use_citus_internal_schema GUC. + */ +const char * +CitusInternalSchemaPrefix(void) +{ + return UseCitusInternalSchema ? "citus_internal." : "pg_catalog.citus_internal_"; +} + static void EnsureObjectMetadataIsSane(int distributionArgumentIndex, int colocationId); @@ -1009,9 +1035,10 @@ MarkObjectsDistributedCreateCommand(List *addresses, appendStringInfo(insertDistributedObjectsCommand, ") "); appendStringInfo(insertDistributedObjectsCommand, - "SELECT citus_internal.add_object_metadata(" + "SELECT %sadd_object_metadata(" "typetext, objnames, objargs, distargumentindex::int, colocationid::int, force_delegation::bool) " - "FROM distributed_object_data;"); + "FROM distributed_object_data;", + CitusInternalSchemaPrefix()); return insertDistributedObjectsCommand->data; } @@ -1144,8 +1171,9 @@ DistributionCreateCommand(CitusTableCacheEntry *cacheEntry) } appendStringInfo(insertDistributionCommand, - "SELECT citus_internal.add_partition_metadata " + "SELECT %sadd_partition_metadata " "(%s::regclass, '%c', %s, %d, '%c')", + CitusInternalSchemaPrefix(), quote_literal_cstr(qualifiedRelationName), distributionMethod, tablePartitionKeyNameString->data, @@ -1186,7 +1214,8 @@ DistributionDeleteMetadataCommand(Oid relationId) char *qualifiedRelationName = generate_qualified_relation_name(relationId); appendStringInfo(deleteCommand, - "SELECT citus_internal.delete_partition_metadata(%s)", + "SELECT %sdelete_partition_metadata(%s)", + CitusInternalSchemaPrefix(), quote_literal_cstr(qualifiedRelationName)); return deleteCommand->data; @@ -1269,9 +1298,10 @@ ShardListInsertCommand(List *shardIntervalList) appendStringInfo(insertPlacementCommand, ") "); appendStringInfo(insertPlacementCommand, - "SELECT citus_internal.add_placement_metadata(" + "SELECT %sadd_placement_metadata(" "shardid, shardlength, groupid, placementid) " - "FROM placement_data;"); + "FROM placement_data;", + CitusInternalSchemaPrefix()); /* now add shards to insertShardCommand */ StringInfo insertShardCommand = makeStringInfo(); @@ -1325,9 +1355,10 @@ ShardListInsertCommand(List *shardIntervalList) appendStringInfo(insertShardCommand, ") "); appendStringInfo(insertShardCommand, - "SELECT citus_internal.add_shard_metadata(relationname, shardid, " + "SELECT %sadd_shard_metadata(relationname, shardid, " "storagetype, shardminvalue, shardmaxvalue) " - "FROM shard_data;"); + "FROM shard_data;", + CitusInternalSchemaPrefix()); /* * There are no active placements for the table, so do not create the @@ -1364,7 +1395,8 @@ ShardDeleteCommandList(ShardInterval *shardInterval) StringInfo deleteShardCommand = makeStringInfo(); appendStringInfo(deleteShardCommand, - "SELECT citus_internal.delete_shard_metadata(%ld);", shardId); + "SELECT %sdelete_shard_metadata(%ld);", + CitusInternalSchemaPrefix(), shardId); return list_make1(deleteShardCommand->data); } @@ -1434,7 +1466,8 @@ ColocationIdUpdateCommand(Oid relationId, uint32 colocationId) StringInfo command = makeStringInfo(); char *qualifiedRelationName = generate_qualified_relation_name(relationId); appendStringInfo(command, - "SELECT citus_internal.update_relation_colocation(%s::regclass, %d)", + "SELECT %supdate_relation_colocation(%s::regclass, %d)", + CitusInternalSchemaPrefix(), quote_literal_cstr(qualifiedRelationName), colocationId); return command->data; @@ -4433,7 +4466,7 @@ ColocationGroupCreateCommand(uint32 colocationId, int shardCount, int replicatio appendStringInfo(insertColocationCommand, "%s)) " - "SELECT citus_internal.add_colocation_metadata(" + "SELECT %sadd_colocation_metadata(" "colocationid, shardcount, replicationfactor, " "coalesce(t.oid, 0), collationid) " "FROM colocation_data " @@ -4442,7 +4475,8 @@ ColocationGroupCreateCommand(uint32 colocationId, int shardCount, int replicatio "AND (typeschema IS NULL OR " "t.typnamespace = " "(SELECT oid FROM pg_namespace WHERE nspname = typeschema)))", - RemoteCollationIdExpression(distributionColumnCollation)); + RemoteCollationIdExpression(distributionColumnCollation), + CitusInternalSchemaPrefix()); return insertColocationCommand->data; } @@ -4572,7 +4606,8 @@ ColocationGroupDeleteCommand(uint32 colocationId) StringInfo deleteColocationCommand = makeStringInfo(); appendStringInfo(deleteColocationCommand, - "SELECT citus_internal.delete_colocation_metadata(%d)", + "SELECT %sdelete_colocation_metadata(%d)", + CitusInternalSchemaPrefix(), colocationId); return deleteColocationCommand->data; @@ -4588,7 +4623,8 @@ TenantSchemaInsertCommand(Oid schemaId, uint32 colocationId) { StringInfo command = makeStringInfo(); appendStringInfo(command, - "SELECT citus_internal.add_tenant_schema(%s, %u)", + "SELECT %sadd_tenant_schema(%s, %u)", + CitusInternalSchemaPrefix(), RemoteSchemaIdExpressionById(schemaId), colocationId); return command->data; @@ -4604,7 +4640,8 @@ TenantSchemaDeleteCommand(char *schemaName) { StringInfo command = makeStringInfo(); appendStringInfo(command, - "SELECT citus_internal.delete_tenant_schema(%s)", + "SELECT %sdelete_tenant_schema(%s)", + CitusInternalSchemaPrefix(), RemoteSchemaIdExpressionByName(schemaName)); return command->data; @@ -4621,7 +4658,8 @@ UpdateNoneDistTableMetadataCommand(Oid relationId, char replicationModel, { StringInfo command = makeStringInfo(); appendStringInfo(command, - "SELECT citus_internal.update_none_dist_table_metadata(%s, '%c', %u, %s)", + "SELECT %supdate_none_dist_table_metadata(%s, '%c', %u, %s)", + CitusInternalSchemaPrefix(), RemoteTableIdExpression(relationId), replicationModel, colocationId, autoConverted ? "true" : "false"); @@ -4639,7 +4677,8 @@ AddPlacementMetadataCommand(uint64 shardId, uint64 placementId, { StringInfo command = makeStringInfo(); appendStringInfo(command, - "SELECT citus_internal.add_placement_metadata(%ld, %ld, %d, %ld)", + "SELECT %sadd_placement_metadata(%ld, %ld, %d, %ld)", + CitusInternalSchemaPrefix(), shardId, shardLength, groupId, placementId); return command->data; } @@ -4654,7 +4693,8 @@ DeletePlacementMetadataCommand(uint64 placementId) { StringInfo command = makeStringInfo(); appendStringInfo(command, - "SELECT citus_internal.delete_placement_metadata(%ld)", + "SELECT %sdelete_placement_metadata(%ld)", + CitusInternalSchemaPrefix(), placementId); return command->data; } @@ -5307,7 +5347,7 @@ SendColocationMetadataCommands(MetadataSyncContext *context) * the type and its schema to be created first by dependency commands. */ appendStringInfo(colocationGroupCreateCommand, - ") SELECT citus_internal.add_colocation_metadata(" + ") SELECT %sadd_colocation_metadata(" "colocationid, shardcount, replicationfactor, " "coalesce(t.oid, 0), coalesce(c.oid, 0)) " "FROM colocation_group_data d " @@ -5319,7 +5359,8 @@ SendColocationMetadataCommands(MetadataSyncContext *context) "LEFT JOIN pg_collation c " "ON (d.distributioncolumncollationname = c.collname " "AND c.collnamespace = (SELECT oid FROM pg_namespace WHERE " - "nspname = d.distributioncolumncollationschema))"); + "nspname = d.distributioncolumncollationschema))", + CitusInternalSchemaPrefix()); List *commandList = list_make1(colocationGroupCreateCommand->data); SendOrCollectCommandListToActivatedNodes(context, commandList); @@ -5364,7 +5405,8 @@ SendTenantSchemaMetadataCommands(MetadataSyncContext *context) StringInfo insertTenantSchemaCommand = makeStringInfo(); appendStringInfo(insertTenantSchemaCommand, - "SELECT citus_internal.add_tenant_schema(%s, %u)", + "SELECT %sadd_tenant_schema(%s, %u)", + CitusInternalSchemaPrefix(), RemoteSchemaIdExpressionById(tenantSchemaForm->schemaid), tenantSchemaForm->colocationid); diff --git a/src/backend/distributed/planner/multi_logical_optimizer.c b/src/backend/distributed/planner/multi_logical_optimizer.c index 3a4aef85042..30397fad91c 100644 --- a/src/backend/distributed/planner/multi_logical_optimizer.c +++ b/src/backend/distributed/planner/multi_logical_optimizer.c @@ -67,6 +67,15 @@ double CountDistinctErrorRate = 0.0; /* precision of count(distinct) approximate int CoordinatorAggregationStrategy = COORDINATOR_AGGREGATION_ROW_GATHER; bool AllowAggregateWorkerCombineOnInternalTypes = true; +/* + * When false, distributed custom-combine aggregates are pushed down using the + * text-based worker_partial_agg() path instead of worker_binary_partial_agg(). + * worker_binary_partial_agg() was introduced in Citus 14.0, so a coordinator + * running ahead of not-yet-upgraded (e.g. 12.1) workers can set this off to + * keep aggregate pushdown working against those older workers. + */ +bool EnableBinaryWorkerPartialAgg = true; + /* Constant used throughout file */ static const uint32 masterTableId = 1; /* first range table reference on the master node */ @@ -2183,7 +2192,8 @@ MasterAggregateExpression(Aggref *originalAggregate, { aggform = (Form_pg_aggregate) GETSTRUCT(aggTuple); combine = aggform->aggcombinefn; - useBinaryCoordinatorCombine = aggform->aggtranstype != InvalidOid && + useBinaryCoordinatorCombine = EnableBinaryWorkerPartialAgg && + aggform->aggtranstype != InvalidOid && IsAggTransTypeBinarySerializable(aggform); ReleaseSysCache(aggTuple); } @@ -3396,7 +3406,8 @@ WorkerAggregateExpressionList(Aggref *originalAggregate, { aggform = (Form_pg_aggregate) GETSTRUCT(aggTuple); combine = aggform->aggcombinefn; - useBinaryWorkerAggregate = (OidIsValid(aggform->aggtranstype) && + useBinaryWorkerAggregate = (EnableBinaryWorkerPartialAgg && + OidIsValid(aggform->aggtranstype) && IsAggTransTypeBinarySerializable(aggform)); ReleaseSysCache(aggTuple); diff --git a/src/backend/distributed/shared_library_init.c b/src/backend/distributed/shared_library_init.c index ed2f0c7dcee..8102aa6f6ce 100644 --- a/src/backend/distributed/shared_library_init.c +++ b/src/backend/distributed/shared_library_init.c @@ -1007,6 +1007,24 @@ RegisterCitusConfigVariables(void) GUC_STANDARD, NULL, NULL, NULL); + DefineCustomBoolVariable( + "citus.enable_binary_worker_partial_agg", + gettext_noop("Enables pushing down custom-combine aggregates using the " + "binary worker_binary_partial_agg() path."), + gettext_noop( + "When enabled, distributed custom-combine aggregates whose partial " + "state is binary-serializable are pushed down using " + "worker_binary_partial_agg(), which was introduced in Citus 14.0. " + "Turn this off so the coordinator falls back to the text-based " + "worker_partial_agg() path, which lets aggregate pushdown keep " + "working against workers that have not yet been upgraded to a Citus " + "version providing worker_binary_partial_agg()."), + &EnableBinaryWorkerPartialAgg, + true, + PGC_USERSET, + GUC_STANDARD, + NULL, NULL, NULL); + DefineCustomBoolVariable( "citus.allow_modifications_from_workers_to_replicated_tables", gettext_noop("Enables modifications from workers to replicated " @@ -1517,6 +1535,23 @@ RegisterCitusConfigVariables(void) GUC_NO_SHOW_ALL | GUC_NOT_IN_SAMPLE, NULL, NULL, NULL); + DefineCustomBoolVariable( + "citus.use_citus_internal_schema", + gettext_noop("Use the citus_internal schema for internal metadata " + "helper functions in commands sent to workers."), + gettext_noop("When enabled (default), metadata-sync commands reference " + "functions such as citus_internal.add_partition_metadata, " + "which were introduced in Citus 13.1. When disabled, the " + "older pg_catalog.citus_internal_* names are emitted instead " + "so the coordinator can sync metadata to workers running a " + "Citus version that predates the citus_internal schema " + "(e.g. 12.1) -- useful during a mixed-version upgrade."), + &UseCitusInternalSchema, + true, + PGC_USERSET, + GUC_STANDARD, + NULL, NULL, NULL); + DefineCustomBoolVariable( "citus.enable_non_colocated_router_query_pushdown", gettext_noop("Enables router planner for the queries that reference " @@ -1724,6 +1759,24 @@ RegisterCitusConfigVariables(void) GUC_NO_SHOW_ALL | GUC_NOT_IN_SAMPLE, NULL, NULL, NULL); + DefineCustomBoolVariable( + "citus.enforce_foreign_key_colocation", + gettext_noop("Enforce that distributed-to-distributed foreign keys are " + "between colocated relations."), + gettext_noop("When enabled (default), creating a foreign key between two " + "distributed tables that are not colocated raises an error. " + "When disabled, the requirement is downgraded to a warning so " + "the foreign key can still be created -- for example during a " + "mixed-version upgrade before citus_finish_citus_upgrade() has " + "run and tables may have landed in different colocation groups. " + "Citus cannot enforce the foreign key across non-colocated " + "shards, so use this only as a temporary migration aid."), + &EnforceForeignKeyColocation, + true, + PGC_USERSET, + GUC_STANDARD, + NULL, NULL, NULL); + DefineCustomBoolVariable( "citus.enforce_foreign_key_restrictions", gettext_noop("Enforce restrictions while querying distributed/reference " diff --git a/src/include/distributed/commands.h b/src/include/distributed/commands.h index c26b1ff61bb..860cede4693 100644 --- a/src/include/distributed/commands.h +++ b/src/include/distributed/commands.h @@ -36,6 +36,13 @@ extern bool EnableLocalReferenceForeignKeys; */ extern bool AllowUnsafeConstraints; +/* + * GUC that controls whether distributed-to-distributed foreign keys require the + * two relations to be colocated. When false, the colocation requirement is + * downgraded from an error to a warning (migration aid; FK is not enforced). + */ +extern bool EnforceForeignKeyColocation; + extern bool EnableUnsafeTriggers; extern int MaxMatViewSizeToAutoRecreate; diff --git a/src/include/distributed/metadata_sync.h b/src/include/distributed/metadata_sync.h index 86e86ee6d80..15a3e72be8f 100644 --- a/src/include/distributed/metadata_sync.h +++ b/src/include/distributed/metadata_sync.h @@ -227,5 +227,8 @@ extern void SendInterTableRelationshipCommands(MetadataSyncContext *context); /* controlled via GUC */ extern char *EnableManualMetadataChangesForUser; extern bool EnableMetadataSync; +extern bool UseCitusInternalSchema; + +extern const char * CitusInternalSchemaPrefix(void); #endif /* METADATA_SYNC_H */ diff --git a/src/include/distributed/multi_logical_optimizer.h b/src/include/distributed/multi_logical_optimizer.h index 940cbc12358..d98c0e19dd5 100644 --- a/src/include/distributed/multi_logical_optimizer.h +++ b/src/include/distributed/multi_logical_optimizer.h @@ -162,6 +162,7 @@ static const char *const AggregateNames[] = { extern int LimitClauseRowFetchCount; extern double CountDistinctErrorRate; extern int CoordinatorAggregationStrategy; +extern bool EnableBinaryWorkerPartialAgg; /* Function declaration for optimizing logical plans */