Thread: Parallel Seq Scan

On 12/04/2014 07:35 AM, Amit Kapila wrote:
> [snip]
>
> The number of worker backends that can be used for
> parallel seq scan can be configured by using a new GUC
> parallel_seqscan_degree, the default value of which is zero
> and it means parallel seq scan will not be considered unless
> user configures this value.

The number of parallel workers should be capped (of course!) at the 
maximum amount of "processors" (cores/vCores, threads/hyperthreads) 
available.

More over, when load goes up, the relative cost of parallel working 
should go up as well.
Something like:    p = number of cores    l = 1min-load
    additional_cost = tuple estimate * cpu_tuple_cost * (l+1)/(c-1)

(for c>1, of course)


> In ExecutorStart phase, initiate the required number of workers
> as per parallel seq scan plan and setup dynamic shared memory and
> share the information required for worker to execute the scan.
> Currently I have just shared the relId, targetlist and number
> of blocks to be scanned by worker, however I think we might want
> to generate a plan for each of the workers in master backend and
> then share the same to individual worker.
[snip]
> Attached patch is just to facilitate the discussion about the
> parallel seq scan and may be some other dependent tasks like
> sharing of various states like combocid, snapshot with parallel
> workers.  It is by no means ready to do any complex test, ofcourse
> I will work towards making it more robust both in terms of adding
> more stuff and doing performance optimizations.
>
> Thoughts/Suggestions?

