Parallel Inserts in CREATE TABLE AS

> Pass the into clause, object id, command id from the leader to
> workers, so that each worker can create its own CTAS dest
> receiver. Leader inserts it's share of tuples if instructed to do, and
> so are workers. Each worker writes atomically it's number of inserted
> tuples into a shared memory variable, the leader combines this with
> it's own number of inserted tuples and shares to the client.
>
> Below things are still pending. Thoughts are most welcome:
> 1. How better we can lift the "cannot insert tuples in a parallel worker"
> from heap_prepare_insert() for only CTAS cases or for that matter parallel
> copy? How about having a variable in any of the worker global contexts and
> use that? Of course, we can remove this restriction entirely in case we
> fully allow parallelism for INSERT INTO SELECT, CTAS, and COPY.

>
> > The design:
>
> I think it'd be good if you could explain a bit more why you think this
> safe to do in the way you have done it.
>
> E.g. from a quick scroll through the patch, there's not even a comment
> explaining that the only reason there doesn't need to be code dealing
> with xid assignment because we already did the catalog changes to create
> the table.
>

>
> > Below things are still pending. Thoughts are most welcome:
> > 1. How better we can lift the "cannot insert tuples in a parallel worker"
> > from heap_prepare_insert() for only CTAS cases or for that matter parallel
> > copy? How about having a variable in any of the worker global contexts and
> > use that? Of course, we can remove this restriction entirely in case we
> > fully allow parallelism for INSERT INTO SELECT, CTAS, and COPY.
>
> And for the purpose of your question, we could then have a
> table_insert_allow_parallel(TableInsertScan *);
> or an additional arg to table_begin_insert().
>

>
> Thanks Andres for the comments.
>
> On Thu, Sep 24, 2020 at 8:11 AM Andres Freund <[hidden email]> wrote:
> >
> > > The design:
> >
> > I think it'd be good if you could explain a bit more why you think this
> > safe to do in the way you have done it.
> >
> > E.g. from a quick scroll through the patch, there's not even a comment
> > explaining that the only reason there doesn't need to be code dealing
> > with xid assignment because we already did the catalog changes to create
> > the table.
> >
>
> Yes we do a bunch of catalog changes related to the created new table.
> We will have both the txn id and command id assigned when catalogue
> changes are being made. But, right after the table is created in the
> leader, the command id is incremented (CommandCounterIncrement() is
> called from create_ctas_internal()) whereas the txn id remains the
> same. The new command id is marked as GetCurrentCommandId(true); in
> intorel_startup, then the parallel mode is entered. The txn id and
> command id are serialized into parallel DSM, they are then available
> to all parallel workers. This is discussed in [1].
>
> Few changes I have to make in the parallel worker code: set
> currentCommandIdUsed = true;, may be via a common API
> SetCurrentCommandIdUsedForWorker() proposed in [1] and remove the
> extra command id sharing from the leader to workers.
>
> I will add a few comments in the upcoming patch related to the above info.
>

> Thoughts?
>
> >
> > > Below things are still pending. Thoughts are most welcome:
> > > 1. How better we can lift the "cannot insert tuples in a parallel worker"
> > > from heap_prepare_insert() for only CTAS cases or for that matter parallel
> > > copy? How about having a variable in any of the worker global contexts and
> > > use that? Of course, we can remove this restriction entirely in case we
> > > fully allow parallelism for INSERT INTO SELECT, CTAS, and COPY.
> >
> > And for the purpose of your question, we could then have a
> > table_insert_allow_parallel(TableInsertScan *);
> > or an additional arg to table_begin_insert().
> >
>
> Removing "cannot insert tuples in a parallel worker" restriction from
> heap_prepare_insert() is a common problem for parallel inserts in
> general, i.e. parallel inserts in CTAS, parallel INSERT INTO
> SELECTs[1] and parallel copy[2]. It will be good if a common solution
> is agreed.
>

