-
Notifications
You must be signed in to change notification settings - Fork 671
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
b335134
commit e9365fa
Showing
31 changed files
with
1,431 additions
and
22 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
233 changes: 233 additions & 0 deletions
233
src/backend/distributed/commands/schema_based_sharding.c
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,233 @@ | ||
| /*------------------------------------------------------------------------- | ||
| * schema_based_sharding.c | ||
| * | ||
| * Routines for schema-based sharding. | ||
| * | ||
| *------------------------------------------------------------------------- | ||
| */ | ||
|
|
||
| #include "postgres.h" | ||
| #include "miscadmin.h" | ||
| #include "catalog/pg_namespace_d.h" | ||
| #include "distributed/backend_data.h" | ||
| #include "distributed/colocation_utils.h" | ||
| #include "distributed/commands.h" | ||
| #include "distributed/metadata_cache.h" | ||
| #include "distributed/metadata_sync.h" | ||
| #include "distributed/tenant_schema_metadata.h" | ||
| #include "distributed/metadata/distobject.h" | ||
| #include "utils/lsyscache.h" | ||
|
|
||
|
|
||
| /* controlled via citus.enable_schema_based_sharding GUC */ | ||
| bool EnableSchemaBasedSharding = false; | ||
|
|
||
|
|
||
| static void SetTenantSchemaColocationId(Oid schemaId, uint32 colocationId); | ||
|
|
||
|
|
||
| /* | ||
| * ShouldUseSchemaAsTenantSchema returns true if schema given name should be | ||
| * used as a tenant schema. | ||
| */ | ||
| bool | ||
| ShouldUseSchemaAsTenantSchema(char *schemaName) | ||
| { | ||
| if (!EnableSchemaBasedSharding) | ||
| { | ||
| return false; | ||
| } | ||
|
|
||
| if (IsBinaryUpgrade) | ||
| { | ||
| return false; | ||
| } | ||
|
|
||
| /* | ||
| * CREATE SCHEMA commands issued by internal backends are not meant to | ||
| * create tenant schemas but to sync metadata. | ||
| */ | ||
| if (IsCitusInternalBackend() || IsRebalancerInternalBackend()) | ||
| { | ||
| return false; | ||
| } | ||
|
|
||
| /* | ||
| * Not do an oid comparison based on PG_PUBLIC_NAMESPACE because | ||
| * we want to treat "public" schema in the same way even if it's | ||
| * recreated. | ||
| */ | ||
| if (strcmp(schemaName, "public") == 0) | ||
| { | ||
| return false; | ||
| } | ||
|
|
||
| return true; | ||
| } | ||
|
|
||
|
|
||
| /* | ||
| * ShouldCreateTenantTable returns true if given parsetree is a CREATE TABLE | ||
| * statement and the table should be treated as a tenant table. | ||
| */ | ||
| bool | ||
| ShouldCreateTenantTable(Oid relationId) | ||
| { | ||
| if (IsBinaryUpgrade) | ||
| { | ||
| return false; | ||
| } | ||
|
|
||
| /* | ||
| * CREATE TABLE commands issued by internal backends are not meant to | ||
| * create tenant tables but to sync metadata. | ||
| */ | ||
| if (IsCitusInternalBackend() || IsRebalancerInternalBackend()) | ||
| { | ||
| return false; | ||
| } | ||
|
|
||
| Oid schemaId = get_rel_namespace(relationId); | ||
| ObjectAddress *schemaAddress = palloc0(sizeof(ObjectAddress)); | ||
| ObjectAddressSet(*schemaAddress, NamespaceRelationId, schemaId); | ||
| if (IsAnyObjectAddressOwnedByExtension(list_make1(schemaAddress), NULL)) | ||
| { | ||
| return false; | ||
| } | ||
|
|
||
| return IsTenantSchema(schemaId); | ||
| } | ||
|
|
||
|
|
||
| /* | ||
| * CreateTenantTable creates a tenant table with given relationId. | ||
| * | ||
| * This means creating a single shard distributed table without a shard | ||
| * key and colocating it with the other tables in its schema. | ||
| */ | ||
| void | ||
| CreateTenantTable(Oid relationId) | ||
| { | ||
| CheckCitusVersion(ERROR); | ||
|
|
||
| if (!IsCoordinator()) | ||
| { | ||
| /* | ||
| * We don't support creating tenant tables from workers. We could | ||
| * let ShouldCreateTenantTable() to return false to allow users to | ||
| * create a local table as usual but that would be confusing because | ||
| * it might sound like we allow creating tenant tables from workers. | ||
| * For this reason, we prefer to throw an error instead. | ||
| * | ||
| * Indeed, CreateNullShardKeyDistTable() would already do so but we | ||
| * prefer to throw an error with a more meaningful message, rather | ||
| * than saying "operation is not allowed on this node". | ||
| */ | ||
| ereport(ERROR, (errmsg("cannot create a tenant table from a worker node"), | ||
| errhint("Connect to the coordinator node and try again."))); | ||
| } | ||
|
|
||
| /* | ||
| * Decide name of the table with lowest oid in the colocation group | ||
| * and use it as the colocate_with parameter. | ||
| */ | ||
| char *colocateWithTableName = "none"; | ||
|
|
||
| /* | ||
| * Acquire default colocation lock to prevent concurrently forming | ||
| * multiple colocation groups for the same schema. | ||
| * | ||
| * Note that the lock is based on schemaId to avoid serializing | ||
| * default colocation group creation for all tenant schemas. | ||
| */ | ||
| Oid schemaId = get_rel_namespace(relationId); | ||
| AcquireCitusTenantSchemaDefaultColocationLock(schemaId); | ||
|
|
||
| uint32 colocationId = SchemaIdGetTenantColocationId(schemaId); | ||
| bool firstTableInSchema = (colocationId == INVALID_COLOCATION_ID); | ||
| if (!firstTableInSchema) | ||
| { | ||
| /* | ||
| * Release the lock if the schema is already associated with a | ||
| * colocation group. | ||
| */ | ||
| ReleaseCitusTenantSchemaDefaultColocationLock(schemaId); | ||
|
|
||
| Oid colocateWithTableId = ColocationGroupGetTableWithLowestOid(colocationId); | ||
| colocateWithTableName = generate_qualified_relation_name(colocateWithTableId); | ||
| } | ||
|
|
||
| CreateNullShardKeyDistTable(relationId, colocateWithTableName); | ||
|
|
||
| if (firstTableInSchema) | ||
| { | ||
| /* | ||
| * Save it into pg_dist_tenant_schema if this is the first tenant | ||
| * table in the schema. | ||
| */ | ||
| SetTenantSchemaColocationId(schemaId, TableColocationId(relationId)); | ||
| } | ||
| } | ||
|
|
||
|
|
||
| /* | ||
| * RegisterTenantSchema registers given schema as a tenant schema locally and | ||
| * returns the command to do the same on the workers. | ||
| */ | ||
| char * | ||
| RegisterTenantSchema(Oid schemaId) | ||
| { | ||
| CheckCitusVersion(ERROR); | ||
|
|
||
| /* not assign a colocation id until creating the first table */ | ||
| uint32 colocationId = INVALID_COLOCATION_ID; | ||
|
|
||
| InsertTenantSchemaLocally(schemaId, colocationId); | ||
|
|
||
| return TenantSchemaInsertCommand(schemaId, colocationId); | ||
| } | ||
|
|
||
|
|
||
| /* | ||
| * UnregisterTenantSchema deletes tenant schema metadata related to given | ||
| * schema. | ||
| */ | ||
| void | ||
| UnregisterTenantSchema(Oid schemaId) | ||
| { | ||
| DeleteTenantSchemaLocally(schemaId); | ||
|
|
||
| SendCommandToWorkersWithMetadataViaSuperUser( | ||
| TenantSchemaDeleteCommand(schemaId)); | ||
| } | ||
|
|
||
|
|
||
| /* | ||
| * DisassociateTenantSchemaIfAny disassociates given colocation id from its | ||
| * tenant schema if any. | ||
| */ | ||
| void | ||
| DisassociateTenantSchemaIfAny(uint32 colocationId) | ||
| { | ||
| Oid schemaId = ColocationIdGetTenantSchemaId(colocationId); | ||
| if (OidIsValid(schemaId)) | ||
| { | ||
| SetTenantSchemaColocationId(schemaId, INVALID_COLOCATION_ID); | ||
| } | ||
| } | ||
|
|
||
|
|
||
| /* | ||
| * SetTenantSchemaColocationId sets the colocation id of given tenant schema. | ||
| * | ||
| * If colocationId is INVALID_COLOCATION_ID, then the colocation_id column | ||
| * is set to NULL. | ||
| */ | ||
| static void | ||
| SetTenantSchemaColocationId(Oid schemaId, uint32 colocationId) | ||
| { | ||
| SetTenantSchemaColocationIdLocally(schemaId, colocationId); | ||
|
|
||
| SendCommandToWorkersWithMetadataViaSuperUser( | ||
| TenantSchemaSetColocationIdCommand(schemaId, colocationId)); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.