Not directly (I haven't had the time to read the code yet), but I'm 
thinking about the ability to simply *replace* executor methods from an 
extension.
This could be an alternative to providing additional nodes that the 
planner can include in the final plan tree, ready to be executed.

The parallel seq scan nodes are definitively the best approach for 
"parallel query", since the planner can optimize them based on cost.
I'm wondering about the ability to modify the implementation of some 
methods themselves once at execution time: given a previously planned 
query, chances are that, at execution time (I'm specifically thinking 
about prepared statements here), a different implementation of the same 
"node" might be more suitable and could be used instead while the 
condition holds.

If this latter line of thinking is too off-topic within this thread and 
there is any interest, we can move the comments to another thread and 
I'd begin work on a PoC patch. It might as well make sense to implement 
the executor overloading mechanism alongide the custom plan API, though.
Any comments appreciated.


Thank you for your work, Amit


Regards,
    / J.L.

José,

* José Luis Tallón (jltallon@adv-solutions.net) wrote:
> On 12/04/2014 07:35 AM, Amit Kapila wrote:
> >The number of worker backends that can be used for
> >parallel seq scan can be configured by using a new GUC
> >parallel_seqscan_degree, the default value of which is zero
> >and it means parallel seq scan will not be considered unless
> >user configures this value.
>
> The number of parallel workers should be capped (of course!) at the
> maximum amount of "processors" (cores/vCores, threads/hyperthreads)
> available.
>
> More over, when load goes up, the relative cost of parallel working
> should go up as well.
> Something like:
>     p = number of cores
>     l = 1min-load
>
>     additional_cost = tuple estimate * cpu_tuple_cost * (l+1)/(c-1)
>
> (for c>1, of course)

While I agree in general that we'll need to come up with appropriate
acceptance criteria, etc, I don't think we want to complicate this patch
with that initially.  A SUSET GUC which caps the parallel GUC would be
enough for an initial implementation, imv.

> Not directly (I haven't had the time to read the code yet), but I'm
> thinking about the ability to simply *replace* executor methods from
> an extension.

You probably want to look at the CustomScan thread+patch directly then..
Thanks,
    Stephen

Amit,

* Amit Kapila (amit.kapila16@gmail.com) wrote:
> postgres=# explain select c1 from t1;
>                       QUERY PLAN
> ------------------------------------------------------
>  Seq Scan on t1  (cost=0.00..101.00 rows=100 width=4)
> (1 row)
>
>
> postgres=# set parallel_seqscan_degree=4;
> SET
> postgres=# explain select c1 from t1;
>                           QUERY PLAN
> --------------------------------------------------------------
>  Parallel Seq Scan on t1  (cost=0.00..25.25 rows=100 width=4)
>    Number of Workers: 4
>    Number of Blocks Per Workers: 25
> (3 rows)

This is all great and interesting, but I feel like folks might be
waiting to see just what kind of performance results come from this (and
what kind of hardware is needed to see gains..).  There's likely to be
situations where this change is an improvement while also being cases
where it makes things worse.

One really interesting case would be parallel seq scans which are
executing against foreign tables/FDWs..
Thanks!
    Stephen

On 12/5/14, 9:08 AM, José Luis Tallón wrote:
>
> More over, when load goes up, the relative cost of parallel working should go up as well.
> Something like:
>      p = number of cores
>      l = 1min-load
>
>      additional_cost = tuple estimate * cpu_tuple_cost * (l+1)/(c-1)
>
> (for c>1, of course)

...

> The parallel seq scan nodes are definitively the best approach for "parallel query", since the planner can optimize
thembased on cost.
 
> I'm wondering about the ability to modify the implementation of some methods themselves once at execution time: given
apreviously planned query, chances are that, at execution time (I'm specifically thinking about prepared statements
here),a different implementation of the same "node" might be more suitable and could be used instead while the
conditionholds.
 

These comments got me wondering... would it be better to decide on parallelism during execution instead of at plan
time?That would allow us to dynamically scale parallelism based on system load. If we don't even consider parallelism
untilwe've pulled some number of tuples/pages from a relation, this would also eliminate all parallel overhead on small
relations.
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

* Amit Kapila (amit.kapila16@gmail.com) wrote:
> 1. As the patch currently stands, it just shares the relevant
> data (like relid, target list, block range each worker should
> perform on etc.) to the worker and then worker receives that
> data and form the planned statement which it will execute and
> send the results back to master backend.  So the question
> here is do you think it is reasonable or should we try to form
> the complete plan for each worker and then share the same
> and may be other information as well like range table entries
> which are required.   My personal gut feeling in this matter
> is that for long term it might be better to form the complete
> plan of each worker in master and share the same, however
> I think the current way as done in patch (okay that needs
> some improvement) is also not bad and quite easier to implement.

For my 2c, I'd like to see it support exactly what the SeqScan node
supports and then also what Foreign Scan supports.  That would mean we'd
then be able to push filtering down to the workers which would be great.
Even better would be figuring out how to parallelize an Append node
(perhaps only possible when the nodes underneath are all SeqScan or
ForeignScan nodes) since that would allow us to then parallelize the
work across multiple tables and remote servers.

One of the big reasons why I was asking about performance data is that,
today, we can't easily split a single relation across multiple i/o
channels.  Sure, we can use RAID and get the i/o channel that the table
sits on faster than a single disk and possibly fast enough that a single
CPU can't keep up, but that's not quite the same.  The historical
recommendations for Hadoop nodes is around one CPU per drive (of course,
it'll depend on workload, etc, etc, but still) and while there's still a
lot of testing, etc, to be done before we can be sure about the 'right'
answer for PG (and it'll also vary based on workload, etc), that strikes
me as a pretty reasonable rule-of-thumb to go on.

Of course, I'm aware that this won't be as easy to implement..

> 2. Next question related to above is what should be the
> output of ExplainPlan, as currently worker is responsible
> for forming its own plan, Explain Plan is not able to show
> the detailed plan for each worker, is that okay?

I'm not entirely following this.  How can the worker be responsible for
its own "plan" when the information passed to it (per the above
paragraph..) is pretty minimal?  In general, I don't think we need to
have specifics like "this worker is going to do exactly X" because we
will eventually need some communication to happen between the worker and
the master process where the worker can ask for more work because it's
finished what it was tasked with and the master will need to give it
another chunk of work to do.  I don't think we want exactly what each
worker process will do to be fully formed at the outset because, even
with the best information available, given concurrent load on the
system, it's not going to be perfect and we'll end up starving workers.
The plan, as formed by the master, should be more along the lines of
"this is what I'm gonna have my workers do" along w/ how many workers,
etc, and then it goes and does it.  Perhaps for an 'explain analyze' we
return information about what workers actually *did* what, but that's a
whole different discussion.

> 3. Some places where optimizations are possible:
> - Currently after getting the tuple from heap, it is deformed by
> worker and sent via message queue to master backend, master
> backend then forms the tuple and send it to upper layer which
> before sending it to frontend again deforms it via slot_getallattrs(slot).

If this is done as I was proposing above, we might be able to avoid
this, but I don't know that it's a huge issue either way..  The bigger
issue is getting the filtering pushed down.

> - Master backend currently receives the data from multiple workers
> serially.  We can optimize in a way that it can check other queues,
> if there is no data in current queue.

Yes, this is pretty critical.  In fact, it's one of the recommendations
I made previously about how to change the Append node to parallelize
Foreign Scan node work.

> - Master backend is just responsible for coordination among workers
> It shares the required information to workers and then fetch the
> data processed by each worker, by using some more logic, we might
> be able to make master backend also fetch data from heap rather than
> doing just co-ordination among workers.

I don't think this is really necessary...

> I think in all above places we can do some optimisation, however
> we can do that later as well, unless they hit the performance badly for
> cases which people care most.

I agree that we can improve the performance through various
optimizations later, but it's important to get the general structure and
design right or we'll end up having to reimplement a lot of it.

> 4. Should parallel_seqscan_degree value be dependent on other
> backend processes like MaxConnections, max_worker_processes,
> autovacuum_max_workers do or should it be independent like
> max_wal_senders?

Well, we're not going to be able to spin off more workers than we have
process slots, but I'm not sure we need anything more than that?  In any
case, this is definitely an area we can work on improving later and I
don't think it really impacts the rest of the design.
Thanks,
    Stephen

On Sat, Dec 6, 2014 at 12:13 AM, David Rowley <dgrowleyml@gmail.com> wrote:
> It's bare-bones core support for allowing aggregate states to be merged
> together with another aggregate state. I would imagine that if a query such
> as:
>
> SELECT MAX(value) FROM bigtable;
>
> was run, then a series of parallel workers could go off and each find the
> max value from their portion of the table and then perhaps some other node
> type would then take all the intermediate results from the workers, once
> they're finished, and join all of the aggregate states into one and return
> that. Naturally, you'd need to check that all aggregates used in the
> targetlist had a merge function first.

I think this is great infrastructure and could also be useful for
pushing down aggregates in cases involving foreign data wrappers.  But
I suggest we discuss it on a separate thread because it's not related
to parallel seq scan per se.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Sat, Dec 6, 2014 at 1:50 AM, Amit Kapila <amit.kapila16@gmail.com> wrote:
> I think we have access to this information in planner (RelOptInfo -> pages),
> if we want, we can use that to eliminate the small relations from
> parallelism, but question is how big relations do we want to consider
> for parallelism, one way is to check via tests which I am planning to
> follow, do you think we have any heuristic which we can use to decide
> how big relations should be consider for parallelism?

Surely the Path machinery needs to decide this in particular cases
based on cost.  We should assign some cost to starting a parallel
worker via some new GUC, like parallel_startup_cost = 100,000.  And
then we should also assign a cost to the act of relaying a tuple from
the parallel worker to the master, maybe cpu_tuple_cost (or some new
GUC).  For a small relation, or a query with a LIMIT clause, the
parallel startup cost will make starting a lot of workers look
unattractive, but for bigger relations it will make sense from a cost
perspective, which is exactly what we want.

There are probably other important considerations based on goals for
overall resource utilization, and also because at a certain point
adding more workers won't help because the disk will be saturated.  I
don't know exactly what we should do about those issues yet, but the
steps described in the previous paragraph seem like a good place to
start anyway.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Sat, Dec 6, 2014 at 7:07 AM, Stephen Frost <sfrost@snowman.net> wrote:
> For my 2c, I'd like to see it support exactly what the SeqScan node
> supports and then also what Foreign Scan supports.  That would mean we'd
> then be able to push filtering down to the workers which would be great.
> Even better would be figuring out how to parallelize an Append node
> (perhaps only possible when the nodes underneath are all SeqScan or
> ForeignScan nodes) since that would allow us to then parallelize the
> work across multiple tables and remote servers.

I don't see how we can support the stuff ForeignScan does; presumably
any parallelism there is up to the FDW to implement, using whatever
in-core tools we provide.  I do agree that parallelizing Append nodes
is useful; but let's get one thing done first before we start trying
to do thing #2.

> I'm not entirely following this.  How can the worker be responsible for
> its own "plan" when the information passed to it (per the above
> paragraph..) is pretty minimal?  In general, I don't think we need to
> have specifics like "this worker is going to do exactly X" because we
> will eventually need some communication to happen between the worker and
> the master process where the worker can ask for more work because it's
> finished what it was tasked with and the master will need to give it
> another chunk of work to do.  I don't think we want exactly what each
> worker process will do to be fully formed at the outset because, even
> with the best information available, given concurrent load on the
> system, it's not going to be perfect and we'll end up starving workers.
> The plan, as formed by the master, should be more along the lines of
> "this is what I'm gonna have my workers do" along w/ how many workers,
> etc, and then it goes and does it.  Perhaps for an 'explain analyze' we
> return information about what workers actually *did* what, but that's a
> whole different discussion.

I agree with this.  For a first version, I think it's OK to start a
worker up for a particular sequential scan and have it help with that
sequential scan until the scan is completed, and then exit.  It should
not, as the present version of the patch does, assign a fixed block
range to each worker; instead, workers should allocate a block or
chunk of blocks to work on until no blocks remain.  That way, even if
every worker but one gets stuck, the rest of the scan can still
finish.

Eventually, we will want to be smarter about sharing works between
multiple parts of the plan, but I think it is just fine to leave that
as a future enhancement for now.

>> - Master backend is just responsible for coordination among workers
>> It shares the required information to workers and then fetch the
>> data processed by each worker, by using some more logic, we might
>> be able to make master backend also fetch data from heap rather than
>> doing just co-ordination among workers.
>
> I don't think this is really necessary...

I think it would be an awfully good idea to make this work.  The
master thread may be significantly faster than any of the others
because it has no IPC costs.  We don't want to leave our best resource
sitting on the bench.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Tue, Dec 9, 2014 at 12:46 AM, Amit Kapila <amit.kapila16@gmail.com> wrote:
>> I agree with this.  For a first version, I think it's OK to start a
>> worker up for a particular sequential scan and have it help with that
>> sequential scan until the scan is completed, and then exit.  It should
>> not, as the present version of the patch does, assign a fixed block
>> range to each worker; instead, workers should allocate a block or
>> chunk of blocks to work on until no blocks remain.  That way, even if
>> every worker but one gets stuck, the rest of the scan can still
>> finish.
>>
> I will check on this point and see if it is feasible to do something on
> those lines, basically currently at Executor initialization phase, we
> set the scan limits and then during Executor Run phase use
> heap_getnext to fetch the tuples accordingly, but doing it dynamically
> means at ExecutorRun phase we need to reset the scan limit for
> which page/pages to scan, still I have to check if there is any problem
> with such an idea.  Do you any different idea in mind?

Hmm.  Well, it looks like there are basically two choices: you can
either (as you propose) deal with this above the level of the
heap_beginscan/heap_getnext API by scanning one or a few pages at a
time and then resetting the scan to a new starting page via
heap_setscanlimits; or alternatively, you can add a callback to
HeapScanDescData that, if non-NULL, will be invoked to get the next
block number to scan.  I'm not entirely sure which is better.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

Amit,

* Amit Kapila (amit.kapila16@gmail.com) wrote:
> 1. Parallel workers help a lot when there is an expensive qualification
> to evaluated, the more expensive the qualification the more better are
> results.

I'd certainly hope so. ;)

> 2. It works well for low selectivity quals and as the selectivity increases,
> the benefit tends to go down due to additional tuple communication cost
> between workers and master backend.

I'm a bit sad to hear that the communication between workers and the
master backend is already being a bottleneck.  Now, that said, the box
you're playing with looks to be pretty beefy and therefore the i/o
subsystem might be particularly good, but generally speaking, it's a lot
faster to move data in memory than it is to pull it off disk, and so I
wouldn't expect the tuple communication between processes to really be
the bottleneck...

> 3. After certain point, increasing having more number of workers won't
> help and rather have negative impact, refer Test-4.

Yes, I see that too and it's also interesting- have you been able to
identify why?  What is the overhead (specifically) which is causing
that?

> I think as discussed previously we need to introduce 2 additional cost
> variables (parallel_startup_cost, cpu_tuple_communication_cost) to
> estimate the parallel seq scan cost so that when the tables are small
> or selectivity is high, it should increase the cost of parallel plan.

I agree that we need to figure out a way to cost out parallel plans, but
I have doubts about these being the right way to do that.  There has
been quite a bit of literature regarding parallel execution and
planning- have you had a chance to review anything along those lines?
We certainly like to draw on previous experiences and analysis rather
than trying to pave our own way.

With these additional costs comes the consideration that we're looking
for a wall-clock runtime proxy and therefore, while we need to add costs
for parallel startup and tuple communication, we have to reduce the
overall cost because of the parallelism or we'd never end up choosing a
parallel plan.  Is the thought to simply add up all the costs and then
divide?  Or perhaps to divide the cost of the actual plan but then add
in the parallel startup cost and the tuple communication cost?

Perhaps there has been prior discussion on these points but I'm thinking
we need a README or similar which discusses all of this and includes any
references out to academic papers or similar as appropriate.
Thanks!
    Stephen

On Fri, Dec 19, 2014 at 7:51 AM, Stephen Frost <sfrost@snowman.net> wrote:
>> 3. After certain point, increasing having more number of workers won't
>> help and rather have negative impact, refer Test-4.
>
> Yes, I see that too and it's also interesting- have you been able to
> identify why?  What is the overhead (specifically) which is causing
> that?

Let's rewind.  Amit's results show that, with a naive algorithm
(pre-distributing equal-sized chunks of the relation to every worker)
and a fairly-naive first cut at how to pass tuples around (I believe
largely from what I did in pg_background) he can sequential-scan a
table with 8 workers at 6.4 times the speed of a single process, and
you're complaining because it's not efficient enough?  It's a first
draft!  Be happy we got 6.4x, for crying out loud!

The barrier to getting parallel sequential scan (or any parallel
feature at all) committed is not going to be whether an 8-way scan is
6.4 times faster or 7.1 times faster or 7.8 times faster.  It's going
to be whether it's robust and won't break things.  We should be
focusing most of our effort here on identifying and fixing robustness
problems.  I'd vote to commit a feature like this with a 3x
performance speedup if I thought it was robust enough.

I'm not saying we shouldn't try to improve the performance here - we
definitely should.  But I don't think we should say, oh, an 8-way scan
isn't good enough, we need a 16-way or 32-way scan in order for this
to be efficient.  That is getting your priorities quite mixed up.

>> I think as discussed previously we need to introduce 2 additional cost
>> variables (parallel_startup_cost, cpu_tuple_communication_cost) to
>> estimate the parallel seq scan cost so that when the tables are small
>> or selectivity is high, it should increase the cost of parallel plan.
>
> I agree that we need to figure out a way to cost out parallel plans, but
> I have doubts about these being the right way to do that.  There has
> been quite a bit of literature regarding parallel execution and
> planning- have you had a chance to review anything along those lines?
> We certainly like to draw on previous experiences and analysis rather
> than trying to pave our own way.

I agree that it would be good to review the literature, but am not
aware of anything relevant.  Could you (or can anyone) provide some
links?

> With these additional costs comes the consideration that we're looking
> for a wall-clock runtime proxy and therefore, while we need to add costs
> for parallel startup and tuple communication, we have to reduce the
> overall cost because of the parallelism or we'd never end up choosing a
> parallel plan.  Is the thought to simply add up all the costs and then
> divide?  Or perhaps to divide the cost of the actual plan but then add
> in the parallel startup cost and the tuple communication cost?

This has been discussed, on this thread.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

Robert,

* Robert Haas (robertmhaas@gmail.com) wrote:
> On Fri, Dec 19, 2014 at 7:51 AM, Stephen Frost <sfrost@snowman.net> wrote:
> >> 3. After certain point, increasing having more number of workers won't
> >> help and rather have negative impact, refer Test-4.
> >
> > Yes, I see that too and it's also interesting- have you been able to
> > identify why?  What is the overhead (specifically) which is causing
> > that?
>
> Let's rewind.  Amit's results show that, with a naive algorithm
> (pre-distributing equal-sized chunks of the relation to every worker)
> and a fairly-naive first cut at how to pass tuples around (I believe
> largely from what I did in pg_background) he can sequential-scan a
> table with 8 workers at 6.4 times the speed of a single process, and
> you're complaining because it's not efficient enough?  It's a first
> draft!  Be happy we got 6.4x, for crying out loud!

He also showed cases where parallelizing a query even with just two
workers caused a serious increase in the total runtime (Test 6).  Even
having four workers was slower in that case, but a modest performance
improvment was reached at eight but then no improvement from that was
seen when running with 16.

Being able to understand what's happening will inform how we cost this
to, hopefully, achieve the 6.4x gains where we can and avoid the
pitfalls of performing worse than a single thread in cases where
parallelism doesn't help.  What would likely be very helpful in the
analysis would be CPU time information- when running with eight workers,
were we using 800% CPU (8x 100%), or something less (perhaps due to
locking, i/o, or other processes).

Perhaps it's my fault for not being surprised that a naive first cut
gives us such gains as my experience with parallel operations and PG has
generally been very good (through the use of multiple connections to the
DB and therefore independent transactions, of course).  I'm very excited
that we're making such great progress towards having parallel execution
in the DB as I've often used PG in data warehouse use-cases.

> The barrier to getting parallel sequential scan (or any parallel
> feature at all) committed is not going to be whether an 8-way scan is
> 6.4 times faster or 7.1 times faster or 7.8 times faster.  It's going
> to be whether it's robust and won't break things.  We should be
> focusing most of our effort here on identifying and fixing robustness
> problems.  I'd vote to commit a feature like this with a 3x
> performance speedup if I thought it was robust enough.

I don't have any problem if an 8-way scan is 6.4x faster or if it's 7.1
times faster, but what if that 3x performance speedup is only achieved
when running with 8 CPUs at 100%?  We'd have to coach our users to
constantly be tweaking the enable_parallel_query (or whatever) option
for the queries where it helps and turning it off for others.  I'm not
so excited about that.

> I'm not saying we shouldn't try to improve the performance here - we
> definitely should.  But I don't think we should say, oh, an 8-way scan
> isn't good enough, we need a 16-way or 32-way scan in order for this
> to be efficient.  That is getting your priorities quite mixed up.

I don't think I said that.  What I was getting at is that we need a cost
system which accounts for the costs accurately enough that we don't end
up with worse performance than single-threaded operation.  In general, I
don't expect that to be very difficult and we can be conservative in the
initial releases to hopefully avoid regressions, but it absolutely needs
consideration.

> >> I think as discussed previously we need to introduce 2 additional cost
> >> variables (parallel_startup_cost, cpu_tuple_communication_cost) to
> >> estimate the parallel seq scan cost so that when the tables are small
> >> or selectivity is high, it should increase the cost of parallel plan.
> >
> > I agree that we need to figure out a way to cost out parallel plans, but
> > I have doubts about these being the right way to do that.  There has
> > been quite a bit of literature regarding parallel execution and
> > planning- have you had a chance to review anything along those lines?
> > We certainly like to draw on previous experiences and analysis rather
> > than trying to pave our own way.
>
> I agree that it would be good to review the literature, but am not
> aware of anything relevant.  Could you (or can anyone) provide some
> links?

There's certainly documentation available from the other RDBMS' which
already support parallel query, as one source.  Other academic papers
exist (and once you've linked into one, the references and prior work
helps bring in others).  Sadly, I don't currently have ACM access (might
have to change that..), but there are publicly available papers also,
such as:

http://i.stanford.edu/pub/cstr/reports/cs/tr/96/1570/CS-TR-96-1570.pdf
http://www.vldb.org/conf/1998/p251.pdf
http://www.cs.uiuc.edu/class/fa05/cs591han/sigmodpods04/sigmod/pdf/I-001c.pdf

> > With these additional costs comes the consideration that we're looking
> > for a wall-clock runtime proxy and therefore, while we need to add costs
> > for parallel startup and tuple communication, we have to reduce the
> > overall cost because of the parallelism or we'd never end up choosing a
> > parallel plan.  Is the thought to simply add up all the costs and then
> > divide?  Or perhaps to divide the cost of the actual plan but then add
> > in the parallel startup cost and the tuple communication cost?
>
> This has been discussed, on this thread.

Fantastic.  What I found in the patch was:

+   /*
+    * We simply assume that cost will be equally shared by parallel
+    * workers which might not be true especially for doing disk access.
+    * XXX - We would like to change these values based on some concrete
+    * tests.
+    */

What I asked for was:

----
I'm thinking we need a README or similar which discusses all of this and
includes any references out to academic papers or similar as appropriate.
----

Perhaps it doesn't deserve its own README, but we clearly need more.
Thanks!
    Stephen

On 12/19/14 3:27 PM, Stephen Frost wrote:
> We'd have to coach our users to
> constantly be tweaking the enable_parallel_query (or whatever) option
> for the queries where it helps and turning it off for others.  I'm not
> so excited about that.

I'd be perfectly (that means 100%) happy if it just defaulted to off, 
but I could turn it up to 11 whenever I needed it.  I don't believe to 
be the only one with this opinion, either.


.marko

* Marko Tiikkaja (marko@joh.to) wrote:
> On 12/19/14 3:27 PM, Stephen Frost wrote:
> >We'd have to coach our users to
> >constantly be tweaking the enable_parallel_query (or whatever) option
> >for the queries where it helps and turning it off for others.  I'm not
> >so excited about that.
>
> I'd be perfectly (that means 100%) happy if it just defaulted to
> off, but I could turn it up to 11 whenever I needed it.  I don't
> believe to be the only one with this opinion, either.

Perhaps we should reconsider our general position on hints then and
add them so users can define the plan to be used..  For my part, I don't
see this as all that much different.

Consider if we were just adding HashJoin support today as an example.
Would we be happy if we had to default to enable_hashjoin = off?  Or if
users had to do that regularly because our costing was horrid?  It's bad
enough that we have to resort to those tweaks today in rare cases.
Thanks,
    Stephen

On Fri, Dec 19, 2014 at 9:39 AM, Stephen Frost <sfrost@snowman.net> wrote:
> Perhaps we should reconsider our general position on hints then and
> add them so users can define the plan to be used..  For my part, I don't
> see this as all that much different.
>
> Consider if we were just adding HashJoin support today as an example.
> Would we be happy if we had to default to enable_hashjoin = off?  Or if
> users had to do that regularly because our costing was horrid?  It's bad
> enough that we have to resort to those tweaks today in rare cases.

If you're proposing that it is not reasonable to have a GUC that
limits the degree of parallelism, then I think that's outright crazy:
that is probably the very first GUC we need to add.  New query
processing capabilities can entail new controlling GUCs, and
parallelism, being as complex at it is, will probably add several of
them.

But the big picture here is that if you want to ever have parallelism
in PostgreSQL at all, you're going to have to live with the first
version being pretty crude.  I think it's quite likely that the first
version of parallel sequential scan will be just as buggy as Hot
Standby was when we first added it, or as buggy as the multi-xact code
was when it went in, and probably subject to an even greater variety
of taxing limitations than any feature we've committed in the 6 years
I've been involved in the project.  We get to pick between that and
not having it at all.

I'll take a look at the papers you sent about parallel query
optimization, but personally I think that's putting the cart not only
before the horse but also before the road.  For V1, we need a query
optimization model that does not completely suck - no more.  The key
criterion here is that this has to WORK.  There will be time enough to
improve everything else once we reach that goal.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On 12/19/2014 04:39 PM, Stephen Frost wrote:
> * Marko Tiikkaja (marko@joh.to) wrote:
>> On 12/19/14 3:27 PM, Stephen Frost wrote:
>>> We'd have to coach our users to
>>> constantly be tweaking the enable_parallel_query (or whatever) option
>>> for the queries where it helps and turning it off for others.  I'm not
>>> so excited about that.
>>
>> I'd be perfectly (that means 100%) happy if it just defaulted to
>> off, but I could turn it up to 11 whenever I needed it.  I don't
>> believe to be the only one with this opinion, either.
>
> Perhaps we should reconsider our general position on hints then and
> add them so users can define the plan to be used..  For my part, I don't
> see this as all that much different.
>
> Consider if we were just adding HashJoin support today as an example.
> Would we be happy if we had to default to enable_hashjoin = off?  Or if
> users had to do that regularly because our costing was horrid?  It's bad
> enough that we have to resort to those tweaks today in rare cases.

This is somewhat different. Imagine that we achieve perfect 
parallelization, so that when you set enable_parallel_query=8, every 
query runs exactly 8x faster on an 8-core system, by using all eight cores.

Now, you might still want to turn parallelization off, or at least set 
it to a lower setting, on an OLTP system. You might not want a single 
query to hog all CPUs to run one query faster; you'd want to leave some 
for other queries. In particular, if you run a mix of short 
transactions, and some background-like tasks that run for minutes or 
hours, you do not want to starve the short transactions by giving all 
eight CPUs to the background task.

Admittedly, this is a rather crude knob to tune for such things,
but it's quite intuitive to a DBA: how many CPU cores is one query 
allowed to utilize? And we don't really have anything better.

In real life, there's always some overhead to parallelization, so that 
even if you can make one query run faster by doing it, you might hurt 
overall throughput. To some extent, it's a latency vs. throughput 
tradeoff, and it's quite reasonable to have a GUC for that because 
people have different priorities.

- Heikki

On 20/12/14 03:54, Heikki Linnakangas wrote:
> On 12/19/2014 04:39 PM, Stephen Frost wrote:
>> * Marko Tiikkaja (marko@joh.to) wrote:
>>> On 12/19/14 3:27 PM, Stephen Frost wrote:
>>>> We'd have to coach our users to
>>>> constantly be tweaking the enable_parallel_query (or whatever) option
>>>> for the queries where it helps and turning it off for others.  I'm not
>>>> so excited about that.
>>>
>>> I'd be perfectly (that means 100%) happy if it just defaulted to
>>> off, but I could turn it up to 11 whenever I needed it.  I don't
>>> believe to be the only one with this opinion, either.
>>
>> Perhaps we should reconsider our general position on hints then and
>> add them so users can define the plan to be used..  For my part, I don't
>> see this as all that much different.
>>
>> Consider if we were just adding HashJoin support today as an example.
>> Would we be happy if we had to default to enable_hashjoin = off?  Or if
>> users had to do that regularly because our costing was horrid? It's bad
>> enough that we have to resort to those tweaks today in rare cases.
>
> This is somewhat different. Imagine that we achieve perfect 
> parallelization, so that when you set enable_parallel_query=8, every 
> query runs exactly 8x faster on an 8-core system, by using all eight 
> cores.
>
> Now, you might still want to turn parallelization off, or at least set 
> it to a lower setting, on an OLTP system. You might not want a single 
> query to hog all CPUs to run one query faster; you'd want to leave 
> some for other queries. In particular, if you run a mix of short 
> transactions, and some background-like tasks that run for minutes or 
> hours, you do not want to starve the short transactions by giving all 
> eight CPUs to the background task.
>
> Admittedly, this is a rather crude knob to tune for such things,
> but it's quite intuitive to a DBA: how many CPU cores is one query 
> allowed to utilize? And we don't really have anything better.
>
> In real life, there's always some overhead to parallelization, so that 
> even if you can make one query run faster by doing it, you might hurt 
> overall throughput. To some extent, it's a latency vs. throughput 
> tradeoff, and it's quite reasonable to have a GUC for that because 
> people have different priorities.
>
> - Heikki
>
>
>
How about 3 numbers:
   minCPUs # > 0   maxCPUs           # >= minCPUs   fractionOfCPUs    # rounded up


If you just have the /*number*/ of CPUs then a setting that is 
appropriate for quad core, may be too /*small*/ for an octo core processor.

If you just have the /*fraction*/ of CPUs then a setting that is 
appropriate for quad core, may be too /*large*/ for an octo core processor.



Cheers,
Gavin

Robert,

* Robert Haas (robertmhaas@gmail.com) wrote:
> On Fri, Dec 19, 2014 at 9:39 AM, Stephen Frost <sfrost@snowman.net> wrote:
> > Perhaps we should reconsider our general position on hints then and
> > add them so users can define the plan to be used..  For my part, I don't
> > see this as all that much different.
> >
> > Consider if we were just adding HashJoin support today as an example.
> > Would we be happy if we had to default to enable_hashjoin = off?  Or if
> > users had to do that regularly because our costing was horrid?  It's bad
> > enough that we have to resort to those tweaks today in rare cases.
>
> If you're proposing that it is not reasonable to have a GUC that
> limits the degree of parallelism, then I think that's outright crazy:

I'm pretty sure that I didn't say anything along those lines.  I'll try
to be clearer.

What I'd like is such a GUC that we can set at a reasonable default of,
say, 4, and trust that our planner will generally do the right thing.
Clearly, this may be something which admins have to tweak but what I
would really like to avoid is users having to set this GUC explicitly
for each of their queries.

> that is probably the very first GUC we need to add.  New query
> processing capabilities can entail new controlling GUCs, and
> parallelism, being as complex at it is, will probably add several of
> them.

That's fine if they're intended for debugging issues or dealing with
unexpected bugs or issues, but let's not go into this thinking we should
add GUCs which are geared with the expectation of users tweaking them
regularly.

> But the big picture here is that if you want to ever have parallelism
> in PostgreSQL at all, you're going to have to live with the first
> version being pretty crude.  I think it's quite likely that the first
> version of parallel sequential scan will be just as buggy as Hot
> Standby was when we first added it, or as buggy as the multi-xact code
> was when it went in, and probably subject to an even greater variety
> of taxing limitations than any feature we've committed in the 6 years
> I've been involved in the project.  We get to pick between that and
> not having it at all.

If it's disabled by default then I'm worried it won't really improve
until it is.  Perhaps that's setting a higher bar than you feel is
necessary but, for my part at least, it doesn't feel like a very high
level.

> I'll take a look at the papers you sent about parallel query
> optimization, but personally I think that's putting the cart not only
> before the horse but also before the road.  For V1, we need a query
> optimization model that does not completely suck - no more.  The key
> criterion here is that this has to WORK.  There will be time enough to
> improve everything else once we reach that goal.

I agree that it's got to work, but it also needs to be generally well
designed, and have the expectation of being on by default.
Thanks,
    Stephen

* Heikki Linnakangas (hlinnakangas@vmware.com) wrote:
> On 12/19/2014 04:39 PM, Stephen Frost wrote:
> >* Marko Tiikkaja (marko@joh.to) wrote:
> >>I'd be perfectly (that means 100%) happy if it just defaulted to
> >>off, but I could turn it up to 11 whenever I needed it.  I don't
> >>believe to be the only one with this opinion, either.
> >
> >Perhaps we should reconsider our general position on hints then and
> >add them so users can define the plan to be used..  For my part, I don't
> >see this as all that much different.
> >
> >Consider if we were just adding HashJoin support today as an example.
> >Would we be happy if we had to default to enable_hashjoin = off?  Or if
> >users had to do that regularly because our costing was horrid?  It's bad
> >enough that we have to resort to those tweaks today in rare cases.
>
> This is somewhat different. Imagine that we achieve perfect
> parallelization, so that when you set enable_parallel_query=8, every
> query runs exactly 8x faster on an 8-core system, by using all eight
> cores.

To be clear, as I mentioned to Robert just now, I'm not objecting to a
GUC being added to turn off or control parallelization.  I don't want
such a GUC to be a crutch for us to lean on when it comes to questions
about the optimizer though.  We need to work through the optimizer
questions of "should this be parallelized?" and, perhaps later, "how
many ways is it sensible to parallelize this?"  I'm worried we'll take
such a GUC as a directive along the lines of "we are being told to
parallelize to exactly this level every time and for every query which
can be."  The GUC should be an input into the planner/optimizer much the
way enable_hashjoin is, unless it's being done as a *limiting* factor
for the administrator to be able to control, but we've generally avoided
doing that (see: work_mem) and, if we're going to start, we should
probably come up with an approach that addresses the considerations for
other resources too.
Thanks,
    Stephen

On 12/21/14, 12:42 AM, Amit Kapila wrote:
> On Fri, Dec 19, 2014 at 6:21 PM, Stephen Frost <sfrost@snowman.net <mailto:sfrost@snowman.net>> wrote:
> a. Instead of passing value array, just pass tuple id, but retain the
> buffer pin till master backend reads the tuple based on tupleid.
> This has side effect that we have to retain buffer pin for longer
> period of time, but again that might not have any problem in
> real world usage of parallel query.
>
> b. Instead of passing value array, pass directly the tuple which could
> be directly propagated by master backend to upper layer or otherwise
> in master backend change some code such that it could propagate the
> tuple array received via shared memory queue directly to frontend.
> Basically save the one extra cycle of form/deform tuple.
>
> Both these need some new message type and handling for same in
> Executor code.
>
> Having said above, I think we can try to optimize this in multiple
> ways, however we need additional mechanism and changes in Executor
> code which is error prone and doesn't seem to be important at this
> stage where we want the basic feature to work.

Would b require some means of ensuring we didn't try and pass raw tuples to frontends? Other than that potential
wrinkle,it seems like less work than a.
 

...

> I think there are mainly two things which can lead to benefit
> by employing parallel workers
> a. Better use of available I/O bandwidth
> b. Better use of available CPU's by doing expression evaluation
> by multiple workers.

...

> In the above tests, it seems to me that the maximum benefit due to
> 'a' is realized upto 4~8 workers

I'd think a good first estimate here would be to just use effective_io_concurrency.
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

I think one thing we could do minimize the chance of such an

error is set the value of parallel workers to be used for plan equal

to max_worker_processes if parallel_seqscan_degree is greater

than max_worker_processes.  Even if we do this, still such an

error can come if user has registered bgworker before we could

start parallel plan execution.

On Thu, Jan 1, 2015 at 12:00 PM, Fabrízio de Royes Mello
<fabriziomello@gmail.com> wrote:
> Can we check the number of free bgworkers slots to set the max workers?

The real solution here is that this patch can't throw an error if it's
unable to obtain the desired number of background workers.  It needs
to be able to smoothly degrade to a smaller number of background
workers, or none at all.  I think a lot of this work will fall out
quite naturally if this patch is reworked to use the parallel
mode/parallel context stuff, the latest version of which includes an
example of how to set up a parallel scan in such a manner that it can
run with any number of workers.

--
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Fri, Jan 2, 2015 at 5:36 AM, Amit Kapila <amit.kapila16@gmail.com> wrote:
> On Thu, Jan 1, 2015 at 11:29 PM, Robert Haas <robertmhaas@gmail.com> wrote:
>> On Thu, Jan 1, 2015 at 12:00 PM, Fabrízio de Royes Mello
>> <fabriziomello@gmail.com> wrote:
>> > Can we check the number of free bgworkers slots to set the max workers?
>>
>> The real solution here is that this patch can't throw an error if it's
>> unable to obtain the desired number of background workers.  It needs
>> to be able to smoothly degrade to a smaller number of background
>> workers, or none at all.
>
> I think handling this way can have one side effect which is that if
> we degrade to smaller number, then the cost of plan (which was
> decided by optimizer based on number of parallel workers) could
> be more than non-parallel scan.
> Ideally before finalizing the parallel plan we should reserve the
> bgworkers required to execute that plan, but I think as of now
> we can workout a solution without it.

I don't think this is very practical.  When cached plans are in use,
we can have a bunch of plans sitting around that may or may not get
reused at some point in the future, possibly far in the future.  The
current situation, which I think we want to maintain, is that such
plans hold no execution-time resources (e.g. locks) and, generally,
don't interfere with other things people might want to execute on the
system.  Nailing down a bunch of background workers just in case we
might want to use them in the future would be pretty unfriendly.

I think it's right to view this in the same way we view work_mem.  We
plan on the assumption that an amount of memory equal to work_mem will
be available at execution time, without actually reserving it.  If the
plan happens to need that amount of memory and if it actually isn't
available when needed, then performance will suck; conceivably, the
OOM killer might trigger.  But it's the user's job to avoid this by
not setting work_mem too high in the first place.  Whether this system
is for the best is arguable: one can certainly imagine a system where,
if there's not enough memory at execution time, we consider
alternatives like (a) replanning with a lower memory target, (b)
waiting until more memory is available, or (c) failing outright in
lieu of driving the machine into swap.  But devising such a system is
complicated -- for example, replanning with a lower memory target
might be latch onto a far more expensive plan, such that we would have
been better off waiting for more memory to be available; yet trying to
waiting until more memory is available might result in waiting
forever.  And that's why we don't have such a system.

We don't need to do any better here.  The GUC should tell us how many
parallel workers we should anticipate being able to obtain.  If other
settings on the system, or the overall system load, preclude us from
obtaining that number of parallel workers, then the query will take
longer to execute; and the plan might be sub-optimal.  If that happens
frequently, the user should lower the planner GUC to a level that
reflects the resources actually likely to be available at execution
time.

By the way, another area where this kind of effect crops up is with
the presence of particular disk blocks in shared_buffers or the system
buffer cache.  Right now, the planner makes no attempt to cost a scan
of a frequently-used, fully-cached relation different than a
rarely-used, probably-not-cached relation; and that sometimes leads to
bad plans.  But if it did try to do that, then we'd have the same kind
of problem discussed here -- things might change between planning and
execution, or even after the beginning of execution.  Also, we might
get nasty feedback effects: since the relation isn't cached, we view a
plan that would involve reading it in as very expensive, and avoid
such a plan.  However, we might be better off picking the "slow" plan
anyway, because it might be that once we've read the data once it will
stay cached and run much more quickly than some plan that seems better
starting from a cold cache.

--
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

* Robert Haas (robertmhaas@gmail.com) wrote:
> I think it's right to view this in the same way we view work_mem.  We
> plan on the assumption that an amount of memory equal to work_mem will
> be available at execution time, without actually reserving it.

Agreed- this seems like a good approach for how to address this.  We
should still be able to end up with plans which use less than the max
possible parallel workers though, as I pointed out somewhere up-thread.
This is also similar to work_mem- we certainly have plans which don't
expect to use all of work_mem and others that expect to use all of it
(per node, of course).
Thanks,
    Stephen

On 1/5/15, 9:21 AM, Stephen Frost wrote:
> * Robert Haas (robertmhaas@gmail.com) wrote:
>> I think it's right to view this in the same way we view work_mem.  We
>> plan on the assumption that an amount of memory equal to work_mem will
>> be available at execution time, without actually reserving it.
>
> Agreed- this seems like a good approach for how to address this.  We
> should still be able to end up with plans which use less than the max
> possible parallel workers though, as I pointed out somewhere up-thread.
> This is also similar to work_mem- we certainly have plans which don't
> expect to use all of work_mem and others that expect to use all of it
> (per node, of course).

I agree, but we should try and warn the user if they set parallel_seqscan_degree close to max_worker_processes, or at
leastgive some indication of what's going on. This is something you could end up beating your head on wondering why
it'snot working.
 

Perhaps we could have EXPLAIN throw a warning if a plan is likely to get less than parallel_seqscan_degree number of
workers.
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

* Jim Nasby (Jim.Nasby@BlueTreble.com) wrote:
> On 1/5/15, 9:21 AM, Stephen Frost wrote:
> >* Robert Haas (robertmhaas@gmail.com) wrote:
> >>I think it's right to view this in the same way we view work_mem.  We
> >>plan on the assumption that an amount of memory equal to work_mem will
> >>be available at execution time, without actually reserving it.
> >
> >Agreed- this seems like a good approach for how to address this.  We
> >should still be able to end up with plans which use less than the max
> >possible parallel workers though, as I pointed out somewhere up-thread.
> >This is also similar to work_mem- we certainly have plans which don't
> >expect to use all of work_mem and others that expect to use all of it
> >(per node, of course).
>
> I agree, but we should try and warn the user if they set parallel_seqscan_degree close to max_worker_processes, or at
leastgive some indication of what's going on. This is something you could end up beating your head on wondering why
it'snot working. 
>
> Perhaps we could have EXPLAIN throw a warning if a plan is likely to get less than parallel_seqscan_degree number of
workers.

Yeah, if we come up with a plan for X workers and end up not being able
to spawn that many then I could see that being worth a warning or notice
or something.  Not sure what EXPLAIN has to do anything with it..
Thanks,
    Stephen

Amit,

* Amit Kapila (amit.kapila16@gmail.com) wrote:
> On Fri, Dec 19, 2014 at 7:57 PM, Stephen Frost <sfrost@snowman.net> wrote:
> > There's certainly documentation available from the other RDBMS' which
> > already support parallel query, as one source.  Other academic papers
> > exist (and once you've linked into one, the references and prior work
> > helps bring in others).  Sadly, I don't currently have ACM access (might
> > have to change that..), but there are publicly available papers also,
>
> I have gone through couple of papers and what some other databases
> do in case of parallel sequential scan and here is brief summarization
> of same and how I am planning to handle in the patch:

Great, thanks!

> Costing:
> In one of the paper's [1] suggested by you, below is the summarisation:
>   a. Startup costs are negligible if processes can be reused
>            rather than created afresh.
> b. Communication cost consists of the CPU cost of sending
>            and receiving messages.
>   c. Communication costs can exceed the cost of operators such
>             as scanning, joining or grouping
> These findings lead to the important conclusion that
> Query optimization should be concerned with communication costs
> but not with startup costs.
>
> In our case as currently we don't have a mechanism to reuse parallel
> workers, so we need to account for that cost as well.  So based on that,
> I am planing to add three new parameters cpu_tuple_comm_cost,
> parallel_setup_cost, parallel_startup_cost
> *  cpu_tuple_comm_cost - Cost of CPU time to pass a tuple from worker
>     to master backend with default value
>     DEFAULT_CPU_TUPLE_COMM_COST as 0.1, this will be multiplied
>     with tuples expected to be selected
> *  parallel_setup_cost - Cost of setting up shared memory for parallelism
>    with default value as 100.0
>  *  parallel_startup_cost  - Cost of starting up parallel workers with
> default
>     value as 1000.0 multiplied by number of workers decided for scan.
>
> I will do some experiments to finalise the default values, but in general,
> I feel developing cost model on above parameters is good.

The parameters sound reasonable but I'm a bit worried about the way
you're describing the implementation.  Specifically this comment:

"Cost of starting up parallel workers with default value as 1000.0
multiplied by number of workers decided for scan."

That appears to imply that we'll decide on the number of workers, figure
out the cost, and then consider "parallel" as one path and
"not-parallel" as another.  I'm worried that if I end up setting the max
parallel workers to 32 for my big, beefy, mostly-single-user system then
I'll actually end up not getting parallel execution because we'll always
be including the full startup cost of 32 threads.  For huge queries,
it'll probably be fine, but there's a lot of room to parallelize things
at levels less than 32 which we won't even consider.

What I was advocating for up-thread was to consider multiple "parallel"
paths and to pick whichever ends up being the lowest overall cost.  The
flip-side to that is increased planning time.  Perhaps we can come up
with an efficient way of working out where the break-point is based on
the non-parallel cost and go at it from that direction instead of
building out whole paths for each increment of parallelism.

I'd really like to be able to set the 'max parallel' high and then have
the optimizer figure out how many workers should actually be spawned for
a given query.

> Execution:
> Most other databases does partition level scan for partition on
> different disks by each individual parallel worker.  However,
> it seems amazon dynamodb [2] also works on something
> similar to what I have used in patch which means on fixed
> blocks.  I think this kind of strategy seems better than dividing
> the blocks at runtime because dividing randomly the blocks
> among workers could lead to random scan for a parallel
> sequential scan.

Yeah, we also need to consider the i/o side of this, which will
definitely be tricky.  There are i/o systems out there which are faster
than a single CPU and ones where a single CPU can manage multiple i/o
channels.  There are also cases where the i/o system handles sequential
access nearly as fast as random and cases where sequential is much
faster than random.  Where we can get an idea of that distinction is
with seq_page_cost vs. random_page_cost as folks running on SSDs tend to
lower random_page_cost from the default to indicate that.

> Also I find in whatever I have read (Oracle, dynamodb) that most
> databases divide work among workers and master backend acts
> as coordinator, atleast that's what I could understand.

Yeah, I agree that's more typical.  Robert's point that the master
backend should participate is interesting but, as I recall, it was based
on the idea that the master could finish faster than the worker- but if
that's the case then we've planned it out wrong from the beginning.
Thanks!
    Stephen

Amit,

* Amit Kapila (amit.kapila16@gmail.com) wrote:
> On Fri, Jan 9, 2015 at 1:02 AM, Jim Nasby <Jim.Nasby@bluetreble.com> wrote:
> > I agree, but we should try and warn the user if they set
> > parallel_seqscan_degree close to max_worker_processes, or at least give
> > some indication of what's going on. This is something you could end up
> > beating your head on wondering why it's not working.
>
> Yet another way to handle the case when enough workers are not
> available is to let user  specify the desired minimum percentage of
> requested parallel workers with parameter like
> PARALLEL_QUERY_MIN_PERCENT. For  example, if you specify
> 50 for this parameter, then at least 50% of the parallel workers
> requested for any  parallel operation must be available in order for
> the operation to succeed else it will give error. If the value is set to
> null, then all parallel operations will proceed as long as at least two
> parallel workers are available for processing.

Ugh.  I'm not a fan of this..  Based on how we're talking about modeling
this, if we decide to parallelize at all, then we expect it to be a win.
I don't like the idea of throwing an error if, at execution time, we end
up not being able to actually get the number of workers we want-
instead, we should degrade gracefully all the way back to serial, if
necessary.  Perhaps we should send a NOTICE or something along those
lines to let the user know we weren't able to get the level of
parallelization that the plan originally asked for, but I really don't
like just throwing an error.

Now, for debugging purposes, I could see such a parameter being
available but it should default to 'off/never-fail'.
Thanks,
    Stephen

On 01/09/2015 08:01 PM, Stephen Frost wrote:
> Amit,
> 
> * Amit Kapila (amit.kapila16@gmail.com) wrote:
>> On Fri, Jan 9, 2015 at 1:02 AM, Jim Nasby <Jim.Nasby@bluetreble.com> wrote:
>>> I agree, but we should try and warn the user if they set
>>> parallel_seqscan_degree close to max_worker_processes, or at least give
>>> some indication of what's going on. This is something you could end up
>>> beating your head on wondering why it's not working.
>>
>> Yet another way to handle the case when enough workers are not
>> available is to let user  specify the desired minimum percentage of
>> requested parallel workers with parameter like
>> PARALLEL_QUERY_MIN_PERCENT. For  example, if you specify
>> 50 for this parameter, then at least 50% of the parallel workers
>> requested for any  parallel operation must be available in order for
>> the operation to succeed else it will give error. If the value is set to
>> null, then all parallel operations will proceed as long as at least two
>> parallel workers are available for processing.
> 
> Ugh.  I'm not a fan of this..  Based on how we're talking about modeling
> this, if we decide to parallelize at all, then we expect it to be a win.
> I don't like the idea of throwing an error if, at execution time, we end
> up not being able to actually get the number of workers we want-
> instead, we should degrade gracefully all the way back to serial, if
> necessary.  Perhaps we should send a NOTICE or something along those
> lines to let the user know we weren't able to get the level of
> parallelization that the plan originally asked for, but I really don't
> like just throwing an error.

yeah this seems like the the behaviour I would expect, if we cant get
enough parallel workers we should just use as much as we can get.
Everything else and especially erroring out will just cause random
application failures and easy DoS vectors.
I think all we need initially is being able to specify a "maximum number
of workers per query" as well as a "maximum number of workers in total
for parallel operations".


> 
> Now, for debugging purposes, I could see such a parameter being
> available but it should default to 'off/never-fail'.

not sure what it really would be useful for - if I execute a query I
would truely expect it to get answered - if it can be made faster if
done in parallel thats nice but why would I want it to fail?


Stefan

* Stefan Kaltenbrunner (stefan@kaltenbrunner.cc) wrote:
> On 01/09/2015 08:01 PM, Stephen Frost wrote:
> > Now, for debugging purposes, I could see such a parameter being
> > available but it should default to 'off/never-fail'.
>
> not sure what it really would be useful for - if I execute a query I
> would truely expect it to get answered - if it can be made faster if
> done in parallel thats nice but why would I want it to fail?

I was thinking for debugging only, though I'm not really sure why you'd
need it if you get a NOTICE when you don't end up with all the workers
you expect.
Thanks,
    Stephen

On 1/9/15, 3:34 PM, Stephen Frost wrote:
> * Stefan Kaltenbrunner (stefan@kaltenbrunner.cc) wrote:
>> On 01/09/2015 08:01 PM, Stephen Frost wrote:
>>> Now, for debugging purposes, I could see such a parameter being
>>> available but it should default to 'off/never-fail'.
>>
>> not sure what it really would be useful for - if I execute a query I
>> would truely expect it to get answered - if it can be made faster if
>> done in parallel thats nice but why would I want it to fail?
>
> I was thinking for debugging only, though I'm not really sure why you'd
> need it if you get a NOTICE when you don't end up with all the workers
> you expect.

Yeah, debugging is my concern as well. You're working on a query, you expect it to be using parallelism, and EXPLAIN is
showingit's not. Now you're scratching your head.
 
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

On 1/9/15, 11:24 AM, Stephen Frost wrote:
> What I was advocating for up-thread was to consider multiple "parallel"
> paths and to pick whichever ends up being the lowest overall cost.  The
> flip-side to that is increased planning time.  Perhaps we can come up
> with an efficient way of working out where the break-point is based on
> the non-parallel cost and go at it from that direction instead of
> building out whole paths for each increment of parallelism.

I think at some point we'll need the ability to stop planning part-way through for queries producing really small
estimates.If the first estimate you get is 1000 units, does it really make sense to do something like try every
possiblejoin permutation, or attempt to parallelize?
 
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

* Amit Kapila (amit.kapila16@gmail.com) wrote:
> At this moment if we can ensure that parallel plan should not be selected
> for cases where it will perform poorly is more than enough considering
> we have lots of other work left to even make any parallel operation work.

The problem with this approach is that it doesn't consider any options
between 'serial' and 'parallelize by factor X'.  If the startup cost is
1000 and the factor is 32, then a seqscan which costs 31000 won't ever
be parallelized, even though a factor of 8 would have parallelized it.

You could forget about the per-process startup cost entirely, in fact,
and simply say "only parallelize if it's more than X".

Again, I don't like the idea of designing this with the assumption that
the user dictates the right level of parallelization for each and every
query.  I'd love to go out and tell users "set the factor to the number
of CPUs you have and we'll just use what makes sense."

The same goes for max number of backends.  If we set the parallel level
to the number of CPUs and set the max backends to the same, then we end
up with only one parallel query running at a time, ever.  That's
terrible.  Now, we could set the parallel level lower or set the max
backends higher, but either way we're going to end up either using less
than we could or over-subscribing, neither of which is good.

I agree that this makes it a bit different from work_mem, but in this
case there's an overall max in the form of the maximum number of
background workers.  If we had something similar for work_mem, then we
could set that higher and still trust the system to only use the amount
of memory necessary (eg: a hashjoin doesn't use all available work_mem
and neither does a sort, unless the set is larger than available
memory).

> > > Execution:
> > > Most other databases does partition level scan for partition on
> > > different disks by each individual parallel worker.  However,
> > > it seems amazon dynamodb [2] also works on something
> > > similar to what I have used in patch which means on fixed
> > > blocks.  I think this kind of strategy seems better than dividing
> > > the blocks at runtime because dividing randomly the blocks
> > > among workers could lead to random scan for a parallel
> > > sequential scan.
> >
> > Yeah, we also need to consider the i/o side of this, which will
> > definitely be tricky.  There are i/o systems out there which are faster
> > than a single CPU and ones where a single CPU can manage multiple i/o
> > channels.  There are also cases where the i/o system handles sequential
> > access nearly as fast as random and cases where sequential is much
> > faster than random.  Where we can get an idea of that distinction is
> > with seq_page_cost vs. random_page_cost as folks running on SSDs tend to
> > lower random_page_cost from the default to indicate that.
> >
> I am not clear, do you expect anything different in execution strategy
> than what I have mentioned or does that sound reasonable to you?

What I'd like is a way to figure out the right amount of CPU for each
tablespace (0.25, 1, 2, 4, etc) and then use that many.  Using a single
CPU for each tablespace is likely to starve the CPU or starve the I/O
system and I'm not sure if there's a way to address that.

Note that I intentionally said tablespace there because that's how users
can tell us what the different i/o channels are.  I realize this ends up
going beyond the current scope, but the parallel seqscan at the per
relation level will only ever be using one i/o channel.  It'd be neat if
we could work out how fast that i/o channel is vs. the CPUs and
determine how many CPUs are necessary to keep up with the i/o channel
and then use more-or-less exactly that many for the scan.

I agree that some of this can come later but I worry that starting out
with a design that expects to always be told exactly how many CPUs to
use when running a parallel query will be difficult to move away from
later.
Thanks,
    Stephen

On Thu, Jan 8, 2015 at 6:42 AM, Amit Kapila <amit.kapila16@gmail.com> wrote:
> Are we sure that in such cases we will consume work_mem during
> execution?  In cases of parallel_workers we are sure to an extent
> that if we reserve the workers then we will use it during execution.
> Nonetheless, I have proceded and integrated the parallel_seq scan
> patch with v0.3 of parallel_mode patch posted by you at below link:
> http://www.postgresql.org/message-id/CA+TgmoYmp_=XcJEhvJZt9P8drBgW-pDpjHxBhZA79+M4o-CZQA@mail.gmail.com

That depends on the costing model.  It makes no sense to do a parallel
sequential scan on a small relation, because the user backend can scan
the whole thing itself faster than the workers can start up.  I
suspect it may also be true that the useful amount of parallelism
increases the larger the relation gets (but maybe not).

> 2. To enable two types of shared memory queue's (error queue and
> tuple queue), we need to ensure that we switch to appropriate queue
> during communication of various messages from parallel worker
> to master backend.  There are two ways to do it
>    a.  Save the information about error queue during startup of parallel
>         worker (ParallelMain()) and then during error, set the same (switch
>         to error queue in errstart() and switch back to tuple queue in
>         errfinish() and errstart() in case errstart() doesn't need to
> propagate
>         error).
>    b.  Do something similar as (a) for tuple queue in printtup or other
> place
>         if any for non-error messages.
> I think approach (a) is slightly better as compare to approach (b) as
> we need to switch many times for tuple queue (for each tuple) and
> there could be multiple places where we need to do the same.  For now,
> I have used approach (a) in Patch which needs some more work if we
> agree on the same.

I don't think you should be "switching" queues.  The tuples should be
sent to the tuple queue, and errors and notices to the error queue.

> 3. As per current implementation of Parallel_seqscan, it needs to use
> some information from parallel.c which was not exposed, so I have
> exposed the same by moving it to parallel.h.  Information that is required
> is as follows:
> ParallelWorkerNumber, FixedParallelState and shm keys -
>     This is used to decide the blocks that needs to be scanned.
>     We might change it in future the way parallel scan/work distribution
>     is done, but I don't see any harm in exposing this information.

Hmm.  I can see why ParallelWorkerNumber might need to be exposed, but
the other stuff seems like it shouldn't be.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Thu, Jan 8, 2015 at 2:46 PM, Stephen Frost <sfrost@snowman.net> wrote:
> Yeah, if we come up with a plan for X workers and end up not being able
> to spawn that many then I could see that being worth a warning or notice
> or something.  Not sure what EXPLAIN has to do anything with it..

That seems mighty odd to me.  If there are 8 background worker
processes available, and you allow each session to use at most 4, then
when there are >2 sessions trying to do parallelism at the same time,
they might not all get their workers.  Emitting a notice for that
seems like it would be awfully chatty.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Fri, Jan 9, 2015 at 12:24 PM, Stephen Frost <sfrost@snowman.net> wrote:
> The parameters sound reasonable but I'm a bit worried about the way
> you're describing the implementation.  Specifically this comment:
>
> "Cost of starting up parallel workers with default value as 1000.0
> multiplied by number of workers decided for scan."
>
> That appears to imply that we'll decide on the number of workers, figure
> out the cost, and then consider "parallel" as one path and
> "not-parallel" as another.  [...]
> I'd really like to be able to set the 'max parallel' high and then have
> the optimizer figure out how many workers should actually be spawned for
> a given query.

+1.

> Yeah, we also need to consider the i/o side of this, which will
> definitely be tricky.  There are i/o systems out there which are faster
> than a single CPU and ones where a single CPU can manage multiple i/o
> channels.  There are also cases where the i/o system handles sequential
> access nearly as fast as random and cases where sequential is much
> faster than random.  Where we can get an idea of that distinction is
> with seq_page_cost vs. random_page_cost as folks running on SSDs tend to
> lower random_page_cost from the default to indicate that.

On my MacOS X system, I've already seen cases where my parallel_count
module runs incredibly slowly some of the time.  I believe that this
is because having multiple workers reading the relation block-by-block
at the same time causes the OS to fail to realize that it needs to do
aggressive readahead.  I suspect we're going to need to account for
this somehow.

> Yeah, I agree that's more typical.  Robert's point that the master
> backend should participate is interesting but, as I recall, it was based
> on the idea that the master could finish faster than the worker- but if
> that's the case then we've planned it out wrong from the beginning.

So, if the workers have been started but aren't keeping up, the master
should do nothing until they produce tuples rather than participating?That doesn't seem right.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

* Robert Haas (robertmhaas@gmail.com) wrote:
> On Thu, Jan 8, 2015 at 2:46 PM, Stephen Frost <sfrost@snowman.net> wrote:
> > Yeah, if we come up with a plan for X workers and end up not being able
> > to spawn that many then I could see that being worth a warning or notice
> > or something.  Not sure what EXPLAIN has to do anything with it..
>
> That seems mighty odd to me.  If there are 8 background worker
> processes available, and you allow each session to use at most 4, then
> when there are >2 sessions trying to do parallelism at the same time,
> they might not all get their workers.  Emitting a notice for that
> seems like it would be awfully chatty.

Yeah, agreed, it could get quite noisy.  Did you have another thought
for how to address the concern raised?  Specifically, that you might not
get as many workers as you thought you would?
Thanks,
    Stephen

* Robert Haas (robertmhaas@gmail.com) wrote:
> On Fri, Jan 9, 2015 at 12:24 PM, Stephen Frost <sfrost@snowman.net> wrote:
> > Yeah, we also need to consider the i/o side of this, which will
> > definitely be tricky.  There are i/o systems out there which are faster
> > than a single CPU and ones where a single CPU can manage multiple i/o
> > channels.  There are also cases where the i/o system handles sequential
> > access nearly as fast as random and cases where sequential is much
> > faster than random.  Where we can get an idea of that distinction is
> > with seq_page_cost vs. random_page_cost as folks running on SSDs tend to
> > lower random_page_cost from the default to indicate that.
>
> On my MacOS X system, I've already seen cases where my parallel_count
> module runs incredibly slowly some of the time.  I believe that this
> is because having multiple workers reading the relation block-by-block
> at the same time causes the OS to fail to realize that it needs to do
> aggressive readahead.  I suspect we're going to need to account for
> this somehow.

So, for my 2c, I've long expected us to parallelize at the relation-file
level for these kinds of operations.  This goes back to my other
thoughts on how we should be thinking about parallelizing inbound data
for bulk data loads but it seems appropriate to consider it here also.
One of the issues there is that 1G still feels like an awful lot for a
minimum work size for each worker and it would mean we don't parallelize
for relations less than that size.

On a random VM on my personal server, an uncached 1G read takes over
10s.  Cached it's less than half that, of course.  This is all spinning
rust (and only 7200 RPM at that) and there's a lot of other stuff going
on but that still seems like too much of a chunk to give to one worker
unless the overall data set to go through is really large.

There's other issues in there too, of course, if we're dumping data in
like this then we have to either deal with jagged relation files somehow
or pad the file out to 1G, and that doesn't even get into the issues
around how we'd have to redesign the interfaces for relation access and
how this thinking is an utter violation of the modularity we currently
have there.

> > Yeah, I agree that's more typical.  Robert's point that the master
> > backend should participate is interesting but, as I recall, it was based
> > on the idea that the master could finish faster than the worker- but if
> > that's the case then we've planned it out wrong from the beginning.
>
> So, if the workers have been started but aren't keeping up, the master
> should do nothing until they produce tuples rather than participating?
>  That doesn't seem right.

Having the master jump in and start working could screw things up also
though.  Perhaps we need the master to start working as a fail-safe but
not plan on having things go that way?  Having more processes trying to
do X doesn't always result in things getting better and the master needs
to keep up with all the tuples being thrown at it from the workers.
Thanks,
    Stephen

Amit,

* Amit Kapila (amit.kapila16@gmail.com) wrote:
> On Sun, Jan 11, 2015 at 9:09 AM, Robert Haas <robertmhaas@gmail.com> wrote:
> > I don't think you should be "switching" queues.  The tuples should be
> > sent to the tuple queue, and errors and notices to the error queue.

Agreed.

> To achieve what you said (The tuples should be sent to the tuple
> queue, and errors and notices to the error queue.), we need to
> switch the queues.
> The difficulty here is that once we set the queue (using
> pq_redirect_to_shm_mq()) through which the communication has to
> happen, it will use the same unless we change again the queue
> using pq_redirect_to_shm_mq().  For example, assume we have
> initially set error queue (using pq_redirect_to_shm_mq()) then to
> send tuples, we need to call pq_redirect_to_shm_mq() to
> set the tuple queue as the queue that needs to be used for communication
> and again if error happens then we need to do the same for error
> queue.
> Do you have any other idea to achieve the same?

I think what Robert's getting at here is that pq_redirect_to_shm_mq()
might be fine for the normal data heading back, but we need something
separate for errors and notices.  Switching everything back and forth
between the normal and error queues definitely doesn't sound right to
me- they need to be independent.

In other words, you need to be able to register a "normal data" queue
and then you need to also register a "error/notice" queue and have
errors and notices sent there directly.  Going off of what I recall,
can't this be done by having the callbacks which are registered for
sending data back look at what they're being asked to send and then
decide which queue it's appropriate for out of the set which have been
registered so far?
Thanks,
    Stephen

On 01/11/2015 11:27 AM, Stephen Frost wrote:
> * Robert Haas (robertmhaas@gmail.com) wrote:
>> On Thu, Jan 8, 2015 at 2:46 PM, Stephen Frost <sfrost@snowman.net> wrote:
>>> Yeah, if we come up with a plan for X workers and end up not being able
>>> to spawn that many then I could see that being worth a warning or notice
>>> or something.  Not sure what EXPLAIN has to do anything with it..
>>
>> That seems mighty odd to me.  If there are 8 background worker
>> processes available, and you allow each session to use at most 4, then
>> when there are >2 sessions trying to do parallelism at the same time,
>> they might not all get their workers.  Emitting a notice for that
>> seems like it would be awfully chatty.
> 
> Yeah, agreed, it could get quite noisy.  Did you have another thought
> for how to address the concern raised?  Specifically, that you might not
> get as many workers as you thought you would?

Wild idea: What about dealing with it as some sort of statistic - ie
track some global counts in the stats collector or on a per-query base
in pg_stat_activity and/or through pg_stat_statements?

Not sure why it is that important to get it on a per-query base, imho it
is simply a configuration limit we have set (similiar to work_mem or
when switching to geqo) - we dont report "per query" through
notice/warning there either (though the effect is kind visible in explain).


Stefan

On Sat, Jan 10, 2015 at 11:14 PM, Amit Kapila <amit.kapila16@gmail.com> wrote:
>> I don't think you should be "switching" queues.  The tuples should be
>> sent to the tuple queue, and errors and notices to the error queue.
> To achieve what you said (The tuples should be sent to the tuple
> queue, and errors and notices to the error queue.), we need to
> switch the queues.
> The difficulty here is that once we set the queue (using
> pq_redirect_to_shm_mq()) through which the communication has to
> happen, it will use the same unless we change again the queue
> using pq_redirect_to_shm_mq().  For example, assume we have
> initially set error queue (using pq_redirect_to_shm_mq()) then to
> send tuples, we need to call pq_redirect_to_shm_mq() to
> set the tuple queue as the queue that needs to be used for communication
> and again if error happens then we need to do the same for error
> queue.
> Do you have any other idea to achieve the same?

Yeah, you need two separate global variables pointing to shm_mq
objects, one of which gets used by pqmq.c for errors and the other of
which gets used by printtup.c for tuples.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Sun, Jan 11, 2015 at 5:27 AM, Stephen Frost <sfrost@snowman.net> wrote:
> * Robert Haas (robertmhaas@gmail.com) wrote:
>> On Thu, Jan 8, 2015 at 2:46 PM, Stephen Frost <sfrost@snowman.net> wrote:
>> > Yeah, if we come up with a plan for X workers and end up not being able
>> > to spawn that many then I could see that being worth a warning or notice
>> > or something.  Not sure what EXPLAIN has to do anything with it..
>>
>> That seems mighty odd to me.  If there are 8 background worker
>> processes available, and you allow each session to use at most 4, then
>> when there are >2 sessions trying to do parallelism at the same time,
>> they might not all get their workers.  Emitting a notice for that
>> seems like it would be awfully chatty.
>
> Yeah, agreed, it could get quite noisy.  Did you have another thought
> for how to address the concern raised?  Specifically, that you might not
> get as many workers as you thought you would?

I'm not sure why that's a condition in need of special reporting.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Sun, Jan 11, 2015 at 6:01 AM, Stephen Frost <sfrost@snowman.net> wrote:
> So, for my 2c, I've long expected us to parallelize at the relation-file
> level for these kinds of operations.  This goes back to my other
> thoughts on how we should be thinking about parallelizing inbound data
> for bulk data loads but it seems appropriate to consider it here also.
> One of the issues there is that 1G still feels like an awful lot for a
> minimum work size for each worker and it would mean we don't parallelize
> for relations less than that size.

Yes, I think that's a killer objection.

> [ .. ] and
> how this thinking is an utter violation of the modularity we currently
> have there.

As is that.

My thinking is more along the lines that we might need to issue
explicit prefetch requests when doing a parallel sequential scan, to
make up for any failure of the OS to do that for us.

>> So, if the workers have been started but aren't keeping up, the master
>> should do nothing until they produce tuples rather than participating?
>>  That doesn't seem right.
>
> Having the master jump in and start working could screw things up also
> though.

I don't think there's any reason why that should screw things up.
There's no reason why the master's participation should look any
different from one more worker.  Look at my parallel_count code on the
other thread to see what I mean: the master and all the workers are
running the same code, and if fewer worker show up than expected, or
run unduly slowly, it's easily tolerated.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Sun, Jan 11, 2015 at 6:09 AM, Stephen Frost <sfrost@snowman.net> wrote:
> I think what Robert's getting at here is that pq_redirect_to_shm_mq()
> might be fine for the normal data heading back, but we need something
> separate for errors and notices.  Switching everything back and forth
> between the normal and error queues definitely doesn't sound right to
> me- they need to be independent.

You've got that backwards.  pq_redirect_to_shm_mq() handles errors and
notices, but we need something separate for the tuple stream.

> In other words, you need to be able to register a "normal data" queue
> and then you need to also register a "error/notice" queue and have
> errors and notices sent there directly.  Going off of what I recall,
> can't this be done by having the callbacks which are registered for
> sending data back look at what they're being asked to send and then
> decide which queue it's appropriate for out of the set which have been
> registered so far?

It's pretty simple, really.  The functions that need to use the tuple
queue are in printtup.c; those, and only those, need to be modified to
write to the other queue.

Or, possibly, we should pass the tuples around in their native format
instead of translating them into binary form and then reconstituting
them on the other end.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On 1/11/15 3:57 PM, Robert Haas wrote:
> On Sun, Jan 11, 2015 at 5:27 AM, Stephen Frost <sfrost@snowman.net> wrote:
>> * Robert Haas (robertmhaas@gmail.com) wrote:
>>> On Thu, Jan 8, 2015 at 2:46 PM, Stephen Frost <sfrost@snowman.net> wrote:
>>>> Yeah, if we come up with a plan for X workers and end up not being able
>>>> to spawn that many then I could see that being worth a warning or notice
>>>> or something.  Not sure what EXPLAIN has to do anything with it..
>>>
>>> That seems mighty odd to me.  If there are 8 background worker
>>> processes available, and you allow each session to use at most 4, then
>>> when there are >2 sessions trying to do parallelism at the same time,
>>> they might not all get their workers.  Emitting a notice for that
>>> seems like it would be awfully chatty.
>>
>> Yeah, agreed, it could get quite noisy.  Did you have another thought
>> for how to address the concern raised?  Specifically, that you might not
>> get as many workers as you thought you would?
>
> I'm not sure why that's a condition in need of special reporting.

The case raised before (that I think is valid) is: what if you have a query that is massively parallel. You expect it
toget 60 cores on the server and take 10 minutes. Instead it gets 10 and takes an hour (or worse, 1 and takes 10
hours).

Maybe it's not worth dealing with that in the first version, but I expect it will come up very quickly. We better make
surewe're not painting ourselves in a corner.
 
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

On Tue, Jan 13, 2015 at 6:25 AM, John Gorman <johngorman2@gmail.com> wrote:
> One approach that I has worked well for me is to break big jobs into much
> smaller bite size tasks. Each task is small enough to complete quickly.
>
> We add the tasks to a task queue and spawn a generic worker pool which eats
> through the task queue items.
>
> This solves a lot of problems.
>
> - Small to medium jobs can be parallelized efficiently.
> - No need to split big jobs perfectly.
> - We don't get into a situation where we are waiting around for a worker to
> finish chugging through a huge task while the other workers sit idle.
> - Worker memory footprint is tiny so we can afford many of them.
> - Worker pool management is a well known problem.
> - Worker spawn time disappears as a cost factor.
> - The worker pool becomes a shared resource that can be managed and reported
> on and becomes considerably more predictable.

I think this is a good idea, but for now I would like to keep our
goals somewhat more modest: let's see if we can get parallel
sequential scan, and only parallel sequential scan, working and
committed.  Ultimately, I think we may need something like what you're
talking about, because if you have a query with three or six or twelve
different parallelizable operations in it, you want the available CPU
resources to switch between those as their respective needs may
dictate.  You certainly don't want to spawn a separate pool of workers
for each scan.

But I think getting that all working in the first version is probably
harder than what we should attempt.  We have a bunch of problems to
solve here just around parallel sequential scan and the parallel mode
infrastructure: heavyweight locking, prefetching, the cost model, and
so on.  Trying to add to that all of the problems that might attend on
a generic task queueing infrastructure fills me with no small amount
of fear.

--
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On 1/13/15 9:42 PM, Amit Kapila wrote:
> As an example one of the the strategy
> could be if the table size is X MB and there are 8 workers, then
> divide the work as X/8 MB for each worker (which I have currently
> used in patch) and another could be each worker does scan
> 1 block at a time and then check some global structure to see which
> next block it needs to scan, according to me this could lead to random
> scan.  I have read that some other databases also divide the work
> based on partitions or segments (size of segment is not very clear).

Long-term I think we'll want a mix between the two approaches. Simply doing something like blkno % num_workers is going
tocause imbalances, but trying to do this on a per-block basis seems like too much overhead.
 

Also long-term, I think we also need to look at a more specialized version of parallelism at the IO layer. For example,
duringan index scan you'd really like to get IO requests for heap blocks started in the background while the backend is
focusedon the mechanics of the index scan itself. While this could be done with the stuff Robert has written I have to
wonderif it'd be a lot more efficient to use fadvise or AIO. Or perhaps it would just be better to deal with an entire
indexpage (remembering TIDs) and then hit the heap.
 

But I agree with Robert; there's a lot yet to be done just to get *any* kind of parallel execution working before we
startthinking about how to optimize it.
 
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

On Wed, Jan 14, 2015 at 9:00 PM, Jim Nasby <Jim.Nasby@bluetreble.com> wrote:
> Simply doing
> something like blkno % num_workers is going to cause imbalances,

Yes.

> but trying
> to do this on a per-block basis seems like too much overhead.

...but no.  Or at least, I doubt it.  The cost of handing out blocks
one at a time is that, for each block, a worker's got to grab a
spinlock, increment and record the block number counter, and release
the spinlock.  Or, use an atomic add.  Now, it's true that spinlock
cycles and atomic ops can have sometimes impose severe overhead, but
you have to look at it as a percentage of the overall work being done.
In this case, the backend has to read, pin, and lock the page and
process every tuple on the page.  Processing every tuple on the page
may involve de-TOASTing the tuple (leading to many more page
accesses), or evaluating a complex expression, or hitting CLOG to
check visibility, but even if it doesn't, I think the amount of work
that it takes to process all the tuples on the page will be far larger
than the cost of one atomic increment operation per block.

As mentioned downthread, a far bigger consideration is the I/O pattern
we create.  A sequential scan is so-called because it reads the
relation sequentially.  If we destroy that property, we will be more
than slightly sad.  It might be OK to do sequential scans of, say,
each 1GB segment separately, but I'm pretty sure it would be a real
bad idea to read 8kB at a time at blocks 0, 64, 128, 1, 65, 129, ...

What I'm thinking about is that we might have something like this:

struct this_lives_in_dynamic_shared_memory
{   BlockNumber last_block;   Size prefetch_distance;   Size prefetch_increment;   slock_t mutex;   BlockNumber
next_prefetch_block;  BlockNumber next_scan_block;
 
};

Each worker takes the mutex and checks whether next_prefetch_block -
next_scan_block < prefetch_distance and also whether
next_prefetch_block < last_block.  If both are true, it prefetches
some number of additional blocks, as specified by prefetch_increment.
Otherwise, it increments next_scan_block and scans the block
corresponding to the old value.

So in this way, the prefetching runs ahead of the scan by a
configurable amount (prefetch_distance), which should be chosen so
that the prefetches have time to compete before the scan actually
reaches those blocks.  Right now, of course, we rely on the operating
system to prefetch for sequential scans, but I have a strong hunch
that may not work on all systems if there are multiple processes doing
the reads.

Now, what of other strategies like dividing up the relation into 1GB
chunks and reading each one in a separate process?  We could certainly
DO that, but what advantage does it have over this?  The only benefit
I can see is that you avoid accessing a data structure of the type
shown above for every block, but that only matters if that cost is
material, and I tend to think it won't be.  On the flip side, it means
that the granularity for dividing up work between processes is now
very coarse - when there are less than 6GB of data left in a relation,
at most 6 processes can work on it.  That might be OK if the data is
being read in from disk anyway, but it's certainly not the best we can
do when the data is in memory.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Fri, Jan 16, 2015 at 11:27 PM, Amit Kapila <amit.kapila16@gmail.com> wrote:
> On Fri, Jan 16, 2015 at 11:49 PM, Robert Haas <robertmhaas@gmail.com> wrote:
>> As mentioned downthread, a far bigger consideration is the I/O pattern
>> we create.  A sequential scan is so-called because it reads the
>> relation sequentially.  If we destroy that property, we will be more
>> than slightly sad.  It might be OK to do sequential scans of, say,
>> each 1GB segment separately, but I'm pretty sure it would be a real
>> bad idea to read 8kB at a time at blocks 0, 64, 128, 1, 65, 129, ...
>>
>> What I'm thinking about is that we might have something like this:
>>
>> struct this_lives_in_dynamic_shared_memory
>> {
>>     BlockNumber last_block;
>>     Size prefetch_distance;
>>     Size prefetch_increment;
>>     slock_t mutex;
>>     BlockNumber next_prefetch_block;
>>     BlockNumber next_scan_block;
>> };
>>
>> Each worker takes the mutex and checks whether next_prefetch_block -
>> next_scan_block < prefetch_distance and also whether
>> next_prefetch_block < last_block.  If both are true, it prefetches
>> some number of additional blocks, as specified by prefetch_increment.
>> Otherwise, it increments next_scan_block and scans the block
>> corresponding to the old value.
>
> Assuming we will increment next_prefetch_block only after prefetching
> blocks (equivalent to prefetch_increment), won't 2 workers can
> simultaneously see the same value for next_prefetch_block and try to
> perform prefetch for same blocks?

The idea is that you can only examine and modify next_prefetch_block
or next_scan_block while holding the mutex.

> What will be value of prefetch_increment?
> Will it be equal to prefetch_distance or prefetch_distance/2 or
> prefetch_distance/4 or .. or will it be totally unrelated to
> prefetch_distance?

I dunno, that might take some experimentation.  prefetch_distance/2
doesn't sound stupid.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Mon, Jan 19, 2015 at 2:24 AM, Amit Kapila <amit.kapila16@gmail.com> wrote:
> Okay, I think I got the idea what you want to achieve via
> prefetching.  So assuming prefetch_distance = 100 and
> prefetch_increment = 50 (prefetch_distance /2), it seems to me
> that as soon as there are less than 100 blocks in prefetch quota,
> it will fetch next 50 blocks which means the system will be always
> approximately 50 blocks ahead, that will ensure that in this algorithm
> it will always perform sequential scan, however eventually this is turning
> to be a system where one worker is reading from disk and then other
> workers are reading from OS buffers to shared buffers and then getting
> the tuple.  In this approach only one downside I can see and that is
> there could be times during execution where some/all workers will have
> to wait on the worker doing prefetching, however I think we should try
> this approach and see how it works.

Right.  We probably want to make prefetch_distance a GUC.  After all,
we currently rely on the operating system for prefetching, and the
operating system has a setting for this, at least on Linux (blockdev
--getra).  It's possible, however, that we don't need this at all,
because the OS might be smart enough to figure it out for us.  It's
probably worth testing, though.

> Another thing is that I think prefetching is not supported on all platforms
> (Windows) and for such systems as per above algorithm we need to
> rely on block-by-block method.

Well, I think we should try to set up a test to see if this is hurting
us.  First, do a sequential-scan of a related too big at least twice
as large as RAM.  Then, do a parallel sequential scan of the same
relation with 2 workers.  Repeat these in alternation several times.
If the operating system is accomplishing meaningful readahead, and the
parallel sequential scan is breaking it, then since the test is
I/O-bound I would expect to see the parallel scan actually being
slower than the normal way.

Or perhaps there is some other test that would be better (ideas
welcome) but the point is we may need something like this, but we
should try to figure out whether we need it before spending too much
time on it.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On 1/19/15 7:20 AM, Robert Haas wrote:
>> >Another thing is that I think prefetching is not supported on all platforms
>> >(Windows) and for such systems as per above algorithm we need to
>> >rely on block-by-block method.
> Well, I think we should try to set up a test to see if this is hurting
> us.  First, do a sequential-scan of a related too big at least twice
> as large as RAM.  Then, do a parallel sequential scan of the same
> relation with 2 workers.  Repeat these in alternation several times.
> If the operating system is accomplishing meaningful readahead, and the
> parallel sequential scan is breaking it, then since the test is
> I/O-bound I would expect to see the parallel scan actually being
> slower than the normal way.
>
> Or perhaps there is some other test that would be better (ideas
> welcome) but the point is we may need something like this, but we
> should try to figure out whether we need it before spending too much
> time on it.

I'm guessing that not all supported platforms have prefetching that actually helps us... but it would be good to
actuallyknow if that's the case.
 

Where I think this gets a lot more interesting is if we could apply this to an index scan. My thought is that would
resultin one worker mostly being responsible for advancing the index scan itself while the other workers were issuing
(andwaiting on) heap IO. So even if this doesn't turn out to be a win for seqscan, there's other places we might well
wantto use it.
 
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

On 20-01-2015 PM 11:29, Amit Kapila wrote:
> 
> I have taken care of integrating the parallel sequence scan with the
> latest patch posted (parallel-mode-v1.patch) by Robert at below
> location:
> http://www.postgresql.org/message-id/CA+TgmoZdUK4K3XHBxc9vM-82khourEZdvQWTfgLhWsd2R2aAGQ@mail.gmail.com
> 
> Changes in this version
> -----------------------------------------------
> 1. As mentioned previously, I have exposed one parameter
> ParallelWorkerNumber as used in parallel-mode patch.
> 2. Enabled tuple queue to be used for passing tuples from
> worker backend to master backend along with error queue
> as per suggestion by Robert in the mail above.
> 3. Involved master backend to scan the heap directly when
> tuples are not available in any shared memory tuple queue.
> 4. Introduced 3 new parameters (cpu_tuple_comm_cost,
> parallel_setup_cost, parallel_startup_cost) for deciding the cost
> of parallel plan.  Currently, I have kept the default values for
> parallel_setup_cost and parallel_startup_cost as 0.0, as those
> require some experiments.
> 5. Fixed some issues (related to memory increase as reported
> upthread by Thom Brown and general feature issues found during
> test)
> 
> Note - I have yet to handle the new node types introduced at some
> of the places and need to verify prepared queries and some other
> things, however I think it will be good if I can get some feedback
> at current stage.
> 

I got an assertion failure:

In src/backend/executor/execTuples.c: ExecStoreTuple()

/* passing shouldFree=true for a tuple on a disk page is not sane */
Assert(BufferIsValid(buffer) ? (!shouldFree) : true);

when called from:

In src/backend/executor/nodeParallelSeqscan.c: ParallelSeqNext()

I think something like the following would be necessary (reading from
comments in the code):

--- a/src/backend/executor/nodeParallelSeqscan.c
+++ b/src/backend/executor/nodeParallelSeqscan.c
@@ -85,7 +85,7 @@ ParallelSeqNext(ParallelSeqScanState *node)       if (tuple)           ExecStoreTuple(tuple,
               slot,
 
-                          scandesc->rs_cbuf,
+                          fromheap ? scandesc->rs_cbuf : InvalidBuffer,                          !fromheap);

After fixing this, the assertion failure seems to be gone though I
observed the blocked (CPU maxed out) state as reported elsewhere by Thom
Brown.

What I was doing:

CREATE TABLE test(a) AS SELECT generate_series(1, 10000000);

postgres=# SHOW max_worker_processes;max_worker_processes
----------------------8
(1 row)

postgres=# SET seq_page_cost TO 100;
SET

postgres=# SET parallel_seqscan_degree TO 4;
SET

postgres=# EXPLAIN SELECT * FROM test;                              QUERY PLAN
-------------------------------------------------------------------------Parallel Seq Scan on test
(cost=0.00..1801071.27rows=8981483 width=4)  Number of Workers: 4  Number of Blocks Per Worker: 8849
 
(3 rows)

Though, EXPLAIN ANALYZE caused the thing.

Thanks,
Amit

On Wed, Jan 21, 2015 at 12:47 PM, Amit Langote <Langote_Amit_f8@lab.ntt.co.jp> wrote:
>
> On 20-01-2015 PM 11:29, Amit Kapila wrote:
> > Note - I have yet to handle the new node types introduced at some
> > of the places and need to verify prepared queries and some other
> > things, however I think it will be good if I can get some feedback
> > at current stage.
> >
>
> I got an assertion failure:
>
> In src/backend/executor/execTuples.c: ExecStoreTuple()
>
> /* passing shouldFree=true for a tuple on a disk page is not sane */
> Assert(BufferIsValid(buffer) ? (!shouldFree) : true);
>

Good Catch!

The reason is that while master backend is scanning from a heap

page, if it finds another tuple/tuples's from shared memory message

queue it will process those tuples first and in such a scenario, the scan

descriptor will still have reference to buffer which it is using from scanning

from heap.  Your proposed fix will work.

> After fixing this, the assertion failure seems to be gone though I
> observed the blocked (CPU maxed out) state as reported elsewhere by Thom
> Brown.
>

Does it happen only when parallel_seqscan_degree > max_worker_processes?

> On Wed, Jan 21, 2015 at 4:31 PM, Amit Langote <amitlangote09@gmail.com>
> wrote:
> > On Wednesday, January 21, 2015, Amit Kapila <amit.kapila16@gmail.com>
> wrote:
> >>
> >>
> >> Does it happen only when parallel_seqscan_degree > max_worker_processes?
> >
> >
> >  I have max_worker_processes set to the default of 8 while
> parallel_seqscan_degree is 4. So, this may be a case different from Thom's.
> >
> 
> I think this is due to reason that memory for forming tuple in master backend
> is retained for longer time which is causing this statement to take much
> longer time than required.  I have fixed the other issue as well reported
> by you in attached patch.
> 
> I think this patch is still not completely ready for general purpose testing,
> however it could be helpful if we can run some tests to see in what kind
> of scenario's it gives benefit like in the test you are doing if rather
> than increasing seq_page_cost, you should add an expensive WHERE condition
> so that it should automatically select parallel plan. I think it is better
> to change one of the new parameter's (parallel_setup_cost,
> parallel_startup_cost and cpu_tuple_comm_cost) if you want your statement
> to use parallel plan, like in your example if you would have reduced
> cpu_tuple_comm_cost, it would have selected parallel plan, that way we can
> get some feedback about what should be the appropriate default values for
> the newly added parameters.  I am already planing to do some tests in that
> regard, however if I get some feedback from other's that would be helpful.
> 
(Please point out me if my understanding is incorrect.)

