Re: POC: Sharing record typmods between backends

Hi,

diff --git a/src/backend/access/common/tupdesc.c b/src/backend/access/common/tupdesc.c
index 9fd7b4e019b..97c0125a4ba 100644
--- a/src/backend/access/common/tupdesc.c
+++ b/src/backend/access/common/tupdesc.c
@@ -337,17 +337,75 @@ DecrTupleDescRefCount(TupleDesc tupdesc)
{
Assert(tupdesc->tdrefcount > 0);

- ResourceOwnerForgetTupleDesc(CurrentResourceOwner, tupdesc);
+ if (CurrentResourceOwner != NULL)
+ ResourceOwnerForgetTupleDesc(CurrentResourceOwner, tupdesc);
if (--tupdesc->tdrefcount == 0)
FreeTupleDesc(tupdesc);
}

What's this about? CurrentResourceOwner should always be valid here, no?
If so, why did that change? I don't think it's good to detach this from
the resowner infrastructure...

/*
- * Compare two TupleDesc structures for logical equality
+ * Compare two TupleDescs' attributes for logical equality
*
* Note: we deliberately do not check the attrelid and tdtypmod fields.
* This allows typcache.c to use this routine to see if a cached record type
* matches a requested type, and is harmless for relcache.c's uses.
+ */
+bool
+equalTupleDescAttrs(Form_pg_attribute attr1, Form_pg_attribute attr2)
+{

comment not really accurate, this routine afaik isn't used by
typcache.c?

/*
- * Magic numbers for parallel state sharing. Higher-level code should use
- * smaller values, leaving these very large ones for use by this module.
+ * Magic numbers for per-context parallel state sharing. Higher-level code
+ * should use smaller values, leaving these very large ones for use by this
+ * module.
*/
#define PARALLEL_KEY_FIXED UINT64CONST(0xFFFFFFFFFFFF0001)
#define PARALLEL_KEY_ERROR_QUEUE UINT64CONST(0xFFFFFFFFFFFF0002)
@@ -63,6 +74,16 @@
#define PARALLEL_KEY_ACTIVE_SNAPSHOT UINT64CONST(0xFFFFFFFFFFFF0007)
#define PARALLEL_KEY_TRANSACTION_STATE UINT64CONST(0xFFFFFFFFFFFF0008)
#define PARALLEL_KEY_ENTRYPOINT UINT64CONST(0xFFFFFFFFFFFF0009)
+#define PARALLEL_KEY_SESSION_DSM UINT64CONST(0xFFFFFFFFFFFF000A)
+
+/* Magic number for per-session DSM TOC. */
+#define PARALLEL_SESSION_MAGIC 0xabb0fbc9
+
+/*
+ * Magic numbers for parallel state sharing in the per-session DSM area.
+ */
+#define PARALLEL_KEY_SESSION_DSA UINT64CONST(0xFFFFFFFFFFFF0001)
+#define PARALLEL_KEY_RECORD_TYPMOD_REGISTRY UINT64CONST(0xFFFFFFFFFFFF0002)

Not this patch's fault, but this infrastructure really isn't great. We
should really replace it with a shmem.h style infrastructure, using a
dht hashtable as backing...

+/* The current per-session DSM segment, if attached. */
+static dsm_segment *current_session_segment = NULL;
+

I think it'd be better if we had a proper 'SessionState' and
'BackendSessionState' infrastructure that then contains the dsm segment
etc. I think we'll otherwise just end up with a bunch of parallel
infrastructures.

+/*
+ * A mechanism for sharing record typmods between backends.
+ */
+struct SharedRecordTypmodRegistry
+{
+ dht_hash_table_handle atts_index_handle;
+ dht_hash_table_handle typmod_index_handle;
+ pg_atomic_uint32 next_typmod;
+};
+

I think the code needs to explain better how these are intended to be
used. IIUC, atts_index is used to find typmods by "identity", and
typmod_index by the typmod, right? And we need both to avoid
all workers generating different tupledescs, right? Kinda guessable by
reading typecache.c, but that shouldn't be needed.

+/*
+ * A flattened/serialized representation of a TupleDesc for use in shared
+ * memory. Can be converted to and from regular TupleDesc format. Doesn't
+ * support constraints and doesn't store the actual type OID, because this is
+ * only for use with RECORD types as created by CreateTupleDesc(). These are
+ * arranged into a linked list, in the hash table entry corresponding to the
+ * OIDs of the first 16 attributes, so we'd expect to get more than one entry
+ * in the list when named and other properties differ.
+ */
+typedef struct SerializedTupleDesc
+{
+ dsa_pointer next; /* next with the same same attribute OIDs */
+ int natts; /* number of attributes in the tuple */
+ int32 typmod; /* typmod for tuple type */
+ bool hasoid; /* tuple has oid attribute in its header */
+
+ /*
+ * The attributes follow. We only ever access the first
+ * ATTRIBUTE_FIXED_PART_SIZE bytes of each element, like the code in
+ * tupdesc.c.
+ */
+ FormData_pg_attribute attributes[FLEXIBLE_ARRAY_MEMBER];
+} SerializedTupleDesc;

Not a fan of a separate tupledesc representation, that's just going to
lead to divergence over time. I think we should rather change the normal
tupledesc representation to be compatible with this, and 'just' have a
wrapper struct for the parallel case (with next and such).

+/*
+ * An entry in SharedRecordTypmodRegistry's attribute index. The key is the
+ * first REC_HASH_KEYS attribute OIDs. That means that collisions are
+ * possible, but that's OK because SerializedTupleDesc objects are arranged
+ * into a list.
+ */

+/* Parameters for SharedRecordTypmodRegistry's attributes hash table. */
+const static dht_parameters srtr_atts_index_params = {
+ sizeof(Oid) * REC_HASH_KEYS,
+ sizeof(SRTRAttsIndexEntry),
+ memcmp,
+ tag_hash,
+ LWTRANCHE_SHARED_RECORD_ATTS_INDEX
+};
+
+/* Parameters for SharedRecordTypmodRegistry's typmod hash table. */
+const static dht_parameters srtr_typmod_index_params = {
+ sizeof(uint32),
+ sizeof(SRTRTypmodIndexEntry),
+ memcmp,
+ tag_hash,
+ LWTRANCHE_SHARED_RECORD_TYPMOD_INDEX
+};
+

I'm very much not a fan of this representation. I know you copied the
logic, but I think it's a bad idea. I think the key should just be a
dsa_pointer, and then we can have a proper tag_hash that hashes the
whole thing, and a proper comparator too. Just have

/*
* Combine two hash values, resulting in another hash value, with decent bit
* mixing.
*
* Similar to boost's hash_combine().
*/
static inline uint32
hash_combine(uint32 a, uint32 b)
{
a ^= b + 0x9e3779b9 + (a << 6) + (a >> 2);
return a;
}

and then hash everything.

+/*
+ * Make sure that RecordCacheArray is large enough to store 'typmod'.
+ */
+static void
+ensure_record_cache_typmod_slot_exists(int32 typmod)
+{
+ if (RecordCacheArray == NULL)
+ {
+ RecordCacheArray = (TupleDesc *)
+ MemoryContextAllocZero(CacheMemoryContext, 64 * sizeof(TupleDesc));
+ RecordCacheArrayLen = 64;
+ }
+
+ if (typmod >= RecordCacheArrayLen)
+ {
+ int32 newlen = RecordCacheArrayLen * 2;
+
+ while (typmod >= newlen)
+ newlen *= 2;
+
+ RecordCacheArray = (TupleDesc *) repalloc(RecordCacheArray,
+ newlen * sizeof(TupleDesc));
+ memset(RecordCacheArray + RecordCacheArrayLen, 0,
+ (newlen - RecordCacheArrayLen) * sizeof(TupleDesc *));
+ RecordCacheArrayLen = newlen;
+ }
+}

Do we really want to keep this? Could just have an equivalent dynahash
for the non-parallel case?

/*
* lookup_rowtype_tupdesc_internal --- internal routine to lookup a rowtype
@@ -1229,15 +1347,49 @@ lookup_rowtype_tupdesc_internal(Oid type_id, int32 typmod, bool noError)
/*
* It's a transient record type, so look in our record-type table.
*/
- if (typmod < 0 || typmod >= NextRecordTypmod)
+ if (typmod >= 0)
{
- if (!noError)
- ereport(ERROR,
- (errcode(ERRCODE_WRONG_OBJECT_TYPE),
- errmsg("record type has not been registered")));
- return NULL;
+ /* It is already in our local cache? */
+ if (typmod < RecordCacheArrayLen &&
+ RecordCacheArray[typmod] != NULL)
+ return RecordCacheArray[typmod];
+
+ /* Are we attached to a SharedRecordTypmodRegistry? */
+ if (CurrentSharedRecordTypmodRegistry.shared != NULL)

Why do we want to do lookups in both? I don't think it's a good idea to
have a chance that you could have the same typmod in both the local
registry (because it'd been created before the shared one) and in the
shared (because it was created in a worker). Ah, that's for caching
purposes? If so, see my above point that we shouldn't have a serialized
version of typdesc (yesyes, constraints will be a bit ugly).

+/*
+ * If we are attached to a SharedRecordTypmodRegistry, find or create a
+ * SerializedTupleDesc that matches 'tupdesc', and return its typmod.
+ * Otherwise return -1.
+ */
+static int32
+find_or_allocate_shared_record_typmod(TupleDesc tupdesc)
+{
+ SRTRAttsIndexEntry *atts_index_entry;
+ SRTRTypmodIndexEntry *typmod_index_entry;
+ SerializedTupleDesc *serialized;
+ dsa_pointer serialized_dp;
+ Oid hashkey[REC_HASH_KEYS];
+ bool found;
+ int32 typmod;
+ int i;
+
+ /* If not even attached, nothing to do. */
+ if (CurrentSharedRecordTypmodRegistry.shared == NULL)
+ return -1;
+
+ /* Try to find a match. */
+ memset(hashkey, 0, sizeof(hashkey));
+ for (i = 0; i < tupdesc->natts; ++i)
+ hashkey[i] = tupdesc->attrs[i]->atttypid;
+ atts_index_entry = (SRTRAttsIndexEntry *)
+ dht_find_or_insert(CurrentSharedRecordTypmodRegistry.atts_index,
+ hashkey,
+ &found);
+ if (!found)
+ {
+ /* Making a new entry. */
+ memcpy(atts_index_entry->leading_attr_oids,
+ hashkey,
+ sizeof(hashkey));
+ atts_index_entry->serialized_tupdesc = InvalidDsaPointer;
+ }
+
+ /* Scan the list we found for a matching serialized one. */
+ serialized_dp = atts_index_entry->serialized_tupdesc;
+ while (DsaPointerIsValid(serialized_dp))
+ {
+ serialized =
+ dsa_get_address(CurrentSharedRecordTypmodRegistry.area,
+ serialized_dp);
+ if (serialized_tupledesc_matches(serialized, tupdesc))
+ {
+ /* Found a match, we are finished. */
+ typmod = serialized->typmod;
+ dht_release(CurrentSharedRecordTypmodRegistry.atts_index,
+ atts_index_entry);
+ return typmod;
+ }
+ serialized_dp = serialized->next;
+ }
+
+ /* We didn't find a matching entry, so let's allocate a new one. */
+ typmod = (int)
+ pg_atomic_fetch_add_u32(&CurrentSharedRecordTypmodRegistry.shared->next_typmod,
+ 1);
+
+ /* Allocate shared memory and serialize the TupleDesc. */
+ serialized_dp = serialize_tupledesc(CurrentSharedRecordTypmodRegistry.area,
+ tupdesc);
+ serialized = (SerializedTupleDesc *)
+ dsa_get_address(CurrentSharedRecordTypmodRegistry.area, serialized_dp);
+ serialized->typmod = typmod;
+
+ /*
+ * While we still hold the atts_index entry locked, add this to
+ * typmod_index. That's important because we don't want anyone to be able
+ * to find a typmod via the former that can't yet be looked up in the
+ * latter.
+ */
+ typmod_index_entry =
+ dht_find_or_insert(CurrentSharedRecordTypmodRegistry.typmod_index,
+ &typmod, &found);
+ if (found)
+ elog(ERROR, "cannot create duplicate shared record typmod");
+ typmod_index_entry->typmod = typmod;
+ typmod_index_entry->serialized_tupdesc = serialized_dp;
+ dht_release(CurrentSharedRecordTypmodRegistry.typmod_index,
+ typmod_index_entry);

What if we fail to allocate memory for the entry in typmod_index?

- Andres