>
> On Tue, Oct 6, 2020 at 10:58 AM Amit Kapila <[hidden email]> wrote:
> >
> >
> > While skimming through the patch, a small thing I noticed:
> > + /*
> > + * SELECT part of the CTAS is parallelizable, so we can make
> > + * each parallel worker insert the tuples that are resulted
> > + * in it's execution into the target table.
> > + */
> > + if (!is_matview &&
> > + IsA(plan->planTree, Gather))
> > + ((DR_intorel *) dest)->is_parallel = true;
> > +
> >
> > I am not sure at this stage if this is the best way to make CTAS as
> > parallel but if so, then probably you can expand the comments a bit to
> > say why you consider only Gather node (and that too when it is the
> > top-most node) and why not another parallel node like GatherMerge?
> >
>
> If somebody expects to preserve the order of the tuples that are
> coming from GatherMerge node of the select part in CTAS or SELECT INTO
> while inserting, now if parallelism is allowed, that may not be the
> case i.e. the order of insertion of tuples may vary. I'm not quite
> sure, if someone wants to use order by in the select parts of CTAS or
> SELECT INTO in a real world use case. Thoughts?
>

>
> On Wed, Oct 14, 2020 at 6:16 PM Amit Kapila <[hidden email]> wrote:
> >
> > > For prepared statements, the parallelism will not be picked and so is
> > > parallel insertion.
> >
> > Hmm, I am not sure what makes you say this statement. The parallelism
> > is enabled for prepared statements since commit 57a6a72b6b.
> >
>
> Thanks for letting me know this. I misunderstood the parallelism for prepared statements. Now, I verified with a proper use case(see below), where I had a prepared statement, CTAS having EXECUTE, in this case too parallelism is picked and parallel insertion happened with the patch proposed in this thread. Do we have any problems if we allow parallel insertion for these cases?
>
> PREPARE myselect AS SELECT * FROM t1;
> EXPLAIN ANALYZE CREATE TABLE t1_test AS EXECUTE myselect;
>
> I think the commit 57a6a72b6b has not added any test cases, isn't it good to add one in prepare.sql or select_parallel.sql?
>

> >
> > > 1. How to represent the parallel insert for CTAS in explain plans? The
> > > explain CTAS shows the plan for only the SELECT part. How about having
> > > some textual info along with the Gather node? I'm not quite sure on
> > > this point, any suggestions are welcome.
> >
> > I am also not sure about this point because we don't display anything
> > for the DDL part in explain. Can you propose by showing some example
> > of what you have in mind?
> >
>
> I thought we could have something like this.
>  -----------------------------------------------------------------------------
>      Gather  (cost=1000.00..108738.90 rows=0 width=8)
>      Workers Planned: 2 Parallel Insert on t_test1
>         ->  Parallel Seq Scan on t_test  (cost=0.00..106748.00 rows=4954 width=8)
>              Filter: (many < 10000)
>  -----------------------------------------------------------------------------
>