What happen if dynamic background worker process tries to reference temporary
tables? Because buffer of temporary table blocks are allocated on private
address space, its recent status is not visible to other process unless it is
not flushed to the storage every time.

Do we need to prohibit create_parallelscan_paths() to generate a path when
target relation is temporary one?

Thanks,
--
NEC OSS Promotion Center / PG-Strom Project
KaiGai Kohei <kaigai@ak.jp.nec.com>

On 21-01-2015 PM 09:43, Amit Kapila wrote:
> On Wed, Jan 21, 2015 at 4:31 PM, Amit Langote <amitlangote09@gmail.com>
> wrote:
>> On Wednesday, January 21, 2015, Amit Kapila <amit.kapila16@gmail.com>
> wrote:
>>>
>>>
>>> Does it happen only when parallel_seqscan_degree > max_worker_processes?
>>
>>
>>  I have max_worker_processes set to the default of 8 while
> parallel_seqscan_degree is 4. So, this may be a case different from Thom's.
>>
> 
> I think this is due to reason that memory for forming
> tuple in master backend is retained for longer time which
> is causing this statement to take much longer time than
> required.  I have fixed the other issue as well reported by
> you in attached patch.
> 

Thanks for fixing.

> I think this patch is still not completely ready for general
> purpose testing, however it could be helpful if we can run
> some tests to see in what kind of scenario's it gives benefit
> like in the test you are doing if rather than increasing
> seq_page_cost, you should add an expensive WHERE condition
> so that it should automatically select parallel plan. I think it is better
> to change one of the new parameter's (parallel_setup_cost,
> parallel_startup_cost and cpu_tuple_comm_cost) if you want
> your statement to use parallel plan, like in your example if
> you would have reduced cpu_tuple_comm_cost, it would have
> selected parallel plan, that way we can get some feedback about
> what should be the appropriate default values for the newly added
> parameters.  I am already planing to do some tests in that regard,
> however if I get some feedback from other's that would be helpful.
> 
> 

