Re: mapjoin with left join

[Gopal Vijayaraghavan]

> select small.* from small s left join large l on s.id  =
>l.id  where l.id  is null;
...
> We simply want to load the 81K rows in to RAM, then for each row in
>large, check the small hash table and if it the row in small is not in
>large, then add it to large.

That seems like a fair description of the problem with the approach, but
it is fairly hard to do left outers in a distributed fashion with
correctness.

Imagine I load partition #1 from the big table + the small table into a
task - how would I know partiton #2 doesn't have a key from the small
hashtable. So the system cannot output any rows in the naive version of
this problem until it compares the small table to all parts of the big
table.

That I think was Sergey's point & I don't know of an easy way around that
yet.

> The business case is loading only new rows into a large fact table.  The
>new rows are the ones that are small in number.

That however is much easier - because that's better written as.

insert into large select * from small where id NOT IN (select id from
large);

FYI, we want people to stop writing these sort of queries once the MERGE
syntax lands in HIVE - https://issues.apache.org/jira/browse/HIVE-10924


The trick is to rewrite the NOT IN condition as two set operations - inner
join + remainder left outer.

Find all ids which already exist in bigtable with an inner map-join.
Remove all those ids from the small table & insert the remainder.

explain rewrite select remainder.* from small remainder where id not in
(select id from small, large where small.id = large.id);

That query, you will find will run much faster than the query you're
currently using. 

If you can, please send the "explain rewrite" & I can probably fine-tune
this approach further.

If you're using Tez, you might want to try out the custom vertex managers
as well

set hive.vectorized.execution.mapjoin.minmax.enabled=true;

set hive.vectorized.execution.mapjoin.native.fast.hashtable.enabled=true;
set hive.optimize.dynamic.partition.hashjoin=true;

Those params should allow you to scale up a map-join to ~100x the
available RAM (aimed at LLAP memory utilization).
 
Cheers,
Gopal

Re: mapjoin with left join

[Steve Howard]

Hi Gopal/All,

Yep, I absolutely understand the limitation of what we are trying to do.
We will try the settings you suggested.

Thanks,

Steve

On Tue, Sep 22, 2015 at 1:44 PM, Gopal Vijayaraghavan 
wrote:

>
> > select small.* from small s left join large l on s.id  =
> >l.id  where l.id  is null;
> ...
> > We simply want to load the 81K rows in to RAM, then for each row in
> >large, check the small hash table and if it the row in small is not in
> >large, then add it to large.
>
> That seems like a fair description of the problem with the approach, but
> it is fairly hard to do left outers in a distributed fashion with
> correctness.
>
> Imagine I load partition #1 from the big table + the small table into a
> task - how would I know partiton #2 doesn't have a key from the small
> hashtable. So the system cannot output any rows in the naive version of
> this problem until it compares the small table to all parts of the big
> table.
>
> That I think was Sergey's point & I don't know of an easy way around that
> yet.
>
> > The business case is loading only new rows into a large fact table.  The
> >new rows are the ones that are small in number.
>
> That however is much easier - because that's better written as.
>
> insert into large select * from small where id NOT IN (select id from
> large);
>
> FYI, we want people to stop writing these sort of queries once the MERGE
> syntax lands in HIVE - https://issues.apache.org/jira/browse/HIVE-10924
>
>
> The trick is to rewrite the NOT IN condition as two set operations - inner
> join + remainder left outer.
>
> Find all ids which already exist in bigtable with an inner map-join.
> Remove all those ids from the small table & insert the remainder.
>
> explain rewrite select remainder.* from small remainder where id not in
> (select id from small, large where small.id = large.id);
>
> That query, you will find will run much faster than the query you're
> currently using.
>
> If you can, please send the "explain rewrite" & I can probably fine-tune
> this approach further.
>
> If you're using Tez, you might want to try out the custom vertex managers
> as well
>
> set hive.vectorized.execution.mapjoin.minmax.enabled=true;
>
> set hive.vectorized.execution.mapjoin.native.fast.hashtable.enabled=true;
> set hive.optimize.dynamic.partition.hashjoin=true;
>
> Those params should allow you to scale up a map-join to ~100x the
> available RAM (aimed at LLAP memory utilization).
>
> Cheers,
> Gopal
>
>
>

Re: mapjoin with left join

[Noam Hasson]

Not sure if will help you, but you can try to use the map-join hint,
basically hinting Hive to put a specific table in memory:

https://cwiki.apache.org/confluence/display/Hive/LanguageManual+JoinOptimization#LanguageManualJoinOptimization-PriorSupportforMAPJOIN

On Fri, Sep 11, 2015 at 11:16 PM, Sergey Shelukhin 
wrote:

