|
39 | 39 | #include"utils/tqual.h" |
40 | 40 |
|
41 | 41 |
|
42 | | -staticvoidclone_fk_constraints(Relationpg_constraint,RelationparentRel, |
43 | | -RelationpartRel,List*clone,List**cloned); |
44 | | - |
45 | | - |
46 | 42 | /* |
47 | 43 | * CreateConstraintEntry |
48 | 44 | *Create a constraint table entry. |
@@ -377,306 +373,6 @@ CreateConstraintEntry(const char *constraintName, |
377 | 373 | returnconOid; |
378 | 374 | } |
379 | 375 |
|
380 | | -/* |
381 | | - * CloneForeignKeyConstraints |
382 | | - *Clone foreign keys from a partitioned table to a newly acquired |
383 | | - *partition. |
384 | | - * |
385 | | - * relationId is a partition of parentId, so we can be certain that it has the |
386 | | - * same columns with the same datatypes. The columns may be in different |
387 | | - * order, though. |
388 | | - * |
389 | | - * The *cloned list is appended ClonedConstraint elements describing what was |
390 | | - * created. |
391 | | - */ |
392 | | -void |
393 | | -CloneForeignKeyConstraints(OidparentId,OidrelationId,List**cloned) |
394 | | -{ |
395 | | -Relationpg_constraint; |
396 | | -RelationparentRel; |
397 | | -Relationrel; |
398 | | -ScanKeyDatakey; |
399 | | -SysScanDescscan; |
400 | | -HeapTupletuple; |
401 | | -List*clone=NIL; |
402 | | - |
403 | | -parentRel=heap_open(parentId,NoLock);/* already got lock */ |
404 | | -/* see ATAddForeignKeyConstraint about lock level */ |
405 | | -rel=heap_open(relationId,AccessExclusiveLock); |
406 | | -pg_constraint=heap_open(ConstraintRelationId,RowShareLock); |
407 | | - |
408 | | -/* Obtain the list of constraints to clone or attach */ |
409 | | -ScanKeyInit(&key, |
410 | | -Anum_pg_constraint_conrelid,BTEqualStrategyNumber, |
411 | | -F_OIDEQ,ObjectIdGetDatum(parentId)); |
412 | | -scan=systable_beginscan(pg_constraint,ConstraintRelidTypidNameIndexId, true, |
413 | | -NULL,1,&key); |
414 | | -while ((tuple=systable_getnext(scan))!=NULL) |
415 | | -{ |
416 | | -Oidoid= ((Form_pg_constraint)GETSTRUCT(tuple))->oid; |
417 | | - |
418 | | -clone=lappend_oid(clone,oid); |
419 | | -} |
420 | | -systable_endscan(scan); |
421 | | - |
422 | | -/* Do the actual work, recursing to partitions as needed */ |
423 | | -clone_fk_constraints(pg_constraint,parentRel,rel,clone,cloned); |
424 | | - |
425 | | -/* We're done. Clean up */ |
426 | | -heap_close(parentRel,NoLock); |
427 | | -heap_close(rel,NoLock);/* keep lock till commit */ |
428 | | -heap_close(pg_constraint,RowShareLock); |
429 | | -} |
430 | | - |
431 | | -/* |
432 | | - * clone_fk_constraints |
433 | | - *Recursive subroutine for CloneForeignKeyConstraints |
434 | | - * |
435 | | - * Clone the given list of FK constraints when a partition is attached. |
436 | | - * |
437 | | - * When cloning foreign keys to a partition, it may happen that equivalent |
438 | | - * constraints already exist in the partition for some of them. We can skip |
439 | | - * creating a clone in that case, and instead just attach the existing |
440 | | - * constraint to the one in the parent. |
441 | | - * |
442 | | - * This function recurses to partitions, if the new partition is partitioned; |
443 | | - * of course, only do this for FKs that were actually cloned. |
444 | | - */ |
445 | | -staticvoid |
446 | | -clone_fk_constraints(Relationpg_constraint,RelationparentRel, |
447 | | -RelationpartRel,List*clone,List**cloned) |
448 | | -{ |
449 | | -AttrNumber*attmap; |
450 | | -List*partFKs; |
451 | | -List*subclone=NIL; |
452 | | -ListCell*cell; |
453 | | - |
454 | | -/* |
455 | | - * The constraint key may differ, if the columns in the partition are |
456 | | - * different. This map is used to convert them. |
457 | | - */ |
458 | | -attmap=convert_tuples_by_name_map(RelationGetDescr(partRel), |
459 | | -RelationGetDescr(parentRel), |
460 | | -gettext_noop("could not convert row type")); |
461 | | - |
462 | | -partFKs=copyObject(RelationGetFKeyList(partRel)); |
463 | | - |
464 | | -foreach(cell,clone) |
465 | | -{ |
466 | | -OidparentConstrOid=lfirst_oid(cell); |
467 | | -Form_pg_constraintconstrForm; |
468 | | -HeapTupletuple; |
469 | | -AttrNumberconkey[INDEX_MAX_KEYS]; |
470 | | -AttrNumbermapped_conkey[INDEX_MAX_KEYS]; |
471 | | -AttrNumberconfkey[INDEX_MAX_KEYS]; |
472 | | -Oidconpfeqop[INDEX_MAX_KEYS]; |
473 | | -Oidconppeqop[INDEX_MAX_KEYS]; |
474 | | -Oidconffeqop[INDEX_MAX_KEYS]; |
475 | | -Constraint*fkconstraint; |
476 | | -boolattach_it; |
477 | | -OidconstrOid; |
478 | | -ObjectAddressparentAddr, |
479 | | -childAddr; |
480 | | -intnelem; |
481 | | -ListCell*cell; |
482 | | -inti; |
483 | | - |
484 | | -tuple=SearchSysCache1(CONSTROID,parentConstrOid); |
485 | | -if (!tuple) |
486 | | -elog(ERROR,"cache lookup failed for constraint %u", |
487 | | -parentConstrOid); |
488 | | -constrForm= (Form_pg_constraint)GETSTRUCT(tuple); |
489 | | - |
490 | | -/* only foreign keys */ |
491 | | -if (constrForm->contype!=CONSTRAINT_FOREIGN) |
492 | | -{ |
493 | | -ReleaseSysCache(tuple); |
494 | | -continue; |
495 | | -} |
496 | | - |
497 | | -ObjectAddressSet(parentAddr,ConstraintRelationId,parentConstrOid); |
498 | | - |
499 | | -DeconstructFkConstraintRow(tuple,&nelem,conkey,confkey, |
500 | | -conpfeqop,conppeqop,conffeqop); |
501 | | -for (i=0;i<nelem;i++) |
502 | | -mapped_conkey[i]=attmap[conkey[i]-1]; |
503 | | - |
504 | | -/* |
505 | | - * Before creating a new constraint, see whether any existing FKs are |
506 | | - * fit for the purpose. If one is, attach the parent constraint to it, |
507 | | - * and don't clone anything. This way we avoid the expensive |
508 | | - * verification step and don't end up with a duplicate FK. This also |
509 | | - * means we don't consider this constraint when recursing to |
510 | | - * partitions. |
511 | | - */ |
512 | | -attach_it= false; |
513 | | -foreach(cell,partFKs) |
514 | | -{ |
515 | | -ForeignKeyCacheInfo*fk=lfirst_node(ForeignKeyCacheInfo,cell); |
516 | | -Form_pg_constraintpartConstr; |
517 | | -HeapTuplepartcontup; |
518 | | - |
519 | | -attach_it= true; |
520 | | - |
521 | | -/* |
522 | | - * Do some quick & easy initial checks. If any of these fail, we |
523 | | - * cannot use this constraint, but keep looking. |
524 | | - */ |
525 | | -if (fk->confrelid!=constrForm->confrelid||fk->nkeys!=nelem) |
526 | | -{ |
527 | | -attach_it= false; |
528 | | -continue; |
529 | | -} |
530 | | -for (i=0;i<nelem;i++) |
531 | | -{ |
532 | | -if (fk->conkey[i]!=mapped_conkey[i]|| |
533 | | -fk->confkey[i]!=confkey[i]|| |
534 | | -fk->conpfeqop[i]!=conpfeqop[i]) |
535 | | -{ |
536 | | -attach_it= false; |
537 | | -break; |
538 | | -} |
539 | | -} |
540 | | -if (!attach_it) |
541 | | -continue; |
542 | | - |
543 | | -/* |
544 | | - * Looks good so far; do some more extensive checks. Presumably |
545 | | - * the check for 'convalidated' could be dropped, since we don't |
546 | | - * really care about that, but let's be careful for now. |
547 | | - */ |
548 | | -partcontup=SearchSysCache1(CONSTROID, |
549 | | -ObjectIdGetDatum(fk->conoid)); |
550 | | -if (!partcontup) |
551 | | -elog(ERROR,"cache lookup failed for constraint %u", |
552 | | -fk->conoid); |
553 | | -partConstr= (Form_pg_constraint)GETSTRUCT(partcontup); |
554 | | -if (OidIsValid(partConstr->conparentid)|| |
555 | | -!partConstr->convalidated|| |
556 | | -partConstr->condeferrable!=constrForm->condeferrable|| |
557 | | -partConstr->condeferred!=constrForm->condeferred|| |
558 | | -partConstr->confupdtype!=constrForm->confupdtype|| |
559 | | -partConstr->confdeltype!=constrForm->confdeltype|| |
560 | | -partConstr->confmatchtype!=constrForm->confmatchtype) |
561 | | -{ |
562 | | -ReleaseSysCache(partcontup); |
563 | | -attach_it= false; |
564 | | -continue; |
565 | | -} |
566 | | - |
567 | | -ReleaseSysCache(partcontup); |
568 | | - |
569 | | -/* looks good! Attach this constraint */ |
570 | | -ConstraintSetParentConstraint(fk->conoid,constrForm->oid); |
571 | | -CommandCounterIncrement(); |
572 | | -attach_it= true; |
573 | | -break; |
574 | | -} |
575 | | - |
576 | | -/* |
577 | | - * If we attached to an existing constraint, there is no need to |
578 | | - * create a new one. In fact, there's no need to recurse for this |
579 | | - * constraint to partitions, either. |
580 | | - */ |
581 | | -if (attach_it) |
582 | | -{ |
583 | | -ReleaseSysCache(tuple); |
584 | | -continue; |
585 | | -} |
586 | | - |
587 | | -constrOid= |
588 | | -CreateConstraintEntry(NameStr(constrForm->conname), |
589 | | -constrForm->connamespace, |
590 | | -CONSTRAINT_FOREIGN, |
591 | | -constrForm->condeferrable, |
592 | | -constrForm->condeferred, |
593 | | -constrForm->convalidated, |
594 | | -constrForm->oid, |
595 | | -RelationGetRelid(partRel), |
596 | | -mapped_conkey, |
597 | | -nelem, |
598 | | -nelem, |
599 | | -InvalidOid,/* not a domain constraint */ |
600 | | -constrForm->conindid,/* same index */ |
601 | | -constrForm->confrelid,/* same foreign rel */ |
602 | | -confkey, |
603 | | -conpfeqop, |
604 | | -conppeqop, |
605 | | -conffeqop, |
606 | | -nelem, |
607 | | -constrForm->confupdtype, |
608 | | -constrForm->confdeltype, |
609 | | -constrForm->confmatchtype, |
610 | | -NULL, |
611 | | -NULL, |
612 | | -NULL, |
613 | | - false, |
614 | | -1, false, true); |
615 | | -subclone=lappend_oid(subclone,constrOid); |
616 | | - |
617 | | -ObjectAddressSet(childAddr,ConstraintRelationId,constrOid); |
618 | | -recordDependencyOn(&childAddr,&parentAddr,DEPENDENCY_INTERNAL_AUTO); |
619 | | - |
620 | | -fkconstraint=makeNode(Constraint); |
621 | | -/* for now this is all we need */ |
622 | | -fkconstraint->conname=pstrdup(NameStr(constrForm->conname)); |
623 | | -fkconstraint->fk_upd_action=constrForm->confupdtype; |
624 | | -fkconstraint->fk_del_action=constrForm->confdeltype; |
625 | | -fkconstraint->deferrable=constrForm->condeferrable; |
626 | | -fkconstraint->initdeferred=constrForm->condeferred; |
627 | | - |
628 | | -createForeignKeyTriggers(partRel,constrForm->confrelid,fkconstraint, |
629 | | -constrOid,constrForm->conindid, false); |
630 | | - |
631 | | -if (cloned) |
632 | | -{ |
633 | | -ClonedConstraint*newc; |
634 | | - |
635 | | -/* |
636 | | - * Feed back caller about the constraints we created, so that they |
637 | | - * can set up constraint verification. |
638 | | - */ |
639 | | -newc=palloc(sizeof(ClonedConstraint)); |
640 | | -newc->relid=RelationGetRelid(partRel); |
641 | | -newc->refrelid=constrForm->confrelid; |
642 | | -newc->conindid=constrForm->conindid; |
643 | | -newc->conid=constrOid; |
644 | | -newc->constraint=fkconstraint; |
645 | | - |
646 | | -*cloned=lappend(*cloned,newc); |
647 | | -} |
648 | | - |
649 | | -ReleaseSysCache(tuple); |
650 | | -} |
651 | | - |
652 | | -pfree(attmap); |
653 | | -list_free_deep(partFKs); |
654 | | - |
655 | | -/* |
656 | | - * If the partition is partitioned, recurse to handle any constraints that |
657 | | - * were cloned. |
658 | | - */ |
659 | | -if (partRel->rd_rel->relkind==RELKIND_PARTITIONED_TABLE&& |
660 | | -subclone!=NIL) |
661 | | -{ |
662 | | -PartitionDescpartdesc=RelationGetPartitionDesc(partRel); |
663 | | -inti; |
664 | | - |
665 | | -for (i=0;i<partdesc->nparts;i++) |
666 | | -{ |
667 | | -RelationchildRel; |
668 | | - |
669 | | -childRel=heap_open(partdesc->oids[i],AccessExclusiveLock); |
670 | | -clone_fk_constraints(pg_constraint, |
671 | | -partRel, |
672 | | -childRel, |
673 | | -subclone, |
674 | | -cloned); |
675 | | -heap_close(childRel,NoLock);/* keep lock till commit */ |
676 | | -} |
677 | | -} |
678 | | -} |
679 | | - |
680 | 376 | /* |
681 | 377 | * Test whether given name is currently used as a constraint name |
682 | 378 | * for the given object (relation or domain). |
|