@@ -35,6 +35,24 @@ static MemoryContext LogicalRepRelMapContext = NULL;
3535static HTAB * LogicalRepRelMap = NULL ;
3636static HTAB * LogicalRepTypMap = NULL ;
3737
38+ /*
39+ * Partition map (LogicalRepPartMap)
40+ *
41+ * When a partitioned table is used as replication target, replicated
42+ * operations are actually performed on its leaf partitions, which requires
43+ * the partitions to also be mapped to the remote relation. Parent's entry
44+ * (LogicalRepRelMapEntry) cannot be used as-is for all partitions, because
45+ * individual partitions may have different attribute numbers, which means
46+ * attribute mappings to remote relation's attributes must be maintained
47+ * separately for each partition.
48+ */
49+ static MemoryContext LogicalRepPartMapContext = NULL ;
50+ static HTAB * LogicalRepPartMap = NULL ;
51+ typedef struct LogicalRepPartMapEntry
52+ {
53+ Oid partoid ;/* LogicalRepPartMap's key */
54+ LogicalRepRelMapEntry relmapentry ;
55+ }LogicalRepPartMapEntry ;
3856
3957/*
4058 * Relcache invalidation callback for our relation map cache.
@@ -472,3 +490,174 @@ logicalrep_typmap_gettypname(Oid remoteid)
472490Assert (OidIsValid (entry -> remoteid ));
473491return psprintf ("%s.%s" ,entry -> nspname ,entry -> typname );
474492}
493+
494+ /*
495+ * Partition cache: look up partition LogicalRepRelMapEntry's
496+ *
497+ * Unlike relation map cache, this is keyed by partition OID, not remote
498+ * relation OID, because we only have to use this cache in the case where
499+ * partitions are not directly mapped to any remote relation, such as when
500+ * replication is occurring with one of their ancestors as target.
501+ */
502+
503+ /*
504+ * Relcache invalidation callback
505+ */
506+ static void
507+ logicalrep_partmap_invalidate_cb (Datum arg ,Oid reloid )
508+ {
509+ LogicalRepRelMapEntry * entry ;
510+
511+ /* Just to be sure. */
512+ if (LogicalRepPartMap == NULL )
513+ return ;
514+
515+ if (reloid != InvalidOid )
516+ {
517+ HASH_SEQ_STATUS status ;
518+
519+ hash_seq_init (& status ,LogicalRepPartMap );
520+
521+ /* TODO, use inverse lookup hashtable? */
522+ while ((entry = (LogicalRepRelMapEntry * )hash_seq_search (& status ))!= NULL )
523+ {
524+ if (entry -> localreloid == reloid )
525+ {
526+ entry -> localreloid = InvalidOid ;
527+ hash_seq_term (& status );
528+ break ;
529+ }
530+ }
531+ }
532+ else
533+ {
534+ /* invalidate all cache entries */
535+ HASH_SEQ_STATUS status ;
536+
537+ hash_seq_init (& status ,LogicalRepPartMap );
538+
539+ while ((entry = (LogicalRepRelMapEntry * )hash_seq_search (& status ))!= NULL )
540+ entry -> localreloid = InvalidOid ;
541+ }
542+ }
543+
544+ /*
545+ * Initialize the partition map cache.
546+ */
547+ static void
548+ logicalrep_partmap_init (void )
549+ {
550+ HASHCTL ctl ;
551+
552+ if (!LogicalRepPartMapContext )
553+ LogicalRepPartMapContext =
554+ AllocSetContextCreate (CacheMemoryContext ,
555+ "LogicalRepPartMapContext" ,
556+ ALLOCSET_DEFAULT_SIZES );
557+
558+ /* Initialize the relation hash table. */
559+ MemSet (& ctl ,0 ,sizeof (ctl ));
560+ ctl .keysize = sizeof (Oid );/* partition OID */
561+ ctl .entrysize = sizeof (LogicalRepPartMapEntry );
562+ ctl .hcxt = LogicalRepPartMapContext ;
563+
564+ LogicalRepPartMap = hash_create ("logicalrep partition map cache" ,64 ,& ctl ,
565+ HASH_ELEM |HASH_BLOBS |HASH_CONTEXT );
566+
567+ /* Watch for invalidation events. */
568+ CacheRegisterRelcacheCallback (logicalrep_partmap_invalidate_cb ,
569+ (Datum )0 );
570+ }
571+
572+ /*
573+ * logicalrep_partition_open
574+ *
575+ * Returned entry reuses most of the values of the root table's entry, save
576+ * the attribute map, which can be different for the partition.
577+ *
578+ * Note there's no logialrep_partition_close, because the caller closes the
579+ * the component relation.
580+ */
581+ LogicalRepRelMapEntry *
582+ logicalrep_partition_open (LogicalRepRelMapEntry * root ,
583+ Relation partrel ,AttrMap * map )
584+ {
585+ LogicalRepRelMapEntry * entry ;
586+ LogicalRepPartMapEntry * part_entry ;
587+ LogicalRepRelation * remoterel = & root -> remoterel ;
588+ Oid partOid = RelationGetRelid (partrel );
589+ AttrMap * attrmap = root -> attrmap ;
590+ bool found ;
591+ int i ;
592+ MemoryContext oldctx ;
593+
594+ if (LogicalRepPartMap == NULL )
595+ logicalrep_partmap_init ();
596+
597+ /* Search for existing entry. */
598+ part_entry = (LogicalRepPartMapEntry * )hash_search (LogicalRepPartMap ,
599+ (void * )& partOid ,
600+ HASH_ENTER ,& found );
601+
602+ if (found )
603+ return & part_entry -> relmapentry ;
604+
605+ memset (part_entry ,0 ,sizeof (LogicalRepPartMapEntry ));
606+
607+ /* Switch to longer-lived context. */
608+ oldctx = MemoryContextSwitchTo (LogicalRepPartMapContext );
609+
610+ part_entry -> partoid = partOid ;
611+
612+ /* Remote relation is used as-is from the root entry. */
613+ entry = & part_entry -> relmapentry ;
614+ entry -> remoterel .remoteid = remoterel -> remoteid ;
615+ entry -> remoterel .nspname = pstrdup (remoterel -> nspname );
616+ entry -> remoterel .relname = pstrdup (remoterel -> relname );
617+ entry -> remoterel .natts = remoterel -> natts ;
618+ entry -> remoterel .attnames = palloc (remoterel -> natts * sizeof (char * ));
619+ entry -> remoterel .atttyps = palloc (remoterel -> natts * sizeof (Oid ));
620+ for (i = 0 ;i < remoterel -> natts ;i ++ )
621+ {
622+ entry -> remoterel .attnames [i ]= pstrdup (remoterel -> attnames [i ]);
623+ entry -> remoterel .atttyps [i ]= remoterel -> atttyps [i ];
624+ }
625+ entry -> remoterel .replident = remoterel -> replident ;
626+ entry -> remoterel .attkeys = bms_copy (remoterel -> attkeys );
627+
628+ entry -> localrel = partrel ;
629+ entry -> localreloid = partOid ;
630+
631+ /*
632+ * If the partition's attributes don't match the root relation's, we'll
633+ * need to make a new attrmap which maps partition attribute numbers to
634+ * remoterel's, instead the original which maps root relation's attribute
635+ * numbers to remoterel's.
636+ *
637+ * Note that 'map' which comes from the tuple routing data structure
638+ * contains 1-based attribute numbers (of the parent relation). However,
639+ * the map in 'entry', a logical replication data structure, contains
640+ * 0-based attribute numbers (of the remote relation).
641+ */
642+ if (map )
643+ {
644+ AttrNumber attno ;
645+
646+ entry -> attrmap = make_attrmap (map -> maplen );
647+ for (attno = 0 ;attno < entry -> attrmap -> maplen ;attno ++ )
648+ {
649+ AttrNumber root_attno = map -> attnums [attno ];
650+
651+ entry -> attrmap -> attnums [attno ]= attrmap -> attnums [root_attno - 1 ];
652+ }
653+ }
654+ else
655+ entry -> attrmap = attrmap ;
656+
657+ entry -> updatable = root -> updatable ;
658+
659+ /* state and statelsn are left set to 0. */
660+ MemoryContextSwitchTo (oldctx );
661+
662+ return entry ;
663+ }