Re: COPY FROM WHEN condition

On 2019-04-03 06:41:49 +1300, David Rowley wrote:
> However, I've ended up not doing it that way as the patch requires
> more than just an array of TupleTableSlots to be stored in the
> ResultRelInfo, it'll pretty much need all of what I have in
> CopyMultiInsertBuffer, which includes line numbers for error
> reporting, a BulkInsertState and a counter to track how many of the
> slots are used. I had thoughts that we could just tag the whole
> CopyMultiInsertBuffer in ResultRelInfo, but that requires moving it
> somewhere a bit more generic. Another thought would be that we have
> something like "void *extra;" in ResultRelInfo that we can just borrow
> for copy.c. It may be interesting to try this out to see if it saves
> much in the way of performance.

Hm, we could just forwad declare struct CopyMultiInsertBuffer in a
header, and only define it in copy.c. That doesn't sound insane to me.

> I've got the patch into a working state now and wrote a bunch of
> comments. I've not really done a thorough read back of the code again.
> I normally wait until the light is coming from a different angle
> before doing that, but there does not seem to be much time to wait for
> that in this case, so here's v2. Performance seems to be about the
> same as what I reported yesterday.

Cool.

> +/*
> + * Multi-Insert handling code for COPY FROM. Inserts are performed in
> + * batches of up to MAX_BUFFERED_TUPLES.

This should probably be moved to the top of the file.

> + * When COPY FROM is used on a partitioned table, we attempt to maintain
> + * only MAX_PARTITION_BUFFERS buffers at once. Buffers that are completely
> + * unused in each batch are removed so that we end up not keeping buffers for
> + * partitions that we might not insert into again.
> + */
> +#define MAX_BUFFERED_TUPLES 1000
> +#define MAX_BUFFERED_BYTES 65535
> +#define MAX_PARTITION_BUFFERS 15

> +/*
> + * CopyMultiInsertBuffer
> + * Stores multi-insert data related to a single relation in CopyFrom.
> + */
> +typedef struct

Please don't create anonymous structs - they can't be forward declared,
and some debugging tools handle them worse than named structs. And it
makes naming CopyMultiInsertBuffer in the comment less necessary.

> +{
> + Oid relid; /* Relation ID this insert data is for */
> + TupleTableSlot **slots; /* Array of MAX_BUFFERED_TUPLES to store
> + * tuples */
> + uint64 *linenos; /* Line # of tuple in copy stream */

It could make sense to allocate those two together, or even as part of
the CopyMultiInsertBuffer itself, to reduce allocator overhead.

> + else
> + {
> + CopyMultiInsertBuffer *buffer = (CopyMultiInsertBuffer *) palloc(sizeof(CopyMultiInsertBuffer));
> +
> + /* Non-partitioned table. Just setup the buffer for the table. */
> + buffer->relid = RelationGetRelid(rri->ri_RelationDesc);
> + buffer->slots = palloc0(MAX_BUFFERED_TUPLES * sizeof(TupleTableSlot *));
> + buffer->linenos = palloc(MAX_BUFFERED_TUPLES * sizeof(uint64));
> + buffer->resultRelInfo = rri;
> + buffer->bistate = GetBulkInsertState();
> + buffer->nused = 0;
> + miinfo->multiInsertBufferTab = NULL;
> + miinfo->buffer = buffer;
> + miinfo->nbuffers = 1;
> + }

Can this be moved into a helper function?

> + /*
> + * heap_multi_insert leaks memory, so switch to short-lived memory context
> + * before calling it.
> + */

s/heap_multi_insert/table_multi_insert/

> + /*
> + * If there are any indexes, update them for all the inserted tuples, and
> + * run AFTER ROW INSERT triggers.
> + */
> + if (resultRelInfo->ri_NumIndices > 0)
> + {
> + for (i = 0; i < nBufferedTuples; i++)
> + {
> + List *recheckIndexes;
> +
> + cstate->cur_lineno = buffer->linenos[i];
> + recheckIndexes =
> + ExecInsertIndexTuples(buffer->slots[i], estate, false, NULL,
> + NIL);
> + ExecARInsertTriggers(estate, resultRelInfo,
> + buffer->slots[i],
> + recheckIndexes, cstate->transition_capture);
> + list_free(recheckIndexes);
> + }
> + }
> +
> + /*
> + * There's no indexes, but see if we need to run AFTER ROW INSERT triggers
> + * anyway.
> + */
> + else if (resultRelInfo->ri_TrigDesc != NULL &&
> + (resultRelInfo->ri_TrigDesc->trig_insert_after_row ||
> + resultRelInfo->ri_TrigDesc->trig_insert_new_table))
> + {
> + for (i = 0; i < nBufferedTuples; i++)
> + {
> + cstate->cur_lineno = buffer->linenos[i];
> + ExecARInsertTriggers(estate, resultRelInfo,
> + bufferedSlots[i],
> + NIL, cstate->transition_capture);
> + }
> + }

> + for (i = 0; i < nBufferedTuples; i++)
> + ExecClearTuple(bufferedSlots[i]);

I wonder about combining these loops somehow. But it's probably ok.

> +/*
> + * CopyMultiInsertBuffer_Cleanup
> + * Drop used slots and free member for this buffer. The buffer
> + * must be flushed before cleanup.
> + */
> +static inline void
> +CopyMultiInsertBuffer_Cleanup(CopyMultiInsertBuffer *buffer)
> +{
> + int i;
> +
> + ReleaseBulkInsertStatePin(buffer->bistate);

Shouldn't this FreeBulkInsertState() rather than
ReleaseBulkInsertStatePin()?

> +
> +/*
> + * CopyMultiInsertBuffer_RemoveBuffer
> + * Remove a buffer from being tracked by miinfo
> + */
> +static inline void
> +CopyMultiInsertBuffer_RemoveBuffer(CopyMultiInsertInfo *miinfo,
> + CopyMultiInsertBuffer *buffer)
> +{
> + Oid relid = buffer->relid;
> +
> + CopyMultiInsertBuffer_Cleanup(buffer);
> +
> + hash_search(miinfo->multiInsertBufferTab, (void *) &relid, HASH_REMOVE,
> + NULL);
> + miinfo->nbuffers--;

Aren't we leaking the CopyMultiInsertBuffer itself here?

> +/*
> + * CopyMultiInsertInfo_Flush
> + * Write out all stored tuples in all buffers out to the tables.
> + *
> + * To save us from ending up with buffers for 1000s of partitions we remove
> + * buffers belonging to partitions that we've seen no tuples for in this batch

That seems a little naive (imagine you have like 5 partitions, and we
constantly cycle through 2-3 of them per batch). It's probably OK for
this version. I guess I'd only do that cleanup if we're actually
flushing due to the number of partitions.

> if (cstate->rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
> {
> ereport(ERROR,
> - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
> - errmsg("cannot perform FREEZE on a partitioned table")));
> + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
> + errmsg("cannot perform FREEZE on a partitioned table")));
> }

accidental hunk?

Greetings,

Andres Freund