Perhaps you are aware or you've postponed working on it, but I see that
a plan executing in a worker does not know about instrumentation. It
results in the EXPLAIN ANALYZE showing incorrect figures. For example
compare the normal seqscan and parallel seqscan below:

postgres=# EXPLAIN ANALYZE SELECT * FROM test WHERE sqrt(a) < 3456 AND
md5(a::text) LIKE 'ac%';                                                 QUERY PLAN

---------------------------------------------------------------------------------------------------------------Seq Scan
ontest  (cost=0.00..310228.52 rows=16120 width=4) (actual
 
time=0.497..17062.436 rows=39028 loops=1)  Filter: ((sqrt((a)::double precision) < 3456::double precision) AND
(md5((a)::text) ~~ 'ac%'::text))  Rows Removed by Filter: 9960972Planning time: 0.206 msExecution time: 17378.413 ms
(5 rows)

postgres=# EXPLAIN ANALYZE SELECT * FROM test WHERE sqrt(a) < 3456 AND
md5(a::text) LIKE 'ac%';                                                     QUERY PLAN


-----------------------------------------------------------------------------------------------------------------------Parallel
SeqScan on test  (cost=0.00..255486.08 rows=16120 width=4)
 
(actual time=7.329..4906.981 rows=39028 loops=1)  Filter: ((sqrt((a)::double precision) < 3456::double precision) AND
(md5((a)::text) ~~ 'ac%'::text))  Rows Removed by Filter: 1992710  Number of Workers: 4  Number of Blocks Per Worker:
8849Planningtime: 0.137 msExecution time: 6077.782 ms
 
(7 rows)

Note the "Rows Removed by Filter". I guess the difference may be
because, all the rows filtered by workers were not accounted for. I'm
not quite sure, but since exec_worker_stmt goes the Portal way,
QueryDesc.instrument_options remains unset and hence no instrumentation
opportunities in a worker backend. One option may be to pass
instrument_options down to worker_stmt?

By the way, 17s and 6s compare really well in favor of parallel seqscan
above, :)

Thanks,
Amit

On 22-01-2015 PM 02:30, Amit Kapila wrote:
>> Perhaps you are aware or you've postponed working on it, but I see that
>> a plan executing in a worker does not know about instrumentation.
> 
> I have deferred it until other main parts are stabilised/reviewed.  Once
> that is done, we can take a call what is best we can do for instrumentation.
> Thom has reported the same as well upthread.
> 

Ah, I missed Thom's report.

>> Note the "Rows Removed by Filter". I guess the difference may be
>> because, all the rows filtered by workers were not accounted for. I'm
>> not quite sure, but since exec_worker_stmt goes the Portal way,
>> QueryDesc.instrument_options remains unset and hence no instrumentation
>> opportunities in a worker backend. One option may be to pass
>> instrument_options down to worker_stmt?
>>
> 
> I think there is more to it, master backend need to process that information
> as well.
> 

I see.

Thanks,
Amit

On Thu, Jan 22, 2015 at 5:57 AM, Amit Kapila <amit.kapila16@gmail.com> wrote:
> 1. Scanning block-by-block has negative impact on performance and
> I thin it will degrade more if we increase parallel count as that can lead
> to more randomness.
>
> 2. Scanning in fixed chunks improves the performance. Increasing
> parallel count to a very large number might impact the performance,
> but I think we can have a lower bound below which we will not allow
> multiple processes to scan the relation.

I'm confused.  Your actual test numbers seem to show that the
performance with the block-by-block approach was slightly higher with
parallelism than without, where as the performance with the
chunk-by-chunk approach was lower with parallelism than without, but
the text quoted above, summarizing those numbers, says the opposite.

Also, I think testing with 2 workers is probably not enough.  I think
we should test with 8 or even 16.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Thu, Jan 22, 2015 at 9:02 AM, Amit Kapila <amit.kapila16@gmail.com> wrote:
>> I'm confused.  Your actual test numbers seem to show that the
>> performance with the block-by-block approach was slightly higher with
>> parallelism than without, where as the performance with the
>> chunk-by-chunk approach was lower with parallelism than without, but
>> the text quoted above, summarizing those numbers, says the opposite.
>
> Sorry for causing confusion, I should have been more explicit about
> explaining the numbers.  Let me try again,
> Values in columns is time in milliseconds to complete the execution,
> so higher means it took more time.  If you see in block-by-block, the
> time taken to complete the execution with 2 workers is more than
> no workers which means parallelism has degraded the performance.

*facepalm*

Oh, yeah, right.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On 01/22/2015 05:53 AM, Robert Haas wrote:
> Also, I think testing with 2 workers is probably not enough.  I think
> we should test with 8 or even 16.

FWIW, based on my experience there will also be demand to use parallel
query using 4 workers, particularly on AWS.

-- 
Josh Berkus
PostgreSQL Experts Inc.
http://pgexperts.com

On 1/23/15 5:42 AM, Amit Kapila wrote:
> *Fixed-Chunks*     
> *No. of workers/Time (ms)*>         *0*     *2*     *4*     *8*     *16*     *24*     *32*
> Run-1     250536     266279     251263     234347     87930     50474     35474
> Run-2     249587     230628     225648     193340     83036     35140     9100
> Run-3     234963     220671     230002     256183     105382     62493     27903
> Run-4     239111     245448     224057     189196     123780     63794     24746
> Run-5     239937     222820     219025     220478     114007     77965     39766
>
>
>
> The trend remains same although there is some variation.
> In block-by-block approach, it performance dips (execution takes
> more time) with more number of workers, though it stabilizes at
> some higher value, still I feel it is random as it leads to random
> scan.
> In Fixed-chunk approach, the performance improves with more
> number of workers especially at slightly higher worker count.

Those fixed chunk numbers look pretty screwy. 2, 4 and 8 workers make no difference, then suddenly 16 cuts times by 1/2
to1/3? Then 32 cuts time by another 1/2 to 1/3?
 
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

On 01/23/2015 10:44 AM, Jim Nasby wrote:
> number of workers especially at slightly higher worker count.
>
> Those fixed chunk numbers look pretty screwy. 2, 4 and 8 workers make no
> difference, then suddenly 16 cuts times by 1/2 to 1/3? Then 32 cuts time
> by another 1/2 to 1/3?

cached? First couple of runs gets the relations into memory?

JD



-- 
Command Prompt, Inc. - http://www.commandprompt.com/  503-667-4564
PostgreSQL Support, Training, Professional Services and Development
High Availability, Oracle Conversion, @cmdpromptinc
"If we send our children to Caesar for their education, we should             not be surprised when they come back as
Romans."

On 1/23/15 10:16 PM, Amit Kapila wrote:
> Further, if we want to just get the benefit of parallel I/O, then
> I think we can get that by parallelising partition scan where different
> table partitions reside on different disk partitions, however that is
> a matter of separate patch.

I don't think we even have to go that far.

My experience with Postgres is that it is *very* sensitive to IO latency (not bandwidth). I believe this is the case
becausecomplex queries tend to interleave CPU intensive code in-between IO requests. So we see this pattern:
 

Wait 5ms on IO
Compute for a few ms
Wait 5ms on IO
Compute for a few ms
...

We blindly assume that the kernel will magically do read-ahead for us, but I've never seen that work so great. It
certainlyfalls apart on something like an index scan.
 

If we could instead do this:

Wait for first IO, issue second IO request
Compute
Already have second IO request, issue third
...

We'd be a lot less sensitive to IO latency.

I wonder what kind of gains we would see if every SeqScan in a query spawned a worker just to read tuples and shove
themin a queue (or shove a pointer to a buffer in the queue). Similarly, have IndexScans have one worker reading the
indexand another worker taking index tuples and reading heap tuples...
 
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

Jim Nasby <Jim.Nasby@BlueTreble.com> writes:
> On 1/23/15 10:16 PM, Amit Kapila wrote:
>> Further, if we want to just get the benefit of parallel I/O, then
>> I think we can get that by parallelising partition scan where different
>> table partitions reside on different disk partitions, however that is
>> a matter of separate patch.

> I don't think we even have to go that far.

> My experience with Postgres is that it is *very* sensitive to IO latency (not bandwidth). I believe this is the case
becausecomplex queries tend to interleave CPU intensive code in-between IO requests. So we see this pattern:
 

> Wait 5ms on IO
> Compute for a few ms
> Wait 5ms on IO
> Compute for a few ms
> ...

> We blindly assume that the kernel will magically do read-ahead for us, but I've never seen that work so great. It
certainlyfalls apart on something like an index scan.
 

> If we could instead do this:

> Wait for first IO, issue second IO request
> Compute
> Already have second IO request, issue third
> ...

> We'd be a lot less sensitive to IO latency.

It would take about five minutes of coding to prove or disprove this:
stick a PrefetchBuffer call into heapgetpage() to launch a request for the
next page as soon as we've read the current one, and then see if that
makes any obvious performance difference.  I'm not convinced that it will,
but if it did then we could think about how to make it work for real.
        regards, tom lane

Hi PG devs!

Tom Lane <tgl@sss.pgh.pa.us> writes:

>> Wait for first IO, issue second IO request
>> Compute
>> Already have second IO request, issue third
>> ...
>
>> We'd be a lot less sensitive to IO latency.
>
> It would take about five minutes of coding to prove or disprove this:
> stick a PrefetchBuffer call into heapgetpage() to launch a request for the
> next page as soon as we've read the current one, and then see if that
> makes any obvious performance difference.  I'm not convinced that it will,
> but if it did then we could think about how to make it work for real.

Sorry for dropping in so late...

I have done all this two years ago.  For TPC-H Q8, Q9, Q17, Q20, and Q21
I see a speedup of ~100% when using IndexScan prefetching + Nested-Loops
Look-Ahead (the outer loop!).
(On SSD with 32 Pages Prefetch/Look-Ahead + Cold Page Cache / Small RAM)

Regards,
Daniel
-- 
MSc. Daniel Bausch
Research Assistant (Computer Science)
Technische Universität Darmstadt
http://www.dvs.tu-darmstadt.de/staff/dbausch

On Tue, Jan 27, 2015 at 08:02:37AM +0100, Daniel Bausch wrote:
> Hi PG devs!
> 
> Tom Lane <tgl@sss.pgh.pa.us> writes:
> 
> >> Wait for first IO, issue second IO request
> >> Compute
> >> Already have second IO request, issue third
> >> ...
> >
> >> We'd be a lot less sensitive to IO latency.
> >
> > It would take about five minutes of coding to prove or disprove this:
> > stick a PrefetchBuffer call into heapgetpage() to launch a request for the
> > next page as soon as we've read the current one, and then see if that
> > makes any obvious performance difference.  I'm not convinced that it will,
> > but if it did then we could think about how to make it work for real.
> 
> Sorry for dropping in so late...
> 
> I have done all this two years ago.  For TPC-H Q8, Q9, Q17, Q20, and Q21
> I see a speedup of ~100% when using IndexScan prefetching + Nested-Loops
> Look-Ahead (the outer loop!).
> (On SSD with 32 Pages Prefetch/Look-Ahead + Cold Page Cache / Small RAM)

Would you be so kind as to pass along any patches (ideally applicable
to git master), tests, and specific measurements you made?

Cheers,
David.
-- 
David Fetter <david@fetter.org> http://fetter.org/
Phone: +1 415 235 3778  AIM: dfetter666  Yahoo!: dfetter
Skype: davidfetter      XMPP: david.fetter@gmail.com

Remember to vote!
Consider donating to Postgres: http://www.postgresql.org/about/donate

On Thu, Jan 22, 2015 at 5:57 AM, Amit Kapila <amit.kapila16@gmail.com> wrote:
> Script used to test is attached (parallel_count.sh)

Why does this use EXPLAIN ANALYZE instead of \timing ?

> IBM POWER-7 16 cores, 64 hardware threads
> RAM = 64GB
>
> Table Size - 120GB
>
> Used below statements to create table -
> create table tbl_perf(c1 int, c2 char(1000));
> insert into tbl_perf values(generate_series(1,10000000),'aaaaa');
> insert into tbl_perf values(generate_series(10000001,30000000),'aaaaa');
> insert into tbl_perf values(generate_series(30000001,110000000),'aaaaa');

I generated this table using this same method and experimented with
copying the whole file to the bit bucket using dd.  I did this on
hydra, which I think is the same machine you used.

time for i in `seq 0 119`; do if [ $i -eq 0 ]; then f=16388; else
f=16388.$i; fi; dd if=$f of=/dev/null bs=8k; done