> On Tue, Oct 6, 2020 at 10:58 AM Amit Kapila <[hidden email]> wrote:
>>
>>> Yes we do a bunch of catalog changes related to the created new table.
>>> We will have both the txn id and command id assigned when catalogue
>>> changes are being made. But, right after the table is created in the
>>> leader, the command id is incremented (CommandCounterIncrement() is
>>> called from create_ctas_internal()) whereas the txn id remains the
>>> same. The new command id is marked as GetCurrentCommandId(true); in
>>> intorel_startup, then the parallel mode is entered. The txn id and
>>> command id are serialized into parallel DSM, they are then available
>>> to all parallel workers. This is discussed in [1].
>>>
>>> Few changes I have to make in the parallel worker code: set
>>> currentCommandIdUsed = true;, may be via a common API
>>> SetCurrentCommandIdUsedForWorker() proposed in [1] and remove the
>>> extra command id sharing from the leader to workers.
>>>
>>> I will add a few comments in the upcoming patch related to the above info.
>>>
>>
>> Yes, that would be good.
>>
>
> Added comments.
>
>>
>>>> But how does that work for SELECT INTO? Are you prohibiting
>>>> that? ...
>>>>
>>>
>>> In case of SELECT INTO, a new table gets created and I'm not
>>> prohibiting the parallel inserts and I think we don't need to.
>>>
>>
>> So, in this case, also do we ensure that table is created before we
>> launch the workers. If so, I think you can explain in comments about
>> it and what you need to do that to ensure the same.
>>
>
> For SELECT INTO, the table gets created by the leader in
> create_ctas_internal(), then ExecInitParallelPlan() gets called which
> launches the workers and then the leader(if asked to do so) and the
> workers insert the rows. So, we don't need to do any extra work to
> ensure the table gets created before the workers start inserting
> tuples.
>
>>
>> While skimming through the patch, a small thing I noticed:
>> + /*
>> + * SELECT part of the CTAS is parallelizable, so we can make
>> + * each parallel worker insert the tuples that are resulted
>> + * in it's execution into the target table.
>> + */
>> + if (!is_matview &&
>> + IsA(plan->planTree, Gather))
>> + ((DR_intorel *) dest)->is_parallel = true;
>> +
>>
>> I am not sure at this stage if this is the best way to make CTAS as
>> parallel but if so, then probably you can expand the comments a bit to
>> say why you consider only Gather node (and that too when it is the
>> top-most node) and why not another parallel node like GatherMerge?
>>
>
> If somebody expects to preserve the order of the tuples that are
> coming from GatherMerge node of the select part in CTAS or SELECT INTO
> while inserting, now if parallelism is allowed, that may not be the
> case i.e. the order of insertion of tuples may vary. I'm not quite
> sure, if someone wants to use order by in the select parts of CTAS or
> SELECT INTO in a real world use case. Thoughts?
>
>>
>> Right, for now, I think you can simply remove that check from the code
>> instead of just commenting it. We will see if there is a better
>> check/Assert we can add there.
>>
>
> Done.
>
> I also worked on some of the open points I listed earlier in my mail.
>
>>
>> 3. Need to restrict parallel inserts, if CTAS tries to create temp/global tables as the workers will not have access to those tables.
>>
>
> Done.
>
>>
>> Need to analyze whether to allow parallelism if CTAS has prepared statements or with no data.
>>
>
> For prepared statements, the parallelism will not be picked and so is
> parallel insertion.
> For CTAS with no data option case the select part is not even planned,
> and so the parallelism will also not be picked.
>
>>
>> 4. Need to stop unnecessary parallel shared state such as tuple queue being created and shared to workers.
>>
>
> Done.
>
> I'm listing the things that are still pending.
>
> 1. How to represent the parallel insert for CTAS in explain plans? The
> explain CTAS shows the plan for only the SELECT part. How about having
> some textual info along with the Gather node? I'm not quite sure on
> this point, any suggestions are welcome.
> 2. Addition of new test cases. Testing with more scenarios and
> different data sets, sizes, tablespaces, select into. Analysis on the
> 2 mismatches in write_parallel.sql regression test.
>
> Attaching v2 patch, thoughts and comments are welcome.
>
> With Regards,
> Bharath Rupireddy.
> EnterpriseDB: http://www.enterprisedb.com
>

>
> Really looking forward to this ending up in postgres as I think it's a
> very nice improvement.
>
> Whilst reviewing your patch I was wondering: is there a reason you did
> not introduce a batch insert in the destreceiver for the CTAS? For me
> this makes a huge difference in ingest speed as otherwise the inserts do
> not really scale so well as lock contention start to be a big problem.
> If you like I can make a patch to introduce this on top?
>

> On Fri, Oct 16, 2020 at 11:33 AM Luc Vlaming <[hidden email]> wrote:
>>
>> Really looking forward to this ending up in postgres as I think it's a
>> very nice improvement.
>>
>> Whilst reviewing your patch I was wondering: is there a reason you did
>> not introduce a batch insert in the destreceiver for the CTAS? For me
>> this makes a huge difference in ingest speed as otherwise the inserts do
>> not really scale so well as lock contention start to be a big problem.
>> If you like I can make a patch to introduce this on top?
>>
>
> Thanks for your interest. You are right, we can get maximum
> improvement if we have multi inserts in destreceiver for the CTAS on
> the similar lines to COPY FROM command. I specified this point in my
> first mail [1]. You may want to take a look at an already existing
> patch [2] for multi inserts, I think there are some review comments to
> be addressed in that patch. I would love to see the multi insert patch
> getting revived.
>
> [1] - https://www.postgresql.org/message-id/CALj2ACWFq6Z4_jd9RPByURB8-Y8wccQWzLf%2B0-Jg%2BKYT7ZO-Ug%40mail.gmail.com
> [2] - https://www.postgresql.org/message-id/CAEET0ZG31mD5SWjTYsAt0JTLReOejPvusJorZ3kGZ1%3DN1AC-Fw%40mail.gmail.com
>
> With Regards,
> Bharath Rupireddy.
> EnterpriseDB: http://www.enterprisedb.com
>