> As far as I know it’s not currently supported.
> The large table will be streamed in multiple tasks with the small table in
> memory, so there’s not one place that knows for sure there was no row in
> the large table for a particular small table row in any of the locations.
> It could have no match in one task but a match in other task.
> You can try rewriting the query as inner join unioned with not in, but
> “not in” might still be slow…
> IIRC there was actually a JIRA to solve this, but no work has been done so
> far.
>
> From: Steve Howard 
> Reply-To: "user@hive.apache.org" 
> Date: Friday, September 11, 2015 at 09:48
> To: "user@hive.apache.org" 
> Subject: mapjoin with left join
>
> We would like to utilize mapjoin for the following SQL construct:
>
> select small.* from small s left join large l on s.id = l.id where l.id is
> null;
>
> We can easily fit small into RAM, but large is over 1TB according to
> optimizer stats. Unless we set
> hive.auto.convert.join.noconditionaltask.size = to at least the size of
> "large", the optimizer falls back to a common map join, which is incredibly
> slow.
>
> Given the fact it is a left join, which means we won't always have rows in
> large for each row in small, is this behavior expected? Could it be that
> reading the large table would miss the new rows in small, so the large one
> has to be the one that is probed for matches?
>
> We simply want to load the 81K rows in to RAM, then for each row in large,
> check the small hash table and if it the row in small is not in large, then
> add it to large.
>
> Again, the optimizer will use a mapjoin if we set
> hive.auto.convert.join.noconditionaltask.size = 1TB (the size of the large
> table). This is of course, not practical. The small table is only 50MB.
>
> At the link below is the entire test case with two tables, one of which
> has three rows and other has 96. We can duplicate it with tables this
> small, which leads me to believe I am missing something, or this is a bug.
>
> The link has the source code that shows each table create, as well as the
> explain with an argument for hive.auto.convert.join.noconditionaltask.size
> that is passed at the command line. The output shows a mergejoin when the
> hive.auto.convert.join.noconditionaltask.size size is less than 192 (the
> size of the larger table), and a mapjoin when
> hive.auto.convert.join.noconditionaltask.size is larger than 192 (large
> table fits).
>
> http://pastebin.com/Qg6hb8yV
>
> The business case is loading only new rows into a large fact table.  The
> new rows are the ones that are small in number.
>

-- 
This e-mail, as well as any attached document, may contain material which 
is confidential and privileged and may include trademark, copyright and 
other intellectual property rights that are proprietary to Kenshoo Ltd, 
 its subsidiaries or affiliates ("Kenshoo"). This e-mail and its 
attachments may be read, copied and used only by the addressee for the 
purpose(s) for which it was disclosed herein. If you have received it in 
error, please destroy the message and any attachment, and contact us 
immediately. If you are not the intended recipient, be aware that any 
review, reliance, disclosure, copying, distribution or use of the contents 
of this message without Kenshoo's express permission is strictly prohibited.

Re: mapjoin with left join

[Sergey Shelukhin]

As far as I know it’s not currently supported.
The large table will be streamed in multiple tasks with the small table in 
memory, so there’s not one place that knows for sure there was no row in the 
large table for a particular small table row in any of the locations. It could 
have no match in one task but a match in other task.
You can try rewriting the query as inner join unioned with not in, but “not in” 
might still be slow…
IIRC there was actually a JIRA to solve this, but no work has been done so far.

From: Steve Howard >
Reply-To: "user@hive.apache.org" 
>
Date: Friday, September 11, 2015 at 09:48
To: "user@hive.apache.org" 
>
Subject: mapjoin with left join

We would like to utilize mapjoin for the following SQL construct:

select small.* from small s left join large l on s.id = 
l.id where l.id is null;

We can easily fit small into RAM, but large is over 1TB according to optimizer 
stats. Unless we set hive.auto.convert.join.noconditionaltask.size = to at 
least the size of "large", the optimizer falls back to a common map join, which 
is incredibly slow.

Given the fact it is a left join, which means we won't always have rows in 
large for each row in small, is this behavior expected? Could it be that 
reading the large table would miss the new rows in small, so the large one has 
to be the one that is probed for matches?

We simply want to load the 81K rows in to RAM, then for each row in large, 
check the small hash table and if it the row in small is not in large, then add 
it to large.

Again, the optimizer will use a mapjoin if we set 
hive.auto.convert.join.noconditionaltask.size = 1TB (the size of the large 
table). This is of course, not practical. The small table is only 50MB.

At the link below is the entire test case with two tables, one of which has 
three rows and other has 96. We can duplicate it with tables this small, which 
leads me to believe I am missing something, or this is a bug.

The link has the source code that shows each table create, as well as the 
explain with an argument for hive.auto.convert.join.noconditionaltask.size that 
is passed at the command line. The output shows a mergejoin when the 
hive.auto.convert.join.noconditionaltask.size size is less than 192 (the size 
of the larger table), and a mapjoin when 
hive.auto.convert.join.noconditionaltask.size is larger than 192 (large table 
fits).

http://pastebin.com/Qg6hb8yV

The business case is loading only new rows into a large fact table.  The new 
rows are the ones that are small in number.