There is a considerable amount of variation in the amount of time this
takes to run based on how much of the relation is cached.  Clearly,
there's no way for the system to cache it all, but it can cache a
significant portion, and that affects the results to no small degree.
dd on hydra prints information on the data transfer rate; on uncached
1GB segments, it runs at right around 400 MB/s, but that can soar to
upwards of 3GB/s when the relation is fully cached.  I tried flushing
the OS cache via echo 1 > /proc/sys/vm/drop_caches, and found that
immediately after doing that, the above command took 5m21s to run -
i.e. ~321000 ms.  Most of your test times are faster than that, which
means they reflect some degree of caching.  When I immediately reran
the command a second time, it finished in 4m18s the second time, or
~258000 ms.  The rate was the same as the first test - about 400 MB/s
- for most of the files, but 27 of the last 28 files went much faster,
between 1.3 GB/s and 3.7 GB/s.

This tells us that the OS cache on this machine has anti-spoliation
logic in it, probably not dissimilar to what we have in PG.  If the
data were cycled through the system cache in strict LRU fashion, any
data that was leftover from the first run would have been flushed out
by the early part of the second run, so that all the results from the
second set of runs would have hit the disk.  But in fact, that's not
what happened: the last pages from the first run remained cached even
after reading an amount of new data that exceeds the size of RAM on
that machine.  What I think this demonstrates is that we're going to
have to be very careful to control for caching effects, or we may find
that we get misleading results.  To make this simpler, I've installed
a setuid binary /usr/bin/drop_caches that you (or anyone who has an
account on that machine) can use you drop the caches; run 'drop_caches
1'.

> Block-By-Block
>
> No. of workers/Time (ms) 0 2
> Run-1 267798 295051
> Run-2 276646 296665
> Run-3 281364 314952
> Run-4 290231 326243
> Run-5 288890 295684

The next thing I did was run test with the block-by-block method after
having dropped the caches.  I did this with 0 workers and with 8
workers.  I dropped the caches and restarted postgres before each
test, but then ran each test a second time to see the effect of
caching by both the OS and by PostgreSQL.  I got these results:

With 0 workers, first run took 883465.352 ms, and second run took 295050.106 ms.
With 8 workers, first run took 340302.250 ms, and second run took 307767.758 ms.

This is a confusing result, because you expect parallelism to help
more when the relation is partly cached, and make little or no
difference when it isn't cached.  But that's not what happened.

I've also got a draft of a prefetching implementation here that I'd
like to test out, but I've just discovered that it's buggy, so I'm
going to send these results for now and work on fixing that.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

Robert, all,

* Robert Haas (robertmhaas@gmail.com) wrote:
> There is a considerable amount of variation in the amount of time this
> takes to run based on how much of the relation is cached.  Clearly,
> there's no way for the system to cache it all, but it can cache a
> significant portion, and that affects the results to no small degree.
> dd on hydra prints information on the data transfer rate; on uncached
> 1GB segments, it runs at right around 400 MB/s, but that can soar to
> upwards of 3GB/s when the relation is fully cached.  I tried flushing
> the OS cache via echo 1 > /proc/sys/vm/drop_caches, and found that
> immediately after doing that, the above command took 5m21s to run -
> i.e. ~321000 ms.  Most of your test times are faster than that, which
> means they reflect some degree of caching.  When I immediately reran
> the command a second time, it finished in 4m18s the second time, or
> ~258000 ms.  The rate was the same as the first test - about 400 MB/s
> - for most of the files, but 27 of the last 28 files went much faster,
> between 1.3 GB/s and 3.7 GB/s.

[...]

> With 0 workers, first run took 883465.352 ms, and second run took 295050.106 ms.
> With 8 workers, first run took 340302.250 ms, and second run took 307767.758 ms.
>
> This is a confusing result, because you expect parallelism to help
> more when the relation is partly cached, and make little or no
> difference when it isn't cached.  But that's not what happened.

These numbers seem to indicate that the oddball is the single-threaded
uncached run.  If I followed correctly, the uncached 'dd' took 321s,
which is relatively close to the uncached-lots-of-workers and the two
cached runs.  What in the world is the uncached single-thread case doing
that it takes an extra 543s, or over twice as long?  It's clearly not
disk i/o which is causing the slowdown, based on your dd tests.

One possibility might be round-trip latency.  The multi-threaded case is
able to keep the CPUs and the i/o system going, and the cached results
don't have as much latency since things are cached, but the
single-threaded uncached case going i/o -> cpu -> i/o -> cpu, ends up
with a lot of wait time as it switches between being on CPU and waiting
on the i/o.

Just some thoughts.
Thanks,
    Stephen

On Fri, Jan 23, 2015 at 6:42 AM, Amit Kapila <amit.kapila16@gmail.com> wrote:
> Fixed-Chunks
>
> No. of workers/Time (ms) 0 2 4 8 16 24 32
> Run-1 250536 266279 251263 234347 87930 50474 35474
> Run-2 249587 230628 225648 193340 83036 35140 9100
> Run-3 234963 220671 230002 256183 105382 62493 27903
> Run-4 239111 245448 224057 189196 123780 63794 24746
> Run-5 239937 222820 219025 220478 114007 77965 39766

I cannot reproduce these results.  I applied your fixed-chunk size
patch and ran SELECT parallel_count('tbl_perf', 32) a few times.  The
first thing I notice is that, as I predicted, there's an issue with
different workers finishing at different times.  For example, from my
first run:

2015-01-27 22:13:09 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34700) exited with exit code 0
2015-01-27 22:13:09 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34698) exited with exit code 0
2015-01-27 22:13:09 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34701) exited with exit code 0
2015-01-27 22:13:10 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34699) exited with exit code 0
2015-01-27 22:15:00 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34683) exited with exit code 0
2015-01-27 22:15:29 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34673) exited with exit code 0
2015-01-27 22:15:58 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34679) exited with exit code 0
2015-01-27 22:16:38 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34689) exited with exit code 0
2015-01-27 22:16:39 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34671) exited with exit code 0
2015-01-27 22:16:47 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34677) exited with exit code 0
2015-01-27 22:16:47 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34672) exited with exit code 0
2015-01-27 22:16:48 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34680) exited with exit code 0
2015-01-27 22:16:50 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34686) exited with exit code 0
2015-01-27 22:16:51 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34670) exited with exit code 0
2015-01-27 22:16:51 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34690) exited with exit code 0
2015-01-27 22:16:51 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34674) exited with exit code 0
2015-01-27 22:16:52 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34684) exited with exit code 0
2015-01-27 22:16:53 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34675) exited with exit code 0
2015-01-27 22:16:53 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34682) exited with exit code 0
2015-01-27 22:16:53 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34691) exited with exit code 0
2015-01-27 22:16:54 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34676) exited with exit code 0
2015-01-27 22:16:54 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34685) exited with exit code 0
2015-01-27 22:16:55 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34692) exited with exit code 0
2015-01-27 22:16:56 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34687) exited with exit code 0
2015-01-27 22:16:56 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34678) exited with exit code 0
2015-01-27 22:16:57 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34681) exited with exit code 0
2015-01-27 22:16:57 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34688) exited with exit code 0
2015-01-27 22:16:59 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34694) exited with exit code 0
2015-01-27 22:16:59 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34693) exited with exit code 0
2015-01-27 22:17:02 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34695) exited with exit code 0
2015-01-27 22:17:02 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34697) exited with exit code 0
2015-01-27 22:17:02 UTC [34660] LOG:  worker process: parallel worker
for PID 34668 (PID 34696) exited with exit code 0

That run started at 22:13:01.  Within 4 seconds, 4 workers exited.  So
clearly we are not getting the promised 32-way parallelism for the
whole test.  Granted, in this instance, *most* of the workers run
until the end, but I think we'll find that there are
uncomfortably-frequent cases where we get significantly less
parallelism than we planned on because the work isn't divided evenly.

But leaving that aside, I've run this test 6 times in a row now, with
a warm cache, and the best time I have is 237310.042 ms and the worst
time I have is 242936.315 ms.  So there's very little variation, and
it's reasonably close to the results I got with dd, suggesting that
the system is fairly well I/O bound.  At a sequential read speed of
400 MB/s, 240 s = 96 GB of data.  Assuming it takes no time at all to
process the cached data (which seems to be not far from wrong judging
by how quickly the first few workers exit), that means we're getting
24 GB of data from cache on a 64 GB machine.  That seems a little low,
but if the kernel is refusing to cache the whole relation to avoid
cache-trashing, it could be right.

Now, when you did what I understand to be the same test on the same
machine, you got times ranging from 9.1 seconds to 35.4 seconds.
Clearly, there is some difference between our test setups.  Moreover,
I'm kind of suspicious about whether your results are actually
physically possible.  Even in the best case where you somehow had the
maximum possible amount of data - 64 GB on a 64 GB machine - cached,
leaving no space for cache duplication between PG and the OS and no
space for the operating system or postgres itself - the table is 120
GB, so you've got to read *at least* 56 GB from disk.  Reading 56 GB
from disk in 9 seconds represents an I/O rate of >6 GB/s. I grant that
there could be some speedup from issuing I/O requests in parallel
instead of serially, but that is a 15x speedup over dd, so I am a
little suspicious that there is some problem with the test setup,
especially because I cannot reproduce the results.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On 1/26/15 11:11 PM, Amit Kapila wrote:
> On Tue, Jan 27, 2015 at 3:18 AM, Jim Nasby <Jim.Nasby@bluetreble.com <mailto:Jim.Nasby@bluetreble.com>> wrote:
>  >
>  > On 1/23/15 10:16 PM, Amit Kapila wrote:
>  >>
>  >> Further, if we want to just get the benefit of parallel I/O, then
>  >> I think we can get that by parallelising partition scan where different
>  >> table partitions reside on different disk partitions, however that is
>  >> a matter of separate patch.
>  >
>  >
>  > I don't think we even have to go that far.
>  >
>  >
>  > We'd be a lot less sensitive to IO latency.
>  >
>  > I wonder what kind of gains we would see if every SeqScan in a query spawned a worker just to read tuples and
shovethem in a queue (or shove a pointer to a buffer in the queue).
 
>  >
>
> Here IIUC, you want to say that just get the read done by one parallel
> worker and then all expression calculation (evaluation of qualification
> and target list) in the main backend, it seems to me that by doing it
> that way, the benefit of parallelisation will be lost due to tuple
> communication overhead (may be the overhead is less if we just
> pass a pointer to buffer but that will have another kind of problems
> like holding buffer pins for a longer period of time).
>
> I could see the advantage of testing on lines as suggested by Tom Lane,
> but that seems to be not directly related to what we want to achieve by
> this patch (parallel seq scan) or if you think otherwise then let me know?

There's some low-hanging fruit when it comes to improving our IO performance (or more specifically, decreasing our
sensitivityto IO latency). Perhaps the way to do that is with the parallel infrastructure, perhaps not. But I think
it'spremature to look at parallelism for increasing IO performance, or worrying about things like how many IO threads
weshould have before we at least look at simpler things we could do. We shouldn't assume there's nothing to be gained
shortof a full parallelization implementation.
 

That's not to say there's nothing else we could use parallelism for. Sort, merge and hash operations come to mind.
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

On 1/27/15 3:46 PM, Stephen Frost wrote:
>> With 0 workers, first run took 883465.352 ms, and second run took 295050.106 ms.
>> >With 8 workers, first run took 340302.250 ms, and second run took 307767.758 ms.
>> >
>> >This is a confusing result, because you expect parallelism to help
>> >more when the relation is partly cached, and make little or no
>> >difference when it isn't cached.  But that's not what happened.
> These numbers seem to indicate that the oddball is the single-threaded
> uncached run.  If I followed correctly, the uncached 'dd' took 321s,
> which is relatively close to the uncached-lots-of-workers and the two
> cached runs.  What in the world is the uncached single-thread case doing
> that it takes an extra 543s, or over twice as long?  It's clearly not
> disk i/o which is causing the slowdown, based on your dd tests.
>
> One possibility might be round-trip latency.  The multi-threaded case is
> able to keep the CPUs and the i/o system going, and the cached results
> don't have as much latency since things are cached, but the
> single-threaded uncached case going i/o -> cpu -> i/o -> cpu, ends up
> with a lot of wait time as it switches between being on CPU and waiting
> on the i/o.

This exactly mirrors what I've seen on production systems. On a single SeqScan I can't get anywhere close to the IO
performanceI could get with dd. Once I got up to 4-8 SeqScans of different tables running together, I saw iostat
numbersthat were similar to what a single dd bs=8k would do. I've tested this with iSCSI SAN volumes on both 1Gbit and
10Gbitethernet.
 

This is why I think that when it comes to IO performance, before we start worrying about real parallelization we should
investigateways to do some kind of async IO.
 

I only have my SSD laptop and a really old server to test on, but I'll try Tom's suggestion of adding a PrefetchBuffer
callinto heapgetpage() unless someone beats me to it. I should be able to do it tomorrow.
 
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

On Tue, Jan 27, 2015 at 4:46 PM, Stephen Frost <sfrost@snowman.net> wrote:
>> With 0 workers, first run took 883465.352 ms, and second run took 295050.106 ms.
>> With 8 workers, first run took 340302.250 ms, and second run took 307767.758 ms.
>>
>> This is a confusing result, because you expect parallelism to help
>> more when the relation is partly cached, and make little or no
>> difference when it isn't cached.  But that's not what happened.
>
> These numbers seem to indicate that the oddball is the single-threaded
> uncached run.  If I followed correctly, the uncached 'dd' took 321s,
> which is relatively close to the uncached-lots-of-workers and the two
> cached runs.  What in the world is the uncached single-thread case doing
> that it takes an extra 543s, or over twice as long?  It's clearly not
> disk i/o which is causing the slowdown, based on your dd tests.

Yeah, I'm wondering if the disk just froze up on that run for a long
while, which has been known to occasionally happen on this machine,
because I can't reproduce that crappy number.  I did the 0-worker test
a few more times, with the block-by-block method, dropping the caches
and restarting PostgreSQL each time, and got:

322222.968 ms
322873.325 ms
322967.722 ms
321759.273 ms

After that last run, I ran it a few more times without restarting
PostgreSQL or dropping the caches, and got:

257629.348 ms
289668.976 ms
290342.970 ms
258035.226 ms
284237.729 ms

Then I redid the 8-client test.  Cold cache, I got 337312.554 ms.  On
the rerun, 323423.813 ms.  Third run, 324940.785.

There is more variability than I would like here.  Clearly, it goes a
bit faster when the cache is warm, but that's about all I can say with
any confidence.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Tue, Jan 27, 2015 at 6:00 PM, Robert Haas <robertmhaas@gmail.com> wrote:
> Now, when you did what I understand to be the same test on the same
> machine, you got times ranging from 9.1 seconds to 35.4 seconds.
> Clearly, there is some difference between our test setups.  Moreover,
> I'm kind of suspicious about whether your results are actually
> physically possible.  Even in the best case where you somehow had the
> maximum possible amount of data - 64 GB on a 64 GB machine - cached,
> leaving no space for cache duplication between PG and the OS and no
> space for the operating system or postgres itself - the table is 120
> GB, so you've got to read *at least* 56 GB from disk.  Reading 56 GB
> from disk in 9 seconds represents an I/O rate of >6 GB/s. I grant that
> there could be some speedup from issuing I/O requests in parallel
> instead of serially, but that is a 15x speedup over dd, so I am a
> little suspicious that there is some problem with the test setup,
> especially because I cannot reproduce the results.

So I thought about this a little more, and I realized after some
poking around that hydra's disk subsystem is actually six disks
configured in a software RAID5[1].  So one advantage of the
chunk-by-chunk approach you are proposing is that you might be able to
get all of the disks chugging away at once, because the data is
presumably striped across all of them.  Reading one block at a time,
you'll never have more than 1 or 2 disks going, but if you do
sequential reads from a bunch of different places in the relation, you
might manage to get all 6.  So that's something to think about.

One could imagine an algorithm like this: as long as there are more
1GB segments remaining than there are workers, each worker tries to
chug through a separate 1GB segment.  When there are not enough 1GB
segments remaining for that to work, then they start ganging up on the
same segments.  That way, you get the benefit of spreading out the I/O
across multiple files (and thus hopefully multiple members of the RAID
group) when the data is coming from disk, but you can still keep
everyone busy until the end, which will be important when the data is
all in-memory and you're just limited by CPU bandwidth.

All that aside, I still can't account for the numbers you are seeing.
When I run with your patch and what I think is your test case, I get
different (slower) numbers.  And even if we've got 6 drives cranking
along at 400MB/s each, that's still only 2.4 GB/s, not >6 GB/s.  So
I'm still perplexed.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

[1] Not my idea.

On 01/28/2015 04:16 AM, Robert Haas wrote:
> On Tue, Jan 27, 2015 at 6:00 PM, Robert Haas <robertmhaas@gmail.com> wrote:
>> Now, when you did what I understand to be the same test on the same
>> machine, you got times ranging from 9.1 seconds to 35.4 seconds.
>> Clearly, there is some difference between our test setups.  Moreover,
>> I'm kind of suspicious about whether your results are actually
>> physically possible.  Even in the best case where you somehow had the
>> maximum possible amount of data - 64 GB on a 64 GB machine - cached,
>> leaving no space for cache duplication between PG and the OS and no
>> space for the operating system or postgres itself - the table is 120
>> GB, so you've got to read *at least* 56 GB from disk.  Reading 56 GB
>> from disk in 9 seconds represents an I/O rate of >6 GB/s. I grant that
>> there could be some speedup from issuing I/O requests in parallel
>> instead of serially, but that is a 15x speedup over dd, so I am a
>> little suspicious that there is some problem with the test setup,
>> especially because I cannot reproduce the results.
>
> So I thought about this a little more, and I realized after some
> poking around that hydra's disk subsystem is actually six disks
> configured in a software RAID5[1].  So one advantage of the
> chunk-by-chunk approach you are proposing is that you might be able to
> get all of the disks chugging away at once, because the data is
> presumably striped across all of them.  Reading one block at a time,
> you'll never have more than 1 or 2 disks going, but if you do
> sequential reads from a bunch of different places in the relation, you
> might manage to get all 6.  So that's something to think about.
>
> One could imagine an algorithm like this: as long as there are more
> 1GB segments remaining than there are workers, each worker tries to
> chug through a separate 1GB segment.  When there are not enough 1GB
> segments remaining for that to work, then they start ganging up on the
> same segments.  That way, you get the benefit of spreading out the I/O
> across multiple files (and thus hopefully multiple members of the RAID
> group) when the data is coming from disk, but you can still keep
> everyone busy until the end, which will be important when the data is
> all in-memory and you're just limited by CPU bandwidth.

OTOH, spreading the I/O across multiple files is not a good thing, if 
you don't have a RAID setup like that. With a single spindle, you'll 
just induce more seeks.

Perhaps the OS is smart enough to read in large-enough chunks that the 
occasional seek doesn't hurt much. But then again, why isn't the OS 
smart enough to read in large-enough chunks to take advantage of the 
RAID even when you read just a single file?

- Heikki

On Wed, Jan 28, 2015 at 2:08 AM, Heikki Linnakangas
<hlinnakangas@vmware.com> wrote:
> OTOH, spreading the I/O across multiple files is not a good thing, if you
> don't have a RAID setup like that. With a single spindle, you'll just induce
> more seeks.
>
> Perhaps the OS is smart enough to read in large-enough chunks that the
> occasional seek doesn't hurt much. But then again, why isn't the OS smart
> enough to read in large-enough chunks to take advantage of the RAID even
> when you read just a single file?

Suppose we have N spindles and N worker processes and it just so
happens that the amount of computation is such that a each spindle can
keep one CPU busy.  Let's suppose the chunk size is 4MB.  If you read
from the relation at N staggered offsets, you might be lucky enough
that each one of them keeps a spindle busy, and you might be lucky
enough to have that stay true as the scans advance.  You don't need
any particularly large amount of read-ahead; you just need to stay at
least one block ahead of the CPU.  But if you read the relation in one
pass from beginning to end, you need at least N*4MB of read-ahead to
have data in cache for all N spindles, and the read-ahead will
certainly fail you at the end of every 1GB segment.

The problem here, as I see it, is that we're flying blind.  If there's
just one spindle, I think it's got to be right to read the relation
sequentially.  But if there are multiple spindles, it might not be,
but it seems hard to predict what we should do.  We don't know what
the RAID chunk size is or how many spindles there are, so any guess as
to how to chunk up the relation and divide up the work between workers
is just a shot in the dark.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

<div dir="ltr"><div class="gmail_extra"><div class="gmail_quote">On 28 January 2015 at 14:03, Robert Haas <span
dir="ltr"><<ahref="mailto:robertmhaas@gmail.com" target="_blank">robertmhaas@gmail.com</a>></span> wrote:<br
/><blockquoteclass="gmail_quote" style="margin:0 0 0 .8ex;border-left:1px #ccc solid;padding-left:1ex">The problem
here,as I see it, is that we're flying blind.  If there's<br /> just one spindle, I think it's got to be right to read
therelation<br /> sequentially.  But if there are multiple spindles, it might not be,<br /> but it seems hard to
predictwhat we should do.  We don't know what<br /> the RAID chunk size is or how many spindles there are, so any guess
as<br/> to how to chunk up the relation and divide up the work between workers<br /> is just a shot in the
dark.</blockquote></div><br/></div><div class="gmail_extra">Can't the planner take effective_io_concurrency into
account?<brclear="all" /></div><div class="gmail_extra"><br /><div class="gmail_signature">Thom</div></div></div> 

On Wed, Jan 28, 2015 at 9:12 AM, Thom Brown <thom@linux.com> wrote:
> On 28 January 2015 at 14:03, Robert Haas <robertmhaas@gmail.com> wrote:
>> The problem here, as I see it, is that we're flying blind.  If there's
>> just one spindle, I think it's got to be right to read the relation
>> sequentially.  But if there are multiple spindles, it might not be,
>> but it seems hard to predict what we should do.  We don't know what
>> the RAID chunk size is or how many spindles there are, so any guess as
>> to how to chunk up the relation and divide up the work between workers
>> is just a shot in the dark.
>
> Can't the planner take effective_io_concurrency into account?

Maybe.  It's answering a somewhat the right question -- to tell us how
many parallel I/O channels we think we've got.  But I'm not quite sure
what the to do with that information in this case.  I mean, if we've
got effective_io_concurrency = 6, does that mean it's right to start
scans in 6 arbitrary places in the relation and hope that keeps all
the drives busy?  That seems like throwing darts at the wall.  We have
no idea which parts are on which underlying devices.  Or maybe it mean
we should prefetch 24MB, on the assumption that the RAID stripe is
4MB?  That's definitely blind guesswork.

Considering the email Amit just sent, it looks like on this machine,
regardless of what algorithm we used, the scan took between 3 minutes
and 5.5 minutes, and most of them took between 4 minutes and 5.5
minutes.  The results aren't very predictable, more workers don't
necessarily help, and it's not really clear that any algorithm we've
tried is clearly better than any other.  I experimented with
prefetching a bit yesterday, too, and it was pretty much the same.
Some settings made it slightly faster.  Others made it slower.  Whee!

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

Robert Haas <robertmhaas@gmail.com> writes:
> The problem here, as I see it, is that we're flying blind.  If there's
> just one spindle, I think it's got to be right to read the relation
> sequentially.  But if there are multiple spindles, it might not be,
> but it seems hard to predict what we should do.  We don't know what
> the RAID chunk size is or how many spindles there are, so any guess as
> to how to chunk up the relation and divide up the work between workers
> is just a shot in the dark.

I thought the proposal to chunk on the basis of "each worker processes
one 1GB-sized segment" should work all right.  The kernel should see that
as sequential reads of different files, issued by different processes;
and if it can't figure out how to process that efficiently then it's a
very sad excuse for a kernel.

You are right that trying to do any detailed I/O scheduling by ourselves
is a doomed exercise.  For better or worse, we have kept ourselves at
sufficient remove from the hardware that we can't possibly do that
successfully.
        regards, tom lane

On Wed, Jan 28, 2015 at 10:40 AM, Tom Lane <tgl@sss.pgh.pa.us> wrote:
> Robert Haas <robertmhaas@gmail.com> writes:
>> The problem here, as I see it, is that we're flying blind.  If there's
>> just one spindle, I think it's got to be right to read the relation
>> sequentially.  But if there are multiple spindles, it might not be,
>> but it seems hard to predict what we should do.  We don't know what
>> the RAID chunk size is or how many spindles there are, so any guess as
>> to how to chunk up the relation and divide up the work between workers
>> is just a shot in the dark.
>
> I thought the proposal to chunk on the basis of "each worker processes
> one 1GB-sized segment" should work all right.  The kernel should see that
> as sequential reads of different files, issued by different processes;
> and if it can't figure out how to process that efficiently then it's a
> very sad excuse for a kernel.

I agree.  But there's only value in doing something like that if we
have evidence that it improves anything.  Such evidence is presently a
bit thin on the ground.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

Robert Haas <robertmhaas@gmail.com> writes:
> On Wed, Jan 28, 2015 at 10:40 AM, Tom Lane <tgl@sss.pgh.pa.us> wrote:
>> I thought the proposal to chunk on the basis of "each worker processes
>> one 1GB-sized segment" should work all right.  The kernel should see that
>> as sequential reads of different files, issued by different processes;
>> and if it can't figure out how to process that efficiently then it's a
>> very sad excuse for a kernel.

> I agree.  But there's only value in doing something like that if we
> have evidence that it improves anything.  Such evidence is presently a
> bit thin on the ground.

Well, of course none of this should get committed without convincing
evidence that it's a win.  But I think that chunking on relation segment
boundaries is a plausible way of dodging the problem that we can't do
explicitly hardware-aware scheduling.
        regards, tom lane

* Robert Haas (robertmhaas@gmail.com) wrote:
> On Wed, Jan 28, 2015 at 10:40 AM, Tom Lane <tgl@sss.pgh.pa.us> wrote:
> > I thought the proposal to chunk on the basis of "each worker processes
> > one 1GB-sized segment" should work all right.  The kernel should see that
> > as sequential reads of different files, issued by different processes;
> > and if it can't figure out how to process that efficiently then it's a
> > very sad excuse for a kernel.

Agreed.

> I agree.  But there's only value in doing something like that if we
> have evidence that it improves anything.  Such evidence is presently a
> bit thin on the ground.