>
> Hi,
>
> I have an issue about the following code:
>
>         econtext = node->ps.ps_ExprContext;
>         ResetExprContext(econtext);
>
> +       if (ISCTAS(node->ps.intoclause))
> +       {
> +               ExecParallelInsertInCTAS(node);
> +               return NULL;
> +       }
>
>         /* If no projection is required, we're done. */
>         if (node->ps.ps_ProjInfo == NULL)
>                 return slot;
>
>         /*
>          * Form the result tuple using ExecProject(), and return it.
>          */
>         econtext->ecxt_outertuple = slot;
>         return ExecProject(node->ps.ps_ProjInfo);
>
> It seems the projection will be skipped.
> Is this because projection is not required in this case ?
> (I'm not very familiar with where the projection will be.)
>

> On Thu, Nov 26, 2020 at 7:47 AM Hou, Zhijie <[hidden email]>
> wrote:
> >
> > Hi,
> >
> > I have an issue about the following code:
> >
> >         econtext = node->ps.ps_ExprContext;
> >         ResetExprContext(econtext);
> >
> > +       if (ISCTAS(node->ps.intoclause))
> > +       {
> > +               ExecParallelInsertInCTAS(node);
> > +               return NULL;
> > +       }
> >
> >         /* If no projection is required, we're done. */
> >         if (node->ps.ps_ProjInfo == NULL)
> >                 return slot;
> >
> >         /*
> >          * Form the result tuple using ExecProject(), and return it.
> >          */
> >         econtext->ecxt_outertuple = slot;
> >         return ExecProject(node->ps.ps_ProjInfo);
> >
> > It seems the projection will be skipped.
> > Is this because projection is not required in this case ?
> > (I'm not very familiar with where the projection will be.)
> >
>
> For parallel inserts in CTAS, I don't think we need to project the tuples
> being returned from the underlying plan nodes, and also we have nothing
> to project from the Gather node further up. The required projection will
> happen while the tuples are being returned from the underlying nodes and
> the projected tuples are being directly fed to CTAS's dest receiver
> intorel_receive(), from there into the created table. We don't need
> ExecProject again in ExecParallelInsertInCTAS().
>
> For instance, projection will always be done when the tuple is being returned
> from an underlying sequential scan node(see ExecScan() -->
> ExecProject() and this is true for both leader and workers. In both leader
> and workers, we are just calling CTAS's dest receiver intorel_receive().
>
> Thoughts?

>
> I took a deep look at the projection logic.
> In most cases, you are right that Gather node does not need projection.
>
> In some rare cases, such as Subplan (or initplan I guess).
> The projection will happen in Gather node.
>
> The example:
>
> Create table test(i int);
> Create table test2(a int, b int);
> insert into test values(generate_series(1,10000000,1));
> insert into test2 values(generate_series(1,1000,1), generate_series(1,1000,1));
>
> postgres=# explain(verbose, costs off) select test.i,(select i from (select * from test2) as tt limit 1) from test where test.i < 2000;
>                QUERY PLAN
> ----------------------------------------
>  Gather
>    Output: test.i, (SubPlan 1)
>    Workers Planned: 2
>    ->  Parallel Seq Scan on public.test
>          Output: test.i
>          Filter: (test.i < 2000)
>    SubPlan 1
>      ->  Limit
>            Output: (test.i)
>            ->  Seq Scan on public.test2
>                  Output: test.i
>
> In this case, projection is necessary,
> because the subplan will be executed in projection.
>
> If skipped, the table created will loss some data.
>