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