You need an i/o subsystem that's fast enough to keep a single CPU busy,
otherwise (as you mentioned elsewhere), you're just going to be i/o
bound and having more processes isn't going to help (and could hurt).

Such i/o systems do exist, but a single RAID5 group over spinning rust
with a simple filter isn't going to cut it with a modern CPU- we're just
too darn efficient to end up i/o bound in that case.  A more complex
filter might be able to change it over to being more CPU bound than i/o
bound and produce the performance improvments you're looking for.

The caveat to this is if you have multiple i/o *channels* (which it
looks like you don't in this case) where you can parallelize across
those channels by having multiple processes involved.  We only support
multiple i/o channels today with tablespaces and we can't span tables
across tablespaces.  That's a problem when working with large data sets,
but I'm hopeful that this work will eventually lead to a parallelized
Append node that operates against a partitioned/inheirited table to work
across multiple tablespaces.
Thanks,
    Stephen

* Stephen Frost (sfrost@snowman.net) wrote:
> Such i/o systems do exist, but a single RAID5 group over spinning rust
> with a simple filter isn't going to cut it with a modern CPU- we're just
> too darn efficient to end up i/o bound in that case.

err, to *not* end up i/o bound.
Thanks,
    Stephen

On 1/28/15 9:56 AM, Stephen Frost wrote:
> * Robert Haas (robertmhaas@gmail.com) wrote:
>> On Wed, Jan 28, 2015 at 10:40 AM, Tom Lane <tgl@sss.pgh.pa.us> wrote:
>>> I thought the proposal to chunk on the basis of "each worker processes
>>> one 1GB-sized segment" should work all right.  The kernel should see that
>>> as sequential reads of different files, issued by different processes;
>>> and if it can't figure out how to process that efficiently then it's a
>>> very sad excuse for a kernel.
>
> Agreed.
>
>> I agree.  But there's only value in doing something like that if we
>> have evidence that it improves anything.  Such evidence is presently a
>> bit thin on the ground.
>
> You need an i/o subsystem that's fast enough to keep a single CPU busy,
> otherwise (as you mentioned elsewhere), you're just going to be i/o
> bound and having more processes isn't going to help (and could hurt).
>
> Such i/o systems do exist, but a single RAID5 group over spinning rust
> with a simple filter isn't going to cut it with a modern CPU- we're just
> too darn efficient to end up i/o bound in that case.  A more complex
> filter might be able to change it over to being more CPU bound than i/o
> bound and produce the performance improvments you're looking for.

Except we're nowhere near being IO efficient. The vast difference between Postgres IO rates and dd shows this. I
suspectthat's because we're not giving the OS a list of IO to perform while we're doing our thing, but that's just a
guess.

> The caveat to this is if you have multiple i/o *channels* (which it
> looks like you don't in this case) where you can parallelize across
> those channels by having multiple processes involved.

Keep in mind that multiple processes is in no way a requirement for that. Async IO would do that, or even just
requestingstuff from the OS before we need it.
 

>  We only support
> multiple i/o channels today with tablespaces and we can't span tables
> across tablespaces.  That's a problem when working with large data sets,
> but I'm hopeful that this work will eventually lead to a parallelized
> Append node that operates against a partitioned/inheirited table to work
> across multiple tablespaces.

Until we can get a single seqscan close to dd performance, I fear worrying about tablespaces and IO channels is
entirelypremature.
 
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

Jim,

* Jim Nasby (Jim.Nasby@BlueTreble.com) wrote:
> On 1/28/15 9:56 AM, Stephen Frost wrote:
> >Such i/o systems do exist, but a single RAID5 group over spinning rust
> >with a simple filter isn't going to cut it with a modern CPU- we're just
> >too darn efficient to end up i/o bound in that case.  A more complex
> >filter might be able to change it over to being more CPU bound than i/o
> >bound and produce the performance improvments you're looking for.
>
> Except we're nowhere near being IO efficient. The vast difference between Postgres IO rates and dd shows this. I
suspectthat's because we're not giving the OS a list of IO to perform while we're doing our thing, but that's just a
guess.

Uh, huh?  The dd was ~321000 and the slowest uncached PG run from
Robert's latest tests was 337312.554, based on my inbox history at
least.  I don't consider ~4-5% difference to be vast.

> >The caveat to this is if you have multiple i/o *channels* (which it
> >looks like you don't in this case) where you can parallelize across
> >those channels by having multiple processes involved.
>
> Keep in mind that multiple processes is in no way a requirement for that. Async IO would do that, or even just
requestingstuff from the OS before we need it. 

While I agree with this in principle, experience has shown that it
doesn't tend to work out as well as we'd like with a single process.

> > We only support
> >multiple i/o channels today with tablespaces and we can't span tables
> >across tablespaces.  That's a problem when working with large data sets,
> >but I'm hopeful that this work will eventually lead to a parallelized
> >Append node that operates against a partitioned/inheirited table to work
> >across multiple tablespaces.
>
> Until we can get a single seqscan close to dd performance, I fear worrying about tablespaces and IO channels is
entirelypremature. 

I feel like one of us is misunderstanding the numbers, which is probably
in part because they're a bit piecemeal over email, but the seqscan
speed in this case looks pretty close to dd performance for this
particular test, when things are uncached.  Cached numbers are
different, but that's not what we're discussing here, I don't think.

Don't get me wrong- I've definitely seen cases where we're CPU bound
because of complex filters, etc, but that doesn't seem to be the case
here.
Thanks!
    Stephen

On Wed, Jan 28, 2015 at 8:27 PM, Stephen Frost <sfrost@snowman.net> wrote:
> I feel like one of us is misunderstanding the numbers, which is probably
> in part because they're a bit piecemeal over email, but the seqscan
> speed in this case looks pretty close to dd performance for this
> particular test, when things are uncached.  Cached numbers are
> different, but that's not what we're discussing here, I don't think.
>
> Don't get me wrong- I've definitely seen cases where we're CPU bound
> because of complex filters, etc, but that doesn't seem to be the case
> here.

To try to clarify a bit: What we've testing here is a function I wrote
called parallel_count(regclass), which counts all the visible tuples
in a named relation.  That runs as fast as dd, and giving it extra
workers or prefetching or the ability to read the relation with
different I/O patterns never seems to speed anything up very much.

The story with parallel sequential scan itself may well be different,
since that has a lot more CPU overhead than a dumb-simple
tuple-counter.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

Robert Haas <robertmhaas@gmail.com> writes:

> On Wed, Jan 28, 2015 at 9:12 AM, Thom Brown <thom@linux.com> wrote:
>> On 28 January 2015 at 14:03, Robert Haas <robertmhaas@gmail.com> wrote:
>>> The problem here, as I see it, is that we're flying blind.  If there's
>>> just one spindle, I think it's got to be right to read the relation
>>> sequentially.  But if there are multiple spindles, it might not be,
>>> but it seems hard to predict what we should do.  We don't know what
>>> the RAID chunk size is or how many spindles there are, so any guess as
>>> to how to chunk up the relation and divide up the work between workers
>>> is just a shot in the dark.
>>
>> Can't the planner take effective_io_concurrency into account?
>
> Maybe.  It's answering a somewhat the right question -- to tell us how
> many parallel I/O channels we think we've got.  But I'm not quite sure
> what the to do with that information in this case.  I mean, if we've
> got effective_io_concurrency = 6, does that mean it's right to start
> scans in 6 arbitrary places in the relation and hope that keeps all
> the drives busy?  That seems like throwing darts at the wall.  We have
> no idea which parts are on which underlying devices.  Or maybe it mean
> we should prefetch 24MB, on the assumption that the RAID stripe is
> 4MB?  That's definitely blind guesswork.
>
> Considering the email Amit just sent, it looks like on this machine,
> regardless of what algorithm we used, the scan took between 3 minutes
> and 5.5 minutes, and most of them took between 4 minutes and 5.5
> minutes.  The results aren't very predictable, more workers don't
> necessarily help, and it's not really clear that any algorithm we've
> tried is clearly better than any other.  I experimented with
> prefetching a bit yesterday, too, and it was pretty much the same.
> Some settings made it slightly faster.  Others made it slower.  Whee!

I have been researching this topic long time ago.  One notably fact is
that active prefetching disables automatic readahead prefetching (by
Linux kernel), which can occour in larger granularities than 8K.
Automatic readahead prefetching occours when consecutive addresses are
read, which may happen by a seqscan but also by "accident" through an
indexscan in correlated cases.

My consequence was to NOT prefetch seqscans, because OS does good enough
without advice.  Prefetching indexscan heap accesses is very valuable
though, but you need to detect the accidential sequential accesses to
not hurt your performance in correlated cases.

In general I can give you the hint to not only focus on HDDs with their
single spindle.  A single SATA SSD scales up to 32 (31 on Linux)
requests in parallel (without RAID or anything else).  The difference in
throughput is extreme for this type of storage device.  While single
spinning HDDs can only gain up to ~20% by NCQ, SATA SSDs can easily gain
up to 700%.

+1 for using effective_io_concurrency to tune for this, since
prefetching random addresses is effectively a type of parallel I/O.

Regards,
Daniel
--
MSc. Daniel Bausch
Research Assistant (Computer Science)
Technische Universität Darmstadt
http://www.dvs.tu-darmstadt.de/staff/dbausch

Jeff Janes <jeff.janes@gmail.com> writes:
> On Tue, Jan 27, 2015 at 11:08 PM, Heikki Linnakangas <
> hlinnakangas@vmware.com> wrote:
>> OTOH, spreading the I/O across multiple files is not a good thing, if you
>> don't have a RAID setup like that. With a single spindle, you'll just
>> induce more seeks.
>> 
>> Perhaps the OS is smart enough to read in large-enough chunks that the
>> occasional seek doesn't hurt much. But then again, why isn't the OS smart
>> enough to read in large-enough chunks to take advantage of the RAID even
>> when you read just a single file?

> In my experience with RAID, it is smart enough to take advantage of that.
> If the raid controller detects a sequential access pattern read, it
> initiates a read ahead on each disk to pre-position the data it will need
> (or at least, the behavior I observe is as-if it did that).  But maybe if
> the sequential read is a bunch of "random" reads from different processes
> which just happen to add up to sequential, that confuses the algorithm?

If seqscan detection is being done at the level of the RAID controller,
I rather imagine that the controller would not know which process had
initiated which read anyway.  But if it's being done at the level of the
kernel, it's a whole nother thing, and I bet it *would* matter.
        regards, tom lane

On Thu, Jan 29, 2015 at 11:40 AM, Tom Lane <tgl@sss.pgh.pa.us> wrote:
>> In my experience with RAID, it is smart enough to take advantage of that.
>> If the raid controller detects a sequential access pattern read, it
>> initiates a read ahead on each disk to pre-position the data it will need
>> (or at least, the behavior I observe is as-if it did that).  But maybe if
>> the sequential read is a bunch of "random" reads from different processes
>> which just happen to add up to sequential, that confuses the algorithm?
>
> If seqscan detection is being done at the level of the RAID controller,
> I rather imagine that the controller would not know which process had
> initiated which read anyway.  But if it's being done at the level of the
> kernel, it's a whole nother thing, and I bet it *would* matter.

That was my feeling too.  On the machine that Amit and I have been
using for testing, we can't find any really convincing evidence that
it matters.  I won't be a bit surprised if there are other systems
where it does matter, but I don't know how to find them except to
encourage other people to help test.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On 1/28/15 7:27 PM, Stephen Frost wrote:
> * Jim Nasby (Jim.Nasby@BlueTreble.com) wrote:
>> >On 1/28/15 9:56 AM, Stephen Frost wrote:
>>> > >Such i/o systems do exist, but a single RAID5 group over spinning rust
>>> > >with a simple filter isn't going to cut it with a modern CPU- we're just
>>> > >too darn efficient to end up i/o bound in that case.  A more complex
>>> > >filter might be able to change it over to being more CPU bound than i/o
>>> > >bound and produce the performance improvments you're looking for.
>> >
>> >Except we're nowhere near being IO efficient. The vast difference between Postgres IO rates and dd shows this. I
suspectthat's because we're not giving the OS a list of IO to perform while we're doing our thing, but that's just a
guess.
> Uh, huh?  The dd was ~321000 and the slowest uncached PG run from
> Robert's latest tests was 337312.554, based on my inbox history at
> least.  I don't consider ~4-5% difference to be vast.

Sorry, I was speaking more generally than this specific test. In the past I've definitely seen SeqScan performance that
wasan order of magnitude slower than what dd would do. This was an older version of Postgres and an older version of
linux,running on an iSCSI SAN. My suspicion is that the added IO latency imposed by iSCSI is what was causing this, but
that'sjust conjecture.
 

I think Robert was saying that he hasn't been able to see this effect on their test server... that makes me think it's
doingread-ahead on the OS level. But I suspect it's pretty touch and go to rely on that; I'd prefer we have some way to
explicitlyget that behavior where we want it.
 
-- 
Jim Nasby, Data Architect, Blue Treble Consulting
Data in Trouble? Get it in Treble! http://BlueTreble.com

Daniel,

* Daniel Bausch (bausch@dvs.tu-darmstadt.de) wrote:
> I have been researching this topic long time ago.  One notably fact is
> that active prefetching disables automatic readahead prefetching (by
> Linux kernel), which can occour in larger granularities than 8K.
> Automatic readahead prefetching occours when consecutive addresses are
> read, which may happen by a seqscan but also by "accident" through an
> indexscan in correlated cases.

That strikes me as a pretty good point to consider.

> My consequence was to NOT prefetch seqscans, because OS does good enough
> without advice.  Prefetching indexscan heap accesses is very valuable
> though, but you need to detect the accidential sequential accesses to
> not hurt your performance in correlated cases.

Seems like we might be able to do that, it's not that different from
what we do with the bitmap scan case, we'd just look at the bitmap and
see if there's long runs of 1's.

> In general I can give you the hint to not only focus on HDDs with their
> single spindle.  A single SATA SSD scales up to 32 (31 on Linux)
> requests in parallel (without RAID or anything else).  The difference in
> throughput is extreme for this type of storage device.  While single
> spinning HDDs can only gain up to ~20% by NCQ, SATA SSDs can easily gain
> up to 700%.

I definitely agree with the idea that we should be looking at SSD-based
systems but I don't know if anyone happens to have easy access to server
gear with SSDs.  I've got an SSD in my laptop, but that's not really the
same thing.
Thanks!
    Stephen

This patch depends on https://commitfest.postgresql.org/3/22/ 

Hi David and others!

David Fetter <david@fetter.org> writes:

> On Tue, Jan 27, 2015 at 08:02:37AM +0100, Daniel Bausch wrote:
>>
>> Tom Lane <tgl@sss.pgh.pa.us> writes:
>>
>> >> Wait for first IO, issue second IO request
>> >> Compute
>> >> Already have second IO request, issue third
>> >> ...
>> >
>> >> We'd be a lot less sensitive to IO latency.
>> >
>> > It would take about five minutes of coding to prove or disprove this:
>> > stick a PrefetchBuffer call into heapgetpage() to launch a request for the
>> > next page as soon as we've read the current one, and then see if that
>> > makes any obvious performance difference.  I'm not convinced that it will,
>> > but if it did then we could think about how to make it work for real.
>>
>> Sorry for dropping in so late...
>>
>> I have done all this two years ago.  For TPC-H Q8, Q9, Q17, Q20, and Q21
>> I see a speedup of ~100% when using IndexScan prefetching + Nested-Loops
>> Look-Ahead (the outer loop!).
>> (On SSD with 32 Pages Prefetch/Look-Ahead + Cold Page Cache / Small RAM)
>
> Would you be so kind as to pass along any patches (ideally applicable
> to git master), tests, and specific measurements you made?

Attached find my patches based on the old revision
36f4c7843cf3d201279855ed9a6ebc1deb3c9463
(Adjust cube.out expected output for new test queries.)

I did not test applicability against HEAD by now.

Disclaimer: This was just a proof-of-concept and so is poor
implementation quality.  Nevertheless, performance looked promising
while it still needs a lot of extra rules for special cases, like
detecting accidential sequential scans.  General assumption is: no
concurrency - a single query owning the machine.

Here is a comparison using dbt3.  Q8, Q9, Q17, Q20, and Q21 are
significantly improved.

|     |   baseline |  indexscan | indexscan+nestloop |
|     |            | patch 1+2  | patch 3            |
|-----+------------+------------+--------------------|
| Q1  |  76.124261 |  73.165161 |          76.323119 |
| Q2  |   9.676956 |  11.211073 |          10.480668 |
| Q3  |  36.836417 |  36.268022 |          36.837226 |
| Q4  |  48.707501 |    64.2255 |          30.872218 |
| Q5  |  59.371467 |  59.205048 |          58.646096 |
| Q6  |  70.514214 |  73.021006 |           72.64643 |
| Q7  |  63.667594 |  63.258499 |          62.758288 |
| Q8  |  70.640973 |  33.144454 |          32.530732 |
| Q9  | 446.630473 | 379.063773 |         219.926094 |
| Q10 |  49.616125 |  49.244744 |          48.411664 |
| Q11 |   6.122317 |   6.158616 |           6.160189 |
| Q12 |  74.294292 |  87.780442 |          87.533936 |
| Q13 |   32.37932 |  32.771938 |          33.483444 |
| Q14 |  47.836053 |  48.093996 |           47.72221 |
| Q15 | 139.350038 | 138.880208 |         138.681336 |
| Q16 |  12.092429 |  12.120661 |          11.668971 |
| Q17 |   9.346636 |   4.106042 |           4.018951 |
| Q18 |  66.106875 | 123.754111 |         122.623193 |
| Q19 |  22.750504 |  23.191532 |           22.34084 |
| Q20 |  80.481986 |  29.906274 |           28.58106 |
| Q21 | 396.897269 |  355.45988 |          214.44184 |
| Q22 |   6.834841 |   6.600922 |           6.524032 |

Regards,
Daniel
--
MSc. Daniel Bausch
Research Assistant (Computer Science)
Technische Universität Darmstadt
http://www.dvs.tu-darmstadt.de/staff/dbausch
>From 569398929d899100b769abfd919bc3383626ac9f Mon Sep 17 00:00:00 2001
From: Daniel Bausch <bausch@dvs.tu-darmstadt.de>
Date: Tue, 22 Oct 2013 15:22:25 +0200
Subject: [PATCH 1/4] Quick proof-of-concept for indexscan prefetching

This implements a prefetching queue of tuples whose tid is read ahead.
Their block number is quickly checked for random properties (not current
block and not the block prefetched last).  Random reads are prefetched.
Up to 32 tuples are considered by default.  The tids are queued in a
fixed ring buffer.

The prefetching is implemented in the generic part of the index scan, so
it applies to all access methods.
---
 src/backend/access/index/indexam.c | 96 ++++++++++++++++++++++++++++++++++++++
 src/include/access/relscan.h       | 12 +++++
 2 files changed, 108 insertions(+)

diff --git a/src/backend/access/index/indexam.c b/src/backend/access/index/indexam.c
index b878155..1c54ef5 100644
--- a/src/backend/access/index/indexam.c
+++ b/src/backend/access/index/indexam.c
@@ -251,6 +251,12 @@ index_beginscan(Relation heapRelation,
     scan->heapRelation = heapRelation;
     scan->xs_snapshot = snapshot;

+#ifdef USE_PREFETCH
+    scan->xs_prefetch_head = scan->xs_prefetch_tail = -1;
+    scan->xs_last_prefetch = -1;
+    scan->xs_done = false;
+#endif
+
     return scan;
 }

@@ -432,6 +438,55 @@ index_restrpos(IndexScanDesc scan)
     FunctionCall1(procedure, PointerGetDatum(scan));
 }

+static int
+index_prefetch_queue_space(IndexScanDesc scan)
+{
+    if (scan->xs_prefetch_tail < 0)
+        return INDEXSCAN_PREFETCH_COUNT;
+
+    Assert(scan->xs_prefetch_head >= 0);
+
+    return (INDEXSCAN_PREFETCH_COUNT
+            - (scan->xs_prefetch_tail - scan->xs_prefetch_head + 1))
+        % INDEXSCAN_PREFETCH_COUNT;
+}
+
+/* makes copy of ItemPointerData */
+static bool
+index_prefetch_queue_push(IndexScanDesc scan, ItemPointer tid)
+{
+    Assert(index_prefetch_queue_space(scan) > 0);
+
+    if (scan->xs_prefetch_tail == -1)
+        scan->xs_prefetch_head = scan->xs_prefetch_tail = 0;
+    else
+        scan->xs_prefetch_tail =
+            (scan->xs_prefetch_tail + 1) % INDEXSCAN_PREFETCH_COUNT;
+
+    scan->xs_prefetch_queue[scan->xs_prefetch_tail] = *tid;
+
+    return true;
+}
+
+static ItemPointer
+index_prefetch_queue_pop(IndexScanDesc scan)
+{
+    ItemPointer res;
+
+    if (scan->xs_prefetch_head < 0)
+        return NULL;
+
+    res = &scan->xs_prefetch_queue[scan->xs_prefetch_head];
+
+    if (scan->xs_prefetch_head == scan->xs_prefetch_tail)
+        scan->xs_prefetch_head = scan->xs_prefetch_tail = -1;
+    else
+        scan->xs_prefetch_head =
+            (scan->xs_prefetch_head + 1) % INDEXSCAN_PREFETCH_COUNT;
+
+    return res;
+}
+
 /* ----------------
  * index_getnext_tid - get the next TID from a scan
  *
@@ -444,12 +499,52 @@ index_getnext_tid(IndexScanDesc scan, ScanDirection direction)
 {
     FmgrInfo   *procedure;
     bool        found;
+    ItemPointer    from_queue;
+    BlockNumber    pf_block;

     SCAN_CHECKS;
     GET_SCAN_PROCEDURE(amgettuple);

     Assert(TransactionIdIsValid(RecentGlobalXmin));

+#ifdef USE_PREFETCH
+    while (!scan->xs_done && index_prefetch_queue_space(scan) > 0) {
+        /*
+         * The AM's amgettuple proc finds the next index entry matching the
+         * scan keys, and puts the TID into scan->xs_ctup.t_self.  It should
+         * also set scan->xs_recheck and possibly scan->xs_itup, though we pay
+         * no attention to those fields here.
+         */
+        found = DatumGetBool(FunctionCall2(procedure,
+                                           PointerGetDatum(scan),
+                                           Int32GetDatum(direction)));
+        if (found)
+        {
+            index_prefetch_queue_push(scan, &scan->xs_ctup.t_self);
+            pf_block = ItemPointerGetBlockNumber(&scan->xs_ctup.t_self);
+            /* prefetch only if not the current buffer and not exactly the
+             * previously prefetched buffer (heuristic random detection)
+             * because sequential read-ahead would be redundant */
+            if ((!BufferIsValid(scan->xs_cbuf) ||
+                 pf_block != BufferGetBlockNumber(scan->xs_cbuf)) &&
+                pf_block != scan->xs_last_prefetch)
+            {
+                PrefetchBuffer(scan->heapRelation, MAIN_FORKNUM, pf_block);
+                scan->xs_last_prefetch = pf_block;
+            }
+        }
+        else
+            scan->xs_done = true;
+    }
+    from_queue = index_prefetch_queue_pop(scan);
+    if (from_queue)
+    {
+        scan->xs_ctup.t_self = *from_queue;
+        found = true;
+    }
+    else
+        found = false;
+#else
     /*
      * The AM's amgettuple proc finds the next index entry matching the scan
      * keys, and puts the TID into scan->xs_ctup.t_self.  It should also set
@@ -459,6 +554,7 @@ index_getnext_tid(IndexScanDesc scan, ScanDirection direction)
     found = DatumGetBool(FunctionCall2(procedure,
                                        PointerGetDatum(scan),
                                        Int32GetDatum(direction)));
+#endif

     /* Reset kill flag immediately for safety */
     scan->kill_prior_tuple = false;
diff --git a/src/include/access/relscan.h b/src/include/access/relscan.h
index 3a86ca4..bccc1a4 100644
--- a/src/include/access/relscan.h
+++ b/src/include/access/relscan.h
@@ -93,6 +93,18 @@ typedef struct IndexScanDescData

     /* state data for traversing HOT chains in index_getnext */
     bool        xs_continue_hot;    /* T if must keep walking HOT chain */
+
+#ifdef USE_PREFETCH
+# ifndef INDEXSCAN_PREFETCH_COUNT
+#  define INDEXSCAN_PREFETCH_COUNT 32
+# endif
+    /* prefetch queue - ringbuffer */
+    ItemPointerData xs_prefetch_queue[INDEXSCAN_PREFETCH_COUNT];
+    int            xs_prefetch_head;
+    int            xs_prefetch_tail;
+    BlockNumber    xs_last_prefetch;
+    bool        xs_done;
+#endif
 }    IndexScanDescData;

 /* Struct for heap-or-index scans of system tables */
--
2.0.5

>From 7cb5839dd7751bcdcae6e4cbf69cfd24af10a694 Mon Sep 17 00:00:00 2001
From: Daniel Bausch <bausch@dvs.tu-darmstadt.de>
Date: Wed, 23 Oct 2013 09:45:11 +0200
Subject: [PATCH 2/4] Fix index-only scan and rescan

Prefetching heap data for index-only scans does not make any sense and
it uses a different field (itup), nevertheless.  Deactivate the prefetch
logic for index-only scans.

Reset xs_done and the queue on rescan, so we find tuples again.
Remember last prefetch to detect correlation.
---
 src/backend/access/index/indexam.c | 85 +++++++++++++++++++++-----------------
 1 file changed, 47 insertions(+), 38 deletions(-)

