Skip to content

Allow for citus 12.1 and citus 14 nodes in the same cluster #8698

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
DmitriMelnikov wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Allow for citus 12.1 and citus 14 nodes in the same cluster #8698

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion src/backend/distributed/commands/foreign_constraint.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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"),
Comment on lines +351 to 355
Expand Down
84 changes: 63 additions & 21 deletions src/backend/distributed/metadata/metadata_sync.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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 "
Expand All @@ -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;
}
Expand Down Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand All @@ -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");

Expand All @@ -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;
}
Expand All @@ -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;
}
Expand Down Expand Up @@ -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 "
Expand All @@ -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);
Expand Down Expand Up @@ -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);

Expand Down
15 changes: 13 additions & 2 deletions src/backend/distributed/planner/multi_logical_optimizer.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 */

Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -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);
Expand Down
53 changes: 53 additions & 0 deletions src/backend/distributed/shared_library_init.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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."),
Comment on lines +1011 to +1013
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,

@neildsh neildsh Jul 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should all of these GUCs be GUC_NO_SHOW_ALL ?

NULL, NULL, NULL);

DefineCustomBoolVariable(
"citus.allow_modifications_from_workers_to_replicated_tables",
gettext_noop("Enables modifications from workers to replicated "
Expand Down Expand Up @@ -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 "
Expand Down Expand Up @@ -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 "
Expand Down
7 changes: 7 additions & 0 deletions src/include/distributed/commands.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
3 changes: 3 additions & 0 deletions src/include/distributed/metadata_sync.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

@neildsh neildsh Jul 28, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this need to be extern? If it is used only in metadata_sync.c, you can just make it a local static function


#endif /* METADATA_SYNC_H */
1 change: 1 addition & 0 deletions src/include/distributed/multi_logical_optimizer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 */
Expand Down
Loading