diff --git a/src/backend/access/index/indexam.c b/src/backend/access/index/indexam.c
index 1c54ef5..d8a4622 100644
--- a/src/backend/access/index/indexam.c
+++ b/src/backend/access/index/indexam.c
@@ -353,6 +353,12 @@ index_rescan(IndexScanDesc scan,

     scan->kill_prior_tuple = false;        /* for safety */

+#ifdef USE_PREFETCH
+    /* I think, it does not hurt to remember xs_last_prefetch */
+    scan->xs_prefetch_head = scan->xs_prefetch_tail = -1;
+    scan->xs_done = false;
+#endif
+
     FunctionCall5(procedure,
                   PointerGetDatum(scan),
                   PointerGetDatum(keys),
@@ -508,7 +514,47 @@ index_getnext_tid(IndexScanDesc scan, ScanDirection direction)
     Assert(TransactionIdIsValid(RecentGlobalXmin));

 #ifdef USE_PREFETCH
-    while (!scan->xs_done && index_prefetch_queue_space(scan) > 0) {
+    if (!scan->xs_want_itup)
+    {
+        while (!scan->xs_done && index_prefetch_queue_space(scan) > 0) {
+            /*
+             * The AM's amgettuple proc finds the next index entry matching
+             * the scan keys, and puts the TID into scan->xs_ctup.t_self.  It
+             * should also set scan->xs_recheck and possibly scan->xs_itup,
+             * though we pay no attention to those fields here.
+             */
+            found = DatumGetBool(FunctionCall2(procedure,
+                                               PointerGetDatum(scan),
+                                               Int32GetDatum(direction)));
+            if (found)
+            {
+                index_prefetch_queue_push(scan, &scan->xs_ctup.t_self);
+                pf_block = ItemPointerGetBlockNumber(&scan->xs_ctup.t_self);
+                /* prefetch only if not the current buffer and not exactly the
+                 * previously prefetched buffer (heuristic random detection)
+                 * because sequential read-ahead would be redundant */
+                if ((!BufferIsValid(scan->xs_cbuf) ||
+                     pf_block != BufferGetBlockNumber(scan->xs_cbuf)) &&
+                    pf_block != scan->xs_last_prefetch)
+                {
+                    PrefetchBuffer(scan->heapRelation, MAIN_FORKNUM, pf_block);
+                    scan->xs_last_prefetch = pf_block;
+                }
+            }
+            else
+                scan->xs_done = true;
+        }
+        from_queue = index_prefetch_queue_pop(scan);
+        if (from_queue)
+        {
+            scan->xs_ctup.t_self = *from_queue;
+            found = true;
+        }
+        else
+            found = false;
+    }
+    else
+#endif
         /*
          * The AM's amgettuple proc finds the next index entry matching the
          * scan keys, and puts the TID into scan->xs_ctup.t_self.  It should
@@ -518,43 +564,6 @@ index_getnext_tid(IndexScanDesc scan, ScanDirection direction)
         found = DatumGetBool(FunctionCall2(procedure,
                                            PointerGetDatum(scan),
                                            Int32GetDatum(direction)));
-        if (found)
-        {
-            index_prefetch_queue_push(scan, &scan->xs_ctup.t_self);
-            pf_block = ItemPointerGetBlockNumber(&scan->xs_ctup.t_self);
-            /* prefetch only if not the current buffer and not exactly the
-             * previously prefetched buffer (heuristic random detection)
-             * because sequential read-ahead would be redundant */
-            if ((!BufferIsValid(scan->xs_cbuf) ||
-                 pf_block != BufferGetBlockNumber(scan->xs_cbuf)) &&
-                pf_block != scan->xs_last_prefetch)
-            {
-                PrefetchBuffer(scan->heapRelation, MAIN_FORKNUM, pf_block);
-                scan->xs_last_prefetch = pf_block;
-            }
-        }
-        else
-            scan->xs_done = true;
-    }
-    from_queue = index_prefetch_queue_pop(scan);
-    if (from_queue)
-    {
-        scan->xs_ctup.t_self = *from_queue;
-        found = true;
-    }
-    else
-        found = false;
-#else
-    /*
-     * The AM's amgettuple proc finds the next index entry matching the scan
-     * keys, and puts the TID into scan->xs_ctup.t_self.  It should also set
-     * scan->xs_recheck and possibly scan->xs_itup, though we pay no attention
-     * to those fields here.
-     */
-    found = DatumGetBool(FunctionCall2(procedure,
-                                       PointerGetDatum(scan),
-                                       Int32GetDatum(direction)));
-#endif

     /* Reset kill flag immediately for safety */
     scan->kill_prior_tuple = false;
--
2.0.5

>From d8b1533955e3471fb2eb6a030619dcbc258955a8 Mon Sep 17 00:00:00 2001
From: Daniel Bausch <bausch@dvs.tu-darmstadt.de>
Date: Mon, 28 Oct 2013 10:43:16 +0100
Subject: [PATCH 3/4] First try on tuple look-ahead in nestloop

Similarly to the prefetching logic just added to the index scan, look
ahead tuples in the outer loop of a nested loop scan.  For every tuple
looked ahead issue an explicit request for prefetching to the inner
plan.  Modify the index scan to react on this request.
---
 src/backend/access/index/indexam.c   |  81 +++++++++-----
 src/backend/executor/execProcnode.c  |  36 +++++++
 src/backend/executor/nodeIndexscan.c |  16 +++
 src/backend/executor/nodeNestloop.c  | 200 ++++++++++++++++++++++++++++++++++-
 src/include/access/genam.h           |   4 +
 src/include/executor/executor.h      |   3 +
 src/include/executor/nodeIndexscan.h |   1 +
 src/include/nodes/execnodes.h        |  12 +++
 8 files changed, 323 insertions(+), 30 deletions(-)

diff --git a/src/backend/access/index/indexam.c b/src/backend/access/index/indexam.c
index d8a4622..5f44dec 100644
--- a/src/backend/access/index/indexam.c
+++ b/src/backend/access/index/indexam.c
@@ -493,6 +493,57 @@ index_prefetch_queue_pop(IndexScanDesc scan)
     return res;
 }

+#ifdef USE_PREFETCH
+int
+index_prefetch(IndexScanDesc scan, int maxPrefetch, ScanDirection direction)
+{
+    FmgrInfo   *procedure;
+    int            numPrefetched = 0;
+    bool        found;
+    BlockNumber    pf_block;
+    FILE       *logfile;
+
+    GET_SCAN_PROCEDURE(amgettuple);
+
+    while (numPrefetched < maxPrefetch && !scan->xs_done &&
+           index_prefetch_queue_space(scan) > 0)
+    {
+        /*
+         * The AM's amgettuple proc finds the next index entry matching the
+         * scan keys, and puts the TID into scan->xs_ctup.t_self.  It should
+         * also set scan->xs_recheck and possibly scan->xs_itup, though we pay
+         * no attention to those fields here.
+         */
+        found = DatumGetBool(FunctionCall2(procedure,
+                                           PointerGetDatum(scan),
+                                           Int32GetDatum(direction)));
+        if (found)
+        {
+            index_prefetch_queue_push(scan, &scan->xs_ctup.t_self);
+            pf_block = ItemPointerGetBlockNumber(&scan->xs_ctup.t_self);
+
+            /*
+             * Prefetch only if not the current buffer and not exactly the
+             * previously prefetched buffer (heuristic random detection)
+             * because sequential read-ahead would be redundant
+             */
+            if ((!BufferIsValid(scan->xs_cbuf) ||
+                 pf_block != BufferGetBlockNumber(scan->xs_cbuf)) &&
+                pf_block != scan->xs_last_prefetch)
+            {
+                PrefetchBuffer(scan->heapRelation, MAIN_FORKNUM, pf_block);
+                scan->xs_last_prefetch = pf_block;
+                numPrefetched++;
+            }
+        }
+        else
+            scan->xs_done = true;
+    }
+
+    return numPrefetched;
+}
+#endif
+
 /* ----------------
  * index_getnext_tid - get the next TID from a scan
  *
@@ -506,7 +557,6 @@ index_getnext_tid(IndexScanDesc scan, ScanDirection direction)
     FmgrInfo   *procedure;
     bool        found;
     ItemPointer    from_queue;
-    BlockNumber    pf_block;

     SCAN_CHECKS;
     GET_SCAN_PROCEDURE(amgettuple);
@@ -516,34 +566,7 @@ index_getnext_tid(IndexScanDesc scan, ScanDirection direction)
 #ifdef USE_PREFETCH
     if (!scan->xs_want_itup)
     {
-        while (!scan->xs_done && index_prefetch_queue_space(scan) > 0) {
-            /*
-             * The AM's amgettuple proc finds the next index entry matching
-             * the scan keys, and puts the TID into scan->xs_ctup.t_self.  It
-             * should also set scan->xs_recheck and possibly scan->xs_itup,
-             * though we pay no attention to those fields here.
-             */
-            found = DatumGetBool(FunctionCall2(procedure,
-                                               PointerGetDatum(scan),
-                                               Int32GetDatum(direction)));
-            if (found)
-            {
-                index_prefetch_queue_push(scan, &scan->xs_ctup.t_self);
-                pf_block = ItemPointerGetBlockNumber(&scan->xs_ctup.t_self);
-                /* prefetch only if not the current buffer and not exactly the
-                 * previously prefetched buffer (heuristic random detection)
-                 * because sequential read-ahead would be redundant */
-                if ((!BufferIsValid(scan->xs_cbuf) ||
-                     pf_block != BufferGetBlockNumber(scan->xs_cbuf)) &&
-                    pf_block != scan->xs_last_prefetch)
-                {
-                    PrefetchBuffer(scan->heapRelation, MAIN_FORKNUM, pf_block);
-                    scan->xs_last_prefetch = pf_block;
-                }
-            }
-            else
-                scan->xs_done = true;
-        }
+        index_prefetch(scan, INDEXSCAN_PREFETCH_COUNT, direction);
         from_queue = index_prefetch_queue_pop(scan);
         if (from_queue)
         {
diff --git a/src/backend/executor/execProcnode.c b/src/backend/executor/execProcnode.c
index 76dd62f..a8f2c90 100644
--- a/src/backend/executor/execProcnode.c
+++ b/src/backend/executor/execProcnode.c
@@ -741,3 +741,39 @@ ExecEndNode(PlanState *node)
             break;
     }
 }
+
+
+#ifdef USE_PREFETCH
+/* ----------------------------------------------------------------
+ *        ExecPrefetchNode
+ *
+ *        Request explicit prefetching from a subtree/node without
+ *        actually forming a tuple.
+ *
+ *        The node shall request at most 'maxPrefetch' pages being
+ *        prefetched.
+ *
+ *        The function returns how many pages have been requested.
+ *
+ *        Calling this function for a type that does not support
+ *        prefetching is not an error.  It just returns 0 as if no
+ *        prefetching was possible.
+ * ----------------------------------------------------------------
+ */
+int
+ExecPrefetchNode(PlanState *node, int maxPrefetch)
+{
+    if (node == NULL)
+        return 0;
+
+    switch (nodeTag(node))
+    {
+        case T_IndexScanState:
+            return ExecPrefetchIndexScan((IndexScanState *) node,
+                                         maxPrefetch);
+
+        default:
+            return 0;
+    }
+}
+#endif
diff --git a/src/backend/executor/nodeIndexscan.c b/src/backend/executor/nodeIndexscan.c
index f1062f1..bab0e7a 100644
--- a/src/backend/executor/nodeIndexscan.c
+++ b/src/backend/executor/nodeIndexscan.c
@@ -192,6 +192,22 @@ ExecReScanIndexScan(IndexScanState *node)
     ExecScanReScan(&node->ss);
 }

+#ifdef USE_PREFETCH
+/* ----------------------------------------------------------------
+ *        ExecPrefetchIndexScan(node, maxPrefetch)
+ *
+ *        Trigger prefetching of index scan without actually fetching
+ *        a tuple.
+ * ----------------------------------------------------------------
+ */
+int
+ExecPrefetchIndexScan(IndexScanState *node, int maxPrefetch)
+{
+    return index_prefetch(node->iss_ScanDesc, maxPrefetch,
+                          node->ss.ps.state->es_direction);
+}
+#endif
+

 /*
  * ExecIndexEvalRuntimeKeys
diff --git a/src/backend/executor/nodeNestloop.c b/src/backend/executor/nodeNestloop.c
index c7a08ed..21ad5f8 100644
--- a/src/backend/executor/nodeNestloop.c
+++ b/src/backend/executor/nodeNestloop.c
@@ -25,6 +25,90 @@
 #include "executor/nodeNestloop.h"
 #include "utils/memutils.h"

+#ifdef USE_PREFETCH
+static int
+NestLoopLookAheadQueueSpace(NestLoopState *node)
+{
+    if (node->nl_lookAheadQueueTail < 0)
+        return NESTLOOP_PREFETCH_COUNT;
+
+    Assert(node->nl_lookAheadQueueHead >= 0);
+
+    return (NESTLOOP_PREFETCH_COUNT
+            - (node->nl_lookAheadQueueTail - node->nl_lookAheadQueueHead + 1))
+        % NESTLOOP_PREFETCH_COUNT;
+}
+
+/* makes materialized copy of tuple table slot */
+static bool
+NestLoopLookAheadQueuePush(NestLoopState *node, TupleTableSlot *tuple)
+{
+    TupleTableSlot **queueEntry;
+
+    Assert(NestLoopLookAheadQueueSpace(node) > 0);
+
+    if (node->nl_lookAheadQueueTail == -1)
+        node->nl_lookAheadQueueHead = node->nl_lookAheadQueueTail = 0;
+    else
+        node->nl_lookAheadQueueTail =
+            (node->nl_lookAheadQueueTail +1) % NESTLOOP_PREFETCH_COUNT;
+
+    queueEntry = &node->nl_lookAheadQueue[node->nl_lookAheadQueueTail];
+
+    if (!(*queueEntry))
+    {
+        *queueEntry = ExecInitExtraTupleSlot(node->js.ps.state);
+        ExecSetSlotDescriptor(*queueEntry,
+                              ExecGetResultType(outerPlanState(node)));
+    }
+
+    ExecCopySlot(*queueEntry, tuple);
+
+    return true;
+}
+
+static TupleTableSlot *
+NestLoopLookAheadQueuePop(NestLoopState *node)
+{
+    TupleTableSlot *res;
+
+    if (node->nl_lookAheadQueueHead < 0)
+        return NULL;
+
+    res = node->nl_lookAheadQueue[node->nl_lookAheadQueueHead];
+
+    if (node->nl_lookAheadQueueHead == node->nl_lookAheadQueueTail)
+        node->nl_lookAheadQueueHead = node->nl_lookAheadQueueTail = -1;
+    else
+        node->nl_lookAheadQueueHead =
+            (node->nl_lookAheadQueueHead + 1) % NESTLOOP_PREFETCH_COUNT;
+
+    return res;
+}
+
+static void
+NestLoopLookAheadQueueClear(NestLoopState *node)
+{
+    TupleTableSlot *lookAheadTuple;
+    int        i;
+
+    /*
+     * As we do not clear the tuple table slots on pop, we need to scan the
+     * whole array, regardless of the current queue fill.
+     *
+     * We cannot really free the slot, as there is no well defined interface
+     * for that, but the emptied slots will be freed when the query ends.
+     */
+    for (i = 0; i < NESTLOOP_PREFETCH_COUNT; i++)
+    {
+        lookAheadTuple = node->nl_lookAheadQueue[i];
+        /* look only on pointer - all non NULL fields are non-empty */
+        if (lookAheadTuple)
+            ExecClearTuple(lookAheadTuple);
+    }
+
+}
+#endif /* USE_PREFETCH */

 /* ----------------------------------------------------------------
  *        ExecNestLoop(node)
@@ -120,7 +204,87 @@ ExecNestLoop(NestLoopState *node)
         if (node->nl_NeedNewOuter)
         {
             ENL1_printf("getting new outer tuple");
-            outerTupleSlot = ExecProcNode(outerPlan);
+
+#ifdef USE_PREFETCH
+            /*
+             * While we have outer tuples and were not able to request enought
+             * prefetching from the inner plan to properly load the system,
+             * request more outer tuples and inner prefetching for them.
+             *
+             * Unfortunately we can do outer look-ahead directed prefetching
+             * only when we are rescanning the inner plan anyway; otherwise we
+             * would break the inner scan.  Only an independent copy of the
+             * inner plan state would allow us to prefetch accross inner loops
+             * regardless of inner scan position.
+             */
+            while (!node->nl_lookAheadDone &&
+                   node->nl_numInnerPrefetched < NESTLOOP_PREFETCH_COUNT &&
+                   NestLoopLookAheadQueueSpace(node) > 0)
+            {
+                TupleTableSlot *lookAheadTupleSlot = ExecProcNode(outerPlan);
+
+                if (!TupIsNull(lookAheadTupleSlot))
+                {
+                    NestLoopLookAheadQueuePush(node, lookAheadTupleSlot);
+
+                    /*
+                     * Set inner params according to look-ahead tuple.
+                     *
+                     * Fetch the values of any outer Vars that must be passed
+                     * to the inner scan, and store them in the appropriate
+                     * PARAM_EXEC slots.
+                     */
+                    foreach(lc, nl->nestParams)
+                    {
+                        NestLoopParam *nlp = (NestLoopParam *) lfirst(lc);
+                        int            paramno = nlp->paramno;
+                        ParamExecData *prm;
+
+                        prm = &(econtext->ecxt_param_exec_vals[paramno]);
+                        /* Param value should be an OUTER_VAR var */
+                        Assert(IsA(nlp->paramval, Var));
+                        Assert(nlp->paramval->varno == OUTER_VAR);
+                        Assert(nlp->paramval->varattno > 0);
+                        prm->value = slot_getattr(lookAheadTupleSlot,
+                                                  nlp->paramval->varattno,
+                                                  &(prm->isnull));
+                        /* Flag parameter value as changed */
+                        innerPlan->chgParam =
+                            bms_add_member(innerPlan->chgParam, paramno);
+                    }
+
+                    /*
+                     * Rescan inner plan with changed parameters and request
+                     * explicit prefetch.  Limit the inner prefetch amount
+                     * according to our own bookkeeping.
+                     *
+                     * When the so processed outer tuple gets finally active
+                     * in the inner loop, the inner plan will autonomously
+                     * prefetch the same tuples again.  This is redundant but
+                     * avoiding that seems too complicated for now.  It should
+                     * not hurt too much and may even help in case the
+                     * prefetched blocks have been evicted again in the
+                     * meantime.
+                     */
+                    ExecReScan(innerPlan);
+                    node->nl_numInnerPrefetched +=
+                        ExecPrefetchNode(innerPlan,
+                                         NESTLOOP_PREFETCH_COUNT -
+                                         node->nl_numInnerPrefetched);
+                }
+                else
+                    node->nl_lookAheadDone = true; /* outer plan exhausted */
+            }
+
+            /*
+             * If there is already the next outerPlan in our look-ahead queue,
+             * get the next outer tuple from there, otherwise execute the
+             * outer plan.
+             */
+            outerTupleSlot = NestLoopLookAheadQueuePop(node);
+            if (TupIsNull(outerTupleSlot) && !node->nl_lookAheadDone)
+#endif /* USE_PREFETCH */
+                outerTupleSlot = ExecProcNode(outerPlan);

             /*
              * if there are no more outer tuples, then the join is complete..
@@ -174,6 +338,18 @@ ExecNestLoop(NestLoopState *node)
         innerTupleSlot = ExecProcNode(innerPlan);
         econtext->ecxt_innertuple = innerTupleSlot;

+#ifdef USE_PREFETCH
+        /*
+         * Decrement prefetch counter as we cosume inner tuples.  We need to
+         * check for >0 because prefetching might not have happened for the
+         * consumed tuple, maybe because explicit prefetching is not supported
+         * by the inner plan or because the explicit prefetching requested by
+         * us is exhausted and the inner plan is doing it on its own now.
+         */
+        if (node->nl_numInnerPrefetched > 0)
+            node->nl_numInnerPrefetched--;
+#endif
+
         if (TupIsNull(innerTupleSlot))
         {
             ENL1_printf("no inner tuple, need new outer tuple");
@@ -296,6 +472,9 @@ NestLoopState *
 ExecInitNestLoop(NestLoop *node, EState *estate, int eflags)
 {
     NestLoopState *nlstate;
+#ifdef USE_PREFETCH
+    int i;
+#endif

     /* check for unsupported flags */
     Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
@@ -381,6 +560,15 @@ ExecInitNestLoop(NestLoop *node, EState *estate, int eflags)
     nlstate->nl_NeedNewOuter = true;
     nlstate->nl_MatchedOuter = false;

+#ifdef USE_PREFETCH
+    nlstate->nl_lookAheadQueueHead = nlstate->nl_lookAheadQueueTail = -1;
+    nlstate->nl_lookAheadDone = false;
+    nlstate->nl_numInnerPrefetched = 0;
+
+    for (i = 0; i < NESTLOOP_PREFETCH_COUNT; i++)
+        nlstate->nl_lookAheadQueue[i] = NULL;
+#endif
+
     NL1_printf("ExecInitNestLoop: %s\n",
                "node initialized");

@@ -409,6 +597,10 @@ ExecEndNestLoop(NestLoopState *node)
      */
     ExecClearTuple(node->js.ps.ps_ResultTupleSlot);

+#ifdef USE_PREFETCH
+    NestLoopLookAheadQueueClear(node);
+#endif
+
     /*
      * close down subplans
      */
@@ -444,4 +636,10 @@ ExecReScanNestLoop(NestLoopState *node)
     node->js.ps.ps_TupFromTlist = false;
     node->nl_NeedNewOuter = true;
     node->nl_MatchedOuter = false;
+
+#ifdef USE_PREFETCH
+    NestLoopLookAheadQueueClear(node);
+    node->nl_lookAheadDone = false;
+    node->nl_numInnerPrefetched = 0;
+#endif
 }
diff --git a/src/include/access/genam.h b/src/include/access/genam.h
index a800041..7733b3c 100644
--- a/src/include/access/genam.h
+++ b/src/include/access/genam.h
@@ -146,6 +146,10 @@ extern void index_markpos(IndexScanDesc scan);
 extern void index_restrpos(IndexScanDesc scan);
 extern ItemPointer index_getnext_tid(IndexScanDesc scan,
                   ScanDirection direction);
+#ifdef USE_PREFETCH
+extern int index_prefetch(IndexScanDesc scan, int maxPrefetch,
+                          ScanDirection direction);
+#endif
 extern HeapTuple index_fetch_heap(IndexScanDesc scan);
 extern HeapTuple index_getnext(IndexScanDesc scan, ScanDirection direction);
 extern int64 index_getbitmap(IndexScanDesc scan, TIDBitmap *bitmap);
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index 75841c8..88d0522 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -221,6 +221,9 @@ extern PlanState *ExecInitNode(Plan *node, EState *estate, int eflags);
 extern TupleTableSlot *ExecProcNode(PlanState *node);
 extern Node *MultiExecProcNode(PlanState *node);
 extern void ExecEndNode(PlanState *node);
+#ifdef USE_PREFETCH
+extern int ExecPrefetchNode(PlanState *node, int maxPrefetch);
+#endif

 /*
  * prototypes from functions in execQual.c
diff --git a/src/include/executor/nodeIndexscan.h b/src/include/executor/nodeIndexscan.h
index 71dbd9c..f93632c 100644
--- a/src/include/executor/nodeIndexscan.h
+++ b/src/include/executor/nodeIndexscan.h
@@ -18,6 +18,7 @@

 extern IndexScanState *ExecInitIndexScan(IndexScan *node, EState *estate, int eflags);
 extern TupleTableSlot *ExecIndexScan(IndexScanState *node);
+extern int ExecPrefetchIndexScan(IndexScanState *node, int maxPrefetch);
 extern void ExecEndIndexScan(IndexScanState *node);
 extern void ExecIndexMarkPos(IndexScanState *node);
 extern void ExecIndexRestrPos(IndexScanState *node);
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 3b430e0..27fe65d 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1526,6 +1526,18 @@ typedef struct NestLoopState
     bool        nl_NeedNewOuter;
     bool        nl_MatchedOuter;
     TupleTableSlot *nl_NullInnerTupleSlot;
+
+#ifdef USE_PREFETCH
+# ifndef NESTLOOP_PREFETCH_COUNT
+#  define NESTLOOP_PREFETCH_COUNT 32
+# endif
+    /* look-ahead queue (for prefetching) - ringbuffer */
+    TupleTableSlot *nl_lookAheadQueue[NESTLOOP_PREFETCH_COUNT];
+    int            nl_lookAheadQueueHead;
+    int            nl_lookAheadQueueTail;
+    bool        nl_lookAheadDone;
+    int            nl_numInnerPrefetched;
+#endif
 } NestLoopState;

 /* ----------------
--
2.0.5

>From a1fcab2d9d001505a5fc25accdca71e88148e4ff Mon Sep 17 00:00:00 2001
From: Daniel Bausch <bausch@dvs.tu-darmstadt.de>
Date: Tue, 29 Oct 2013 16:41:09 +0100
Subject: [PATCH 4/4] Limit recursive prefetching for merge join

Add switch facility to limit the prefetching of a subtree recursively.
In a first try add support for some variants of merge join.  Distribute
the prefetch allowance evenly between outer and inner subplan.
---
 src/backend/access/index/indexam.c   |  5 +++-
 src/backend/executor/execProcnode.c  | 47 +++++++++++++++++++++++++++++++++++-
 src/backend/executor/nodeAgg.c       | 10 ++++++++
 src/backend/executor/nodeIndexscan.c | 18 ++++++++++++++
 src/backend/executor/nodeMaterial.c  | 14 +++++++++++
 src/backend/executor/nodeMergejoin.c | 22 +++++++++++++++++
 src/include/access/relscan.h         |  1 +
 src/include/executor/executor.h      |  1 +
 src/include/executor/nodeAgg.h       |  3 +++
 src/include/executor/nodeIndexscan.h |  3 +++
 src/include/executor/nodeMaterial.h  |  3 +++
 src/include/executor/nodeMergejoin.h |  3 +++
 src/include/nodes/execnodes.h        |  6 +++++
 13 files changed, 134 insertions(+), 2 deletions(-)

diff --git a/src/backend/access/index/indexam.c b/src/backend/access/index/indexam.c
index 5f44dec..354bde6 100644
--- a/src/backend/access/index/indexam.c
+++ b/src/backend/access/index/indexam.c
@@ -255,6 +255,7 @@ index_beginscan(Relation heapRelation,
     scan->xs_prefetch_head = scan->xs_prefetch_tail = -1;
     scan->xs_last_prefetch = -1;
     scan->xs_done = false;
+    scan->xs_prefetch_limit = INDEXSCAN_PREFETCH_COUNT;
 #endif

     return scan;
@@ -506,7 +507,9 @@ index_prefetch(IndexScanDesc scan, int maxPrefetch, ScanDirection direction)
     GET_SCAN_PROCEDURE(amgettuple);

     while (numPrefetched < maxPrefetch && !scan->xs_done &&
-           index_prefetch_queue_space(scan) > 0)
+           index_prefetch_queue_space(scan) > 0 &&
+           index_prefetch_queue_space(scan) >
+           (INDEXSCAN_PREFETCH_COUNT - scan->xs_prefetch_limit))
     {
         /*
          * The AM's amgettuple proc finds the next index entry matching the
diff --git a/src/backend/executor/execProcnode.c b/src/backend/executor/execProcnode.c
index a8f2c90..a14a0d0 100644
--- a/src/backend/executor/execProcnode.c
+++ b/src/backend/executor/execProcnode.c
@@ -745,6 +745,51 @@ ExecEndNode(PlanState *node)

 #ifdef USE_PREFETCH
 /* ----------------------------------------------------------------
+ *        ExecLimitPrefetchNode
+ *
+ *        Limit the amount of prefetching that may be requested by
+ *        a subplan.
+ *
+ *        Most of the handlers just pass-through the received value
+ *        to their subplans.  That is the case, when they have just
+ *        one subplan that might prefetch.  If they have two subplans
+ *        intelligent heuristics need to be applied to distribute the
+ *        prefetch allowance in a way delivering overall advantage.
+ * ----------------------------------------------------------------
+ */
+void
+ExecLimitPrefetchNode(PlanState *node, int limit)
+{
+    if (node == NULL)
+        return;
+
+    switch (nodeTag(node))
+    {
+        case T_IndexScanState:
+            ExecLimitPrefetchIndexScan((IndexScanState *) node, limit);
+            break;
+
+        case T_MergeJoinState:
+            ExecLimitPrefetchMergeJoin((MergeJoinState *) node, limit);
+            break;
+
+        case T_MaterialState:
+            ExecLimitPrefetchMaterial((MaterialState *) node, limit);
+            break;
+
+        case T_AggState:
+            ExecLimitPrefetchAgg((AggState *) node, limit);
+            break;
+
+        default:
+            elog(INFO,
+                 "missing ExecLimitPrefetchNode handler for node type: %d",
+                 (int) nodeTag(node));
+            break;
+    }
+}
+
+/* ----------------------------------------------------------------
  *        ExecPrefetchNode
  *
  *        Request explicit prefetching from a subtree/node without
@@ -776,4 +821,4 @@ ExecPrefetchNode(PlanState *node, int maxPrefetch)
             return 0;
     }
 }
-#endif
+#endif /* USE_PREFETCH */
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index e02a6ff..94f6d77 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -1877,6 +1877,16 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
     return aggstate;
 }

+#ifdef USE_PREFETCH
+void
+ExecLimitPrefetchAgg(AggState *node, int limit)
+{
+    Assert(node != NULL);
+
+    ExecLimitPrefetchNode(outerPlanState(node), limit);
+}
+#endif
+
 static Datum
 GetAggInitVal(Datum textInitVal, Oid transtype)
 {
diff --git a/src/backend/executor/nodeIndexscan.c b/src/backend/executor/nodeIndexscan.c
index bab0e7a..6ea236e 100644
--- a/src/backend/executor/nodeIndexscan.c
+++ b/src/backend/executor/nodeIndexscan.c
@@ -640,6 +640,24 @@ ExecInitIndexScan(IndexScan *node, EState *estate, int eflags)
     return indexstate;
 }

+#ifdef USE_PREFETCH
+/* ----------------------------------------------------------------
+ *        ExecLimitPrefetchIndexScan
+ *
+ *        Sets/changes the number of tuples whose pages to request in
+ *        advance.
+ * ----------------------------------------------------------------
+ */
+void
+ExecLimitPrefetchIndexScan(IndexScanState *node, int limit)
+{
+    Assert(node != NULL);
+    Assert(node->iss_ScanDesc != NULL);
+
+    node->iss_ScanDesc->xs_prefetch_limit = limit;
+}
+#endif
+

 /*
  * ExecIndexBuildScanKeys
diff --git a/src/backend/executor/nodeMaterial.c b/src/backend/executor/nodeMaterial.c
index 7a82f56..3370362 100644
--- a/src/backend/executor/nodeMaterial.c
+++ b/src/backend/executor/nodeMaterial.c
@@ -232,6 +232,20 @@ ExecInitMaterial(Material *node, EState *estate, int eflags)
     return matstate;
 }

+#ifdef USE_PREFETCH
+/* ----------------------------------------------------------------
+ *        ExecLimitPrefetchMaterial
+ * ----------------------------------------------------------------
+ */
+void
+ExecLimitPrefetchMaterial(MaterialState *node, int limit)
+{
+    Assert(node != NULL);
+
+    ExecLimitPrefetchNode(outerPlanState(node), limit);
+}
+#endif
+
 /* ----------------------------------------------------------------
  *        ExecEndMaterial
  * ----------------------------------------------------------------
diff --git a/src/backend/executor/nodeMergejoin.c b/src/backend/executor/nodeMergejoin.c
index e69bc64..f25e074 100644
--- a/src/backend/executor/nodeMergejoin.c
+++ b/src/backend/executor/nodeMergejoin.c
@@ -1627,6 +1627,10 @@ ExecInitMergeJoin(MergeJoin *node, EState *estate, int eflags)
     mergestate->mj_OuterTupleSlot = NULL;
     mergestate->mj_InnerTupleSlot = NULL;

+#ifdef USE_PREFETCH
+    ExecLimitPrefetchMergeJoin(mergestate, MERGEJOIN_PREFETCH_COUNT);
+#endif
+
     /*
      * initialization successful
      */
@@ -1636,6 +1640,24 @@ ExecInitMergeJoin(MergeJoin *node, EState *estate, int eflags)
     return mergestate;
 }

+#ifdef USE_PREFETCH
+/* ----------------------------------------------------------------
+ *        ExecLimitPrefetchMergeJoin
+ * ----------------------------------------------------------------
+ */
+void
+ExecLimitPrefetchMergeJoin(MergeJoinState *node, int limit)
+{
+    int outerLimit = limit/2;
+    int innerLimit = limit/2;
+
+    Assert(node != NULL);
+
+    ExecLimitPrefetchNode(outerPlanState(node), outerLimit);
+    ExecLimitPrefetchNode(innerPlanState(node), innerLimit);
+}
+#endif
+
 /* ----------------------------------------------------------------
  *        ExecEndMergeJoin
  *
diff --git a/src/include/access/relscan.h b/src/include/access/relscan.h
index bccc1a4..3297900 100644
--- a/src/include/access/relscan.h
+++ b/src/include/access/relscan.h
@@ -104,6 +104,7 @@ typedef struct IndexScanDescData
     int            xs_prefetch_tail;
     BlockNumber    xs_last_prefetch;
     bool        xs_done;
+    int            xs_prefetch_limit;
 #endif
 }    IndexScanDescData;

diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index 88d0522..09b94e0 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -222,6 +222,7 @@ extern TupleTableSlot *ExecProcNode(PlanState *node);
 extern Node *MultiExecProcNode(PlanState *node);
 extern void ExecEndNode(PlanState *node);
 #ifdef USE_PREFETCH
+extern void ExecLimitPrefetchNode(PlanState *node, int limit);
 extern int ExecPrefetchNode(PlanState *node, int maxPrefetch);
 #endif

diff --git a/src/include/executor/nodeAgg.h b/src/include/executor/nodeAgg.h
index 38823d6..f775ec8 100644
--- a/src/include/executor/nodeAgg.h
+++ b/src/include/executor/nodeAgg.h
@@ -17,6 +17,9 @@
 #include "nodes/execnodes.h"

 extern AggState *ExecInitAgg(Agg *node, EState *estate, int eflags);
+#ifdef USE_PREFETCH
+extern void ExecLimitPrefetchAgg(AggState *node, int limit);
+#endif
 extern TupleTableSlot *ExecAgg(AggState *node);
 extern void ExecEndAgg(AggState *node);
 extern void ExecReScanAgg(AggState *node);
diff --git a/src/include/executor/nodeIndexscan.h b/src/include/executor/nodeIndexscan.h
index f93632c..ccf3121 100644
--- a/src/include/executor/nodeIndexscan.h
+++ b/src/include/executor/nodeIndexscan.h
@@ -17,6 +17,9 @@
 #include "nodes/execnodes.h"

 extern IndexScanState *ExecInitIndexScan(IndexScan *node, EState *estate, int eflags);
+#ifdef USE_PREFETCH
+extern void ExecLimitPrefetchIndexScan(IndexScanState *node, int limit);
+#endif
 extern TupleTableSlot *ExecIndexScan(IndexScanState *node);
 extern int ExecPrefetchIndexScan(IndexScanState *node, int maxPrefetch);
 extern void ExecEndIndexScan(IndexScanState *node);
diff --git a/src/include/executor/nodeMaterial.h b/src/include/executor/nodeMaterial.h
index cfca0a5..5c81fe8 100644
--- a/src/include/executor/nodeMaterial.h
+++ b/src/include/executor/nodeMaterial.h
@@ -17,6 +17,9 @@
 #include "nodes/execnodes.h"

 extern MaterialState *ExecInitMaterial(Material *node, EState *estate, int eflags);
+#ifdef USE_PREFETCH
+extern void ExecLimitPrefetchMaterial(MaterialState *node, int limit);
+#endif
 extern TupleTableSlot *ExecMaterial(MaterialState *node);
 extern void ExecEndMaterial(MaterialState *node);
 extern void ExecMaterialMarkPos(MaterialState *node);
diff --git a/src/include/executor/nodeMergejoin.h b/src/include/executor/nodeMergejoin.h
index fa6b5e0..e402b42 100644
--- a/src/include/executor/nodeMergejoin.h
+++ b/src/include/executor/nodeMergejoin.h
@@ -17,6 +17,9 @@
 #include "nodes/execnodes.h"

 extern MergeJoinState *ExecInitMergeJoin(MergeJoin *node, EState *estate, int eflags);
+#ifdef USE_PREFETCH
+extern void ExecLimitPrefetchMergeJoin(MergeJoinState *node, int limit);
+#endif
 extern TupleTableSlot *ExecMergeJoin(MergeJoinState *node);
 extern void ExecEndMergeJoin(MergeJoinState *node);
 extern void ExecReScanMergeJoin(MergeJoinState *node);
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 27fe65d..64ed6fb 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1585,6 +1585,12 @@ typedef struct MergeJoinState
     ExprContext *mj_InnerEContext;
 } MergeJoinState;

+#ifdef USE_PREFETCH
+# ifndef MERGEJOIN_PREFETCH_COUNT
+#  define MERGEJOIN_PREFETCH_COUNT 32
+# endif
+#endif
+
 /* ----------------
  *     HashJoinState information
  *
--
2.0.5

On Fri, Feb 6, 2015 at 9:43 AM, Amit Kapila <amit.kapila16@gmail.com> wrote:
> Here is the latest patch which fixes reported issues and supported
> Prepared Statements and Explain Statement for parallel sequential
> scan.
>
> The main purpose is to get the feedback if possible on overall
> structure/design of code before I goahead.

I'm not very happy with the way this is modularized:

1. The new parallel sequential scan node runs only in the master.  The
workers are running a regular sequential scan with a hack to make them
scan a subset of the blocks.  I think this is wrong; parallel
sequential scan shouldn't require this kind of modifications to the
non-parallel case.

2. InitiateWorkers() is entirely specific to the concerns of parallel
sequential scan.  After looking this over, I think there are three
categories of things that need to be clearly separated.  Some stuff is
going to be needed for any parallel query; some stuff is going to be
needed only for parallel scans but will be needed for any type of
parallel scan, not just parallel sequential scan[1]; some stuff is
needed for any type of node that returns tuples but not for nodes that
don't return tuples (e.g. needed for ParallelSeqScan and
ParallelHashJoin, but not needed for ParallelHash); and some stuff is
only going to be needed for parallel sequential scan specifically.
This patch mixes all of those concerns together in a single function.
That won't do; this needs to be easily extensible to whatever someone
wants to parallelize next.

3. I think the whole idea of using the portal infrastructure for this
is wrong.  We've talked about this before, but the fact that you're
doing it this way is having a major impact on the whole design of the
patch, and I can't believe it's ever going to be committable this way.
To create a portal, you have to pretend that you received a protocol
message, which you didn't; and you have to pretend there is an SQL
query so you can call PortalDefineQuery.   That's ugly.  As far as I
can see the only thing we really get out of any of that is that we can
use the DestReceiver stuff to get the tuples back to the master, but
that doesn't really work either, because you're having to hack
printtup.c anyway.  So from my point of view you're going through a
bunch of layers that really don't have any value.  Considering the way
the parallel mode patch has evolved, I no longer think there's much
point to passing anything other than raw tuples between the backends,
so the whole idea of going through a deform/send/recv/form cycle seems
like something we can entirely skip.

4.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

[1] It is of course arguable whether a parallel index-scan or parallel
bitmap index-scan or parallel index-only-scan or parallel custom scan
makes sense, but this patch shouldn't assume that we won't want to do
those things.  We have other places in the code that know about the
concept of a scan as opposed to some other kind of executor construct,
and we should preserve that distinction here.

On Fri, Feb 6, 2015 at 12:34 PM, Robert Haas <robertmhaas@gmail.com> wrote:
> 4.

Obviously that went out a bit too soon.  Anyway, what I think we
should do here is back up a bit and talk about what the problems are
that we need to solve here and how each of them should be solved.  I
think there is some good code in this patch, but we really need to
think about what the interfaces should look like and achieve a clean
separation of concerns.

Looking at the code for the non-parallel SeqScan node, there are
basically two things going on here:

1. We call heap_getnext() to get the next tuple and store it into a
TupleTableSlot.
2. Via ExecScan(), we do projection and apply quals.

My first comment here is that I think we should actually teach
heapam.c about parallelism.  In other words, let's have an interface
like this:

extern Size heap_parallelscan_estimate(Snapshot snapshot);
extern void heap_parallelscan_initialize(ParallelHeapScanDesc target,
Relation relation, Snapshot snapshot);
extern HeapScanDesc heap_beginscan_parallel(ParallelHeapScanDesc);

So the idea is that if you want to do a parallel scan, you call
heap_parallelscan_estimate() to figure out how much space to reserve
in your dynamic shared memory segment.  Then you call
heap_parallelscan_initialize() to initialize the chunk of memory once
you have it.  Then each backend that wants to assist in the parallel
scan calls heap_beginscan_parallel() on that chunk of memory and gets
its own HeapScanDesc.  Then, they can all call heap_getnext() just as
in the non-parallel case.  The ParallelHeapScanDesc needs to contain
the relation OID, the snapshot, the ending block number, and a
current-block counter.  Instead of automatically advancing to the next
block, they use one of Andres's nifty new atomic ops to bump the
current-block counter and then scan the block just before the new
value.  All this seems pretty straightforward, and if we decide to
later change the way the relation gets scanned (e.g. in 1GB chunks
rather than block-by-block) it can be handled here pretty easily.

Now, let's suppose that we have this interface and for some reason we
don't care about quals and projection - we just want to get the tuples
back to the master.  It's easy enough to create a parallel context
that fires up a worker and lets the worker call
heap_beginscan_parallel() and then heap_getnext() in a loop, but what
does it do with the resulting tuples?  We need a tuple queue that can
be used to send the tuples back to master.  That's also pretty easy:
just set up a shm_mq and use shm_mq_send() to send each tuple.  Use
shm_mq_receive() in the master to read them back out.  The only thing
we need to be careful about is that the tuple descriptors match.  It
must be that they do, because the way the current parallel context
patch works, the master is guaranteed to hold a lock on the relation
from before the worker starts up until after it dies.  But we could
stash the tuple descriptor in shared memory and cross-check that it
matches just to be sure.  Anyway, this doesn't seem terribly complex
although we might want to wrap some abstraction around it somehow so
that every kind of parallelism that uses tuple queues can benefit from
it.  Perhaps this could even be built into the parallel context
machinery somehow, or maybe it's something executor-specific.  At any
rate it looks simpler than what you've got now.

The complicated part here seems to me to figure out what we need to
pass from the parallel leader to the parallel worker to create enough
state for quals and projection.  If we want to be able to call
ExecScan() without modification, which seems like a good goal, we're
going to need a ScanState node, which is going to need to contain
valid pointers to (at least) a ProjectionInfo, an ExprContext, and a
List of quals.  That in turn is going to require an ExecutorState.
Serializing those things directly doesn't seem very practical; what we
instead want to do is figure out what we can pass that will allow easy
reconstruction of those data structures.  Right now, you're passing
the target list, the qual list, the range table, and the params, but
the range table doesn't seem to be getting used anywhere.  I wonder if
we need it.  If we could get away with just passing the target list
and qual list, and params, we'd be doing pretty well, I think.  But
I'm not sure exactly what that looks like.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Fri, Feb 6, 2015 at 2:13 PM, Robert Haas <robertmhaas@gmail.com> wrote:
> The complicated part here seems to me to figure out what we need to
> pass from the parallel leader to the parallel worker to create enough
> state for quals and projection.  If we want to be able to call
> ExecScan() without modification, which seems like a good goal, we're
> going to need a ScanState node, which is going to need to contain
> valid pointers to (at least) a ProjectionInfo, an ExprContext, and a
> List of quals.  That in turn is going to require an ExecutorState.
> Serializing those things directly doesn't seem very practical; what we
> instead want to do is figure out what we can pass that will allow easy
> reconstruction of those data structures.  Right now, you're passing
> the target list, the qual list, the range table, and the params, but
> the range table doesn't seem to be getting used anywhere.  I wonder if
> we need it.  If we could get away with just passing the target list
> and qual list, and params, we'd be doing pretty well, I think.  But
> I'm not sure exactly what that looks like.

IndexBuildHeapRangeScan shows how to do qual evaluation with
relatively little setup:
   estate = CreateExecutorState();   econtext = GetPerTupleExprContext(estate);   slot =
MakeSingleTupleTableSlot(RelationGetDescr(heapRelation));
   /* Arrange for econtext's scan tuple to be the tuple under test */   econtext->ecxt_scantuple = slot;
   /* Set up execution state for predicate, if any. */   predicate = (List *)       ExecPrepareExpr((Expr *)
indexInfo->ii_Predicate,                      estate);
 

Then, for each tuple:
      ExecStoreTuple(heapTuple, slot, InvalidBuffer, false);

And:
           if (!ExecQual(predicate, econtext, false))               continue;

This looks like a good model to follow for parallel sequential scan.
The point though is that I think we should do it directly rather than
letting the portal machinery do it for us.  Not sure how to get
projection working yet.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On Fri, Feb 6, 2015 at 2:13 PM, Robert Haas <robertmhaas@gmail.com> wrote:
> My first comment here is that I think we should actually teach
> heapam.c about parallelism.

I coded this up; see attached.  I'm also attaching an updated version
of the parallel count code revised to use this API.  It's now called
"parallel_count" rather than "parallel_dummy" and I removed some
stupid stuff from it.  I'm curious to see what other people think, but
this seems much cleaner to me.  With the old approach, the
parallel-count code was duplicating some of the guts of heapam.c and
dropping the rest on the floor; now it just asks for a parallel scan
and away it goes.  Similarly, if your parallel-seqscan patch wanted to
scan block-by-block rather than splitting the relation into equal
parts, or if it wanted to participate in the synchronized-seqcan
stuff, there was no clean way to do that.  With this approach, those
decisions are - as they quite properly should be - isolated within
heapam.c, rather than creeping into the executor.

(These patches should be applied over parallel-mode-v4.patch.)

--
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

On 2015-02-06 22:57:43 -0500, Robert Haas wrote:
> On Fri, Feb 6, 2015 at 2:13 PM, Robert Haas <robertmhaas@gmail.com> wrote:
> > My first comment here is that I think we should actually teach
> > heapam.c about parallelism.
> 
> I coded this up; see attached.  I'm also attaching an updated version
> of the parallel count code revised to use this API.  It's now called
> "parallel_count" rather than "parallel_dummy" and I removed some
> stupid stuff from it.  I'm curious to see what other people think, but
> this seems much cleaner to me.  With the old approach, the
> parallel-count code was duplicating some of the guts of heapam.c and
> dropping the rest on the floor; now it just asks for a parallel scan
> and away it goes.  Similarly, if your parallel-seqscan patch wanted to
> scan block-by-block rather than splitting the relation into equal
> parts, or if it wanted to participate in the synchronized-seqcan
> stuff, there was no clean way to do that.  With this approach, those
> decisions are - as they quite properly should be - isolated within
> heapam.c, rather than creeping into the executor.

I'm not convinced that that reasoning is generally valid. While it may
work out nicely for seqscans - which might be useful enough on its own -
the more stuff we parallelize the *more* the executor will have to know
about it to make it sane. To actually scale nicely e.g. a parallel sort
will have to execute the nodes below it on each backend, instead of
doing that in one as a separate step, ferrying over all tuples to
indivdual backends through queues, and only then parallezing the
sort.

Now. None of that is likely to matter immediately, but I think starting
to build the infrastructure at the points where we'll later need it does
make some sense.

Greetings,

Andres Freund

-- Andres Freund                       http://www.2ndQuadrant.com/PostgreSQL Development, 24x7 Support, Training &
Services

On Sat, Feb 7, 2015 at 4:30 PM, Andres Freund <andres@2ndquadrant.com> wrote:
> On 2015-02-06 22:57:43 -0500, Robert Haas wrote:
>> On Fri, Feb 6, 2015 at 2:13 PM, Robert Haas <robertmhaas@gmail.com> wrote:
>> > My first comment here is that I think we should actually teach
>> > heapam.c about parallelism.
>>
>> I coded this up; see attached.  I'm also attaching an updated version
>> of the parallel count code revised to use this API.  It's now called
>> "parallel_count" rather than "parallel_dummy" and I removed some
>> stupid stuff from it.  I'm curious to see what other people think, but
>> this seems much cleaner to me.  With the old approach, the
>> parallel-count code was duplicating some of the guts of heapam.c and
>> dropping the rest on the floor; now it just asks for a parallel scan
>> and away it goes.  Similarly, if your parallel-seqscan patch wanted to
>> scan block-by-block rather than splitting the relation into equal
>> parts, or if it wanted to participate in the synchronized-seqcan
>> stuff, there was no clean way to do that.  With this approach, those
>> decisions are - as they quite properly should be - isolated within
>> heapam.c, rather than creeping into the executor.
>
> I'm not convinced that that reasoning is generally valid. While it may
> work out nicely for seqscans - which might be useful enough on its own -
> the more stuff we parallelize the *more* the executor will have to know
> about it to make it sane. To actually scale nicely e.g. a parallel sort
> will have to execute the nodes below it on each backend, instead of
> doing that in one as a separate step, ferrying over all tuples to
> indivdual backends through queues, and only then parallezing the
> sort.
>
> Now. None of that is likely to matter immediately, but I think starting
> to build the infrastructure at the points where we'll later need it does
> make some sense.

Well, I agree with you, but I'm not really sure what that has to do
with the issue at hand.  I mean, if we were to apply Amit's patch,
we'd been in a situation where, for a non-parallel heap scan, heapam.c
decides the order in which blocks get scanned, but for a parallel heap
scan, nodeParallelSeqscan.c makes that decision.  Maybe I'm an old
fuddy-duddy[1] but that seems like an abstraction violation to me.  I
think the executor should see a parallel scan as a stream of tuples
that streams into a bunch of backends in parallel, without really
knowing how heapam.c is dividing up the work.  That's how it's
modularized today, and I don't see a reason to change it.  Do you?

Regarding tuple flow between backends, I've thought about that before,
I agree that we need it, and I don't think I know how to do it.  I can
see how to have a group of processes executing a single node in
parallel, or a single process executing a group of nodes we break off
from the query tree and push down to it, but what you're talking about
here is a group of processes executing a group of nodes jointly.  That
seems like an excellent idea, but I don't know how to design it.
Actually routing the tuples between whichever backends we want to
exchange them between is easy enough, but how do we decide whether to
generate such a plan?  What does the actual plan tree look like?
Maybe we designate nodes as can-generate-multiple-tuple-streams (seq
scan, mostly, I would think) and can-absorb-parallel-tuple-streams
(sort, hash, materialize), or something like that, but I'm really
fuzzy on the details.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

[1] Actually, there's not really any "maybe" about this.

On Sat, Feb 7, 2015 at 10:36 PM, Amit Kapila <amit.kapila16@gmail.com> wrote:
>> Well, I agree with you, but I'm not really sure what that has to do
>> with the issue at hand.  I mean, if we were to apply Amit's patch,
>> we'd been in a situation where, for a non-parallel heap scan, heapam.c
>> decides the order in which blocks get scanned, but for a parallel heap
>> scan, nodeParallelSeqscan.c makes that decision.
>
> I think other places also decides about the order/way heapam.c has
> to scan, example the order in which rows/pages has to traversed is
> decided at portal/executor layer and the same is passed till heap and
> in case of index, the scanlimits (heap_setscanlimits()) are decided
> outside heapam.c and something similar is done for parallel seq scan.
> In general, the scan is driven by Scandescriptor which is constructed
> at upper level and there are some API's exposed to derive the scan.
> If you are not happy with the current way nodeParallelSeqscan has
> set the scan limits, we can have some form of callback which do the
> required work and this callback can be called from heapam.c.

I thought about a callback, but what's the benefit of doing that vs.
hard-coding it in heapam.c?  If the upper-layer wants to impose a TID
qual or similar then heap_setscanlimits() makes sense, but that's
effectively a filter condition, not a policy decision about the access
pattern.

